Hadoop的前世今生（2）——ipc的Client(1)

向右看齐 2022-06-12 06:08 223阅读 0赞

今天的任务是解决Client类，它在源码的ipc包下，作为分布式C-S结构的C。本文的前提是你要了解C的功能。本文使用的源码是hadoop-0.10.1。start!

在它的里面包括5个子类，分别是：

Call:

Connection:

ParallelCall:

ParallelResult:

ConnectionCuller:

下面我们依次分析这五个类的内容：

**Call:**

这个类是一次调用过程的一个封装，它的成员变量如下所示：

int id; // call id
    Writable param; // parameter
    Writable value; // value, null if error
    RemoteException error; // error, null if value
    long lastActivity; // time of last i/o
    boolean done; // true when call is done

可以看出，这里面包括了调用的参数，返回的结果以及一些相关的属性。在他的方法中，值得一提的就是callComplete方法。这个方法调用了自身的notify，从而使因它阻塞的类被唤醒而继续进行后面的工作（在此只是大概地形容，具体的关于wait和notify的内容请参见第一篇）。

那么这个类的功能和生命周期是什么样的呢？接下来我们直接来到Client类的call方法，具体的代码如下：

/**
    	 * Make a call, passing <code>param</code>, to the IPC server running at
    	 * <code>address</code>, returning the value. Throws exceptions if there are
    	 * network problems or if the remote code threw an exception.
    	 */
    	public Writable call(Writable param, InetSocketAddress address)
    			throws IOException {
    		Connection connection = getConnection(address);
    		Call call = new Call(param);
    		synchronized (call) {
    			connection.sendParam(call); // send the parameter
    			long wait = timeout;
    			do {
    				try {
    					call.wait(wait); // wait for the result
    				} catch (InterruptedException e) {
    				}
    				wait = timeout
    						- (System.currentTimeMillis() - call.lastActivity);
    			} while (!call.done && wait > 0);
    
    			if (call.error != null) {
    				throw call.error;
    			} else if (!call.done) {
    				throw new SocketTimeoutException(
    						"timed out waiting for rpc response");
    			} else {
    				return call.value;
    			}
    		}
    	}

我们看到，这个方法要做的工作是发起一个调用并获得相应的返回（无论正确与否）。当这个方法调用后，一个Call的对象建立，然后将Call通过Connnection发出（Connection的问题后面再说），然后这段代码就进入了等待的状态（call.wait(duration)），直到时间到达或者被notify。这样我们就大概明白了整个Call的过程：

（1）Client创建Call并填充相应的参数；

（2）Client把Call交给Connection进行监控处理，而自己进入wait状态

（3）当满足了等待的结束条件，Client将相应的后续任务处理并完成这次调用。

（思考：call是本地线程创建的一个变量，为什么要做同步？）

那么我们就知道了这个流程，实际上Client与Server的交互也就是这个流程，那么我们现在缺少了一个关键的流程，就是Connection，下面我们来看Connection。

**Connection**

首先是它的成员变量：

private InetSocketAddress address; // address of server
    private Socket socket = null; // connected socket
    private DataInputStream in;
    private DataOutputStream out;
    private Hashtable calls = new Hashtable(); // currently active calls
    private Call readingCall;
    private Call writingCall;
    private int inUse = 0;
    private long lastActivity = 0;
    private boolean shouldCloseConnection = false;

前面四个变量都是和网络连接相关的变量，而第五个变量在原代码中没有做泛型，所以一下子猜不出来，但是大概可以知道它是做连接管理的。剩下的一些我们在后面一起分析。我们要注意的是这个类继承自Thread.

接下来我们来分析它的生命周期，同时把剩下变量的含义弄明白。

（1）它的起源来自刚才我们看过的call函数的第一句：

Connection connection = getConnection(address);

于是我们继续跟进它，来看这个函数：

/**
    	 * Get a connection from the pool, or create a new one and add it to the
    	 * pool. Connections to a given host/port are reused.
    	 */
    	private Connection getConnection(InetSocketAddress address)
    			throws IOException {
    		Connection connection;
    		synchronized (connections) {
    			connection = (Connection) connections.get(address);
    			if (connection == null) {
    				connection = new Connection(address);
    				connections.put(address, connection);
    				connection.start();
    			}
    			connection.incrementRef();
    		}
    		// we don't invoke the method below inside "synchronized (connections)"
    		// block above. The reason for that is if the server happens to be slow,
    		// it will take longer to establish a connection and that will slow the
    		// entire system down.
    		connection.setupIOstreams();
    		return connection;
    	}

看过了这个函数，我们明白了这些事情：

1.Client类中的connections变量（Hashtable）是为了记录每一个建立的connection的。

2.connection的管理是采用标记引用的方法的，也就是说每有一个针对它的使用它都要incrementRef一次（跟进此函数我们发现了InUse变量的作用：就是记录同时被引用的数量）。获得了connection的变量，我们开始setupIOstreams并返回完成连接的创建。下面我们进入Connection.setupIOstreams函数，由于这个函数比较长，我们一点点分析：

public synchronized void setupIOstreams() throws IOException {
    	if (socket != null) {
    		notify();
    		return;
    	}

这个代码给了我们一种感觉，就是说如果连接已经建立，那么那个使用这个类的对象会处于wait状态，现在“活来了”，所以它需要被唤醒了。go on

short failures = 0;
    			while (true) {
    				try {
    					this.socket = new Socket();
    					this.socket.connect(address, FSConstants.READ_TIMEOUT);
    					break;
    				} catch (IOException ie) { // SocketTimeoutException is also
    											// caught
    					if (failures == maxRetries) {
    						// reset inUse so that the culler gets a chance to throw
    						// this
    						// connection object out of the table. We don't want to
    						// increment
    						// inUse to infinity (everytime getConnection is called
    						// inUse is
    						// incremented)!
    						inUse = 0;
    						throw ie;
    					}
    					failures++;
    					LOG.info("Retrying connect to server: " + address
    							+ ". Already tried " + failures + " time(s).");
    					try {
    						Thread.sleep(1000);
    					} catch (InterruptedException iex) {
    					}
    				}
    			}
    			socket.setSoTimeout(timeout);

接下来这个代码是做了socket连接的工作，这里面很有技巧，它可以支持多次连接。

同时为了性能优化，它还设置了socket的timeout。这里给出setSoTimeout()方法的doc：

##### void java.net. Socket.setSoTimeout(int timeout) throws SocketException #####

Enable/disable SO\_TIMEOUT with the specified timeout, in milliseconds. With this option set to a non-zero timeout, a read() call on the InputStream associated with this Socket will block for only this amount of time. If the timeout expires, a **java.net.SocketTimeoutException** is raised, though the Socket is still valid. The option **must** be enabled prior to entering the blocking operation to have effect. The timeout must be > 0. A timeout of zero is interpreted as an infinite timeout.

\[译\]启动或关闭SO\_TIMEOUT并给定指定的超时时间，单位为毫秒。如果参数是一个非零的超时时间，一个针对关联这个Socket的InputStream的read()方法将在指定的时间后关闭。如果超时，一个异常将抛出，尽管Socket仍然合法。此项设置必须在对Socket的操作前设置方能有效，同时timeout必须大于0，如果等于0意味着无限的等待。 Parameters:

**timeout** the specified timeout, in milliseconds.

Throws:

SocketException \- if there is an error in the underlying protocol, such as a TCP error.

Since:

JDK 1.1

See Also:

getSoTimeout()

\[Doc完\]  
这样的设置显然是很有好处的。如果网络状况不佳时，我们就可以以异常的方式结束一个很费时甚至不可达的读请求。下面继续~

this.in = new DataInputStream(new BufferedInputStream(
    					new FilterInputStream(socket.getInputStream()) {
    						public int read(byte[] buf, int off, int len)
    								throws IOException {
    							int value = super.read(buf, off, len);
    							if (readingCall != null) {
    								readingCall.touch();
    							}
    							return value;
    						}
    					}));
    			this.out = new DataOutputStream(new BufferedOutputStream(
    					new FilterOutputStream(socket.getOutputStream()) {
    						public void write(byte[] buf, int o, int len)
    								throws IOException {
    							out.write(buf, o, len);
    							if (writingCall != null) {
    								writingCall.touch();
    							}
    						}
    					}));

下面是创建了IO的stream，注意到他们与readingCall和WritingCall还有一点联系，这个我相信会在后面有更加详细的解释。总体来说这段代码展示了FilterStream的用法，这个话题说起来有点大，在此打住了。go on~

if (SEND_HEADER) {
    				out.write(Server.HEADER.array());
    				out.write(CURRENT_VERSION);
    			}
    			notify();

这段代码同样包含唤醒，当然前面有一个和Server端的互动，这个不是重点了。这样我们就搞定了setup这个函数，在此总结一下，这个函数用于完成连接的基本建立，当然其中的很多考量是很值得我们学习的，比如考虑网络的不确定性，IO的插入式写法……

那么connection的建立就结束了？还没有。我们得再研究下作者的一段注释，前面已经给出了，这里再次给出：

// we don't invoke the method below inside "synchronized (connections)"
    		// block above. The reason for that is if the server happens to be slow,
    		// it will take longer to establish a connection and that will slow the
    		// entire system down.

这里不做翻译了，意思就是说由于connection的建立时间比较长，我们应该把它和创建connection对象的过程分离开。这个想法对于提高系统室友帮助的，而且它不会导致同步问题（ 为什么？）

可能思路有点混乱了，这里把上面的要点列一下：

Client中call调用了getConnection函数->getConnection细节->Connection.setupIO细节

下面我们要回到call函数：

connection.sendParam(call); // send the parameter

我们来看这个函数：

/**
    		 * Initiates a call by sending the parameter to the remote server. Note:
    		 * this is not called from the Connection thread, but by other threads.
    		 */
    		public void sendParam(Call call) throws IOException {
    			boolean error = true;
    			try {
    				calls.put(new Integer(call.id), call);
    				synchronized (out) {
    					if (LOG.isDebugEnabled())
    						LOG.debug(getName() + " sending #" + call.id);
    					try {
    						writingCall = call;
    						DataOutputBuffer d = new DataOutputBuffer(); // for
    																		// serializing
    																		// the
    																		// data
    																		// to be
    																		// written
    						d.writeInt(call.id);
    						call.param.write(d);
    						
    						byte[] data = d.getData();
    						int dataLength = d.getLength();
    						out.writeInt(dataLength); // first put the data length
    						out.write(data, 0, dataLength);// write the data
    						out.flush();
    					} finally {
    						writingCall = null;
    					}
    				}
    				error = false;
    			} finally {
    				if (error) {
    					synchronized (connections) {
    						if (connections.get(address) == this)
    							connections.remove(address);
    					}
    					close(); // close on error
    				}
    			}
    		}

这个函数做的事情够多的。但是因为它形成了一个整体，所以需要一起说：

（1）calls的作用浮出水面，它用来记录所有来往call的记录，这个函数没有call离开记录的代码，我们为了完整性在这里将它找到，它的删除代码出现在run()方法里（还记得它继承自thread了么？）

（2）由于一个连接要负责N个call的调用，所以对于IO我们都要保证彼此间的隔离性，所以我们将out上了同步锁，这样每一个call之间传给Server的数据不会混乱。  
（3）下面就是整个传参数的过程了。由于Hadoop这一套系统喜欢用DataStream，所以我们需要将自己的格式转换成它的标准格式，然后在发给Server。具体的方法不说了。

（4）这段代码中出现了一个writingcall的使用……很像打酱油是不是？我也是这么觉得的，所以把它留下，我们一会儿回头看。

所以总结一下，这段代码将每一个被调用的call都管理起来，并将参数发给Server......所以还差一个把返回值接受的步骤。看来看去我们只剩下run函数没有看了，所以go on~因为这个函数里面有一些LOG，我们把LOG去掉，剩下的代码如下所示：

这段代码也很复杂，我们同样拆开，一步一步看：

if (waitForWork() == false)
    						break;

这里涉及到一个waitForWork函数，那么展开它一看：

private synchronized boolean waitForWork() {
    			// wait till someone signals us to start reading RPC response or
    			// close the connection. If we are idle long enough (blocked in
    			// wait),
    			// the ConnectionCuller thread will wake us up and ask us to close
    			// the
    			// connection.
    			// We need to wait when inUse is 0 or socket is null (it may be null
    			// if
    			// the Connection object has been created but the socket connection
    			// has not been setup yet). We stop waiting if we have been asked to
    			// close
    			// connection
    			while ((inUse == 0 || socket == null) && !shouldCloseConnection) {
    				try {
    					wait();
    				} catch (InterruptedException e) {
    				}
    			}
    			return !shouldCloseConnection;
    		}

这个函数的注释已经写得很明白了（我觉得这段注释应该写在函数端），如果没活干（还记得inUse的作用么？用在记录引用方法中的），那么就wait（这回明白前面setup的时候为什么要notify了吧）当然，里面还提到了一个叫做ConnectionCuller的东东，它也是五个类之一。我们暂且按住好奇心，回到run函数：

try {
    						id = in.readInt(); // try to read an id
    					} catch (SocketTimeoutException e) {
    						continue;
    					}
    					Call call = (Call) calls.remove(new Integer(id));
    					boolean isError = in.readBoolean(); // read if error

这里应该是属于协议的内容，Server返回的前两个是Caller的id和正确错误情况。go on~

if (isError) {
    						RemoteException ex = new RemoteException(
    								WritableUtils.readString(in),
    								WritableUtils.readString(in));
    						call.setResult(null, ex);
    					} else {
    						Writable value = (Writable) ReflectionUtils
    								.newInstance(valueClass, conf);
    						try {
    							readingCall = call;
    							value.readFields(in); // read value
    						} finally {
    							readingCall = null;
    						}
    						call.setResult(value, null);
    					}

这个就是根据结果的分别操作了。至于里面涉及到的反射不是我们的重点，我们以后再说。go on~

call.callComplete(); // deliver result to caller
    					// received the response. So decrement the ref count
    					decrementRef();

这里调用了callComplete（还记得它的作用么？ 这个方法调用了自身的notify，从而使因它阻塞的类被唤醒而继续进行后面的工作），这就是说，从这里，我们的Client的call函数理论上可以继续进行了。于是call函数就走通了。剩下的就是去掉引用的代码。后面还有一些异常和finally的回收代码就不分析了，内容很简单。

到这里，Connection的代码基本上搞定，总结一下：

（1）连接管理关系：Client->Connections->Connection->calls->call

（2）setIO方法负责建立连接（或者增加引用）

（3）setParam负责发送参数到Server

（4）Thread自身负责接收返回。

（5）考虑到了网络的不稳定的特点

结束了么？还没有……上面还有一个问题，就是writingcall，即使readingcall的问题也是一样的，我们刚才遇到了也没有提。它们究竟是什么用？我们还得回到前面的setIO那个函数的代码：

虽然已经粘过一遍了，但是还是再粘贴一遍。所以它们附上的意义就在于那个touch（）。那么我们还要接着问，它是如何保证线程安全的呢？（这个问题在多线程框架中一定要问）

首先是writingcall，它的赋值与解赋值都在out的同步块内。也就是说只有向Server写数据才用它，所以它的线程安全被out保证了。

然后是readingcall，它则不需要线程安全保证，因为Thread线程每一次只取一个call的返回值，相当于串行，所以它的线程安全也保证了。

到此我们解决了五个子类中的两个，实际上是因为这两个类和Client的功能紧密相关，下一次我们将分析其他几个类的功能和Client所做的更多的功能。