HADOOP-1841. Prevent slow clients from consuming threads in the NameNode.

(dhruba)



git-svn-id: https://svn.apache.org/repos/asf/lucene/hadoop/trunk@603795 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -238,6 +238,9 @@ Trunk (unreleased changes)
     HADOOP-2359.  Remove warning for interruptted exception when closing down
     minidfs. (dhruba via omalley)
 
+    HADOOP-1841. Prevent slow clients from consuming threads in the NameNode. 
+    (dhruba)
+
 Branch 0.15 (unreleased)
 
   BUG FIXES

src/java/org/apache/hadoop/ipc/Client.java

@@ -73,6 +73,7 @@ public class Client {
   private int maxRetries; //the max. no. of retries for socket connections
   private Thread connectionCullerThread;
   private SocketFactory socketFactory;           // how to create sockets
+  private boolean simulateError = false;         // unit tests
 
   /** A call waiting for a value. */
   private class Call {
@@ -286,6 +287,7 @@ public class Client {
           } else {
             Writable value = (Writable)ReflectionUtils.newInstance(valueClass, conf);
             try {
+              waitForEndSimulation();
               readingCall = call;
               value.readFields(in);                 // read value
             } finally {
@@ -609,4 +611,19 @@ public class Client {
       return address.hashCode() ^ System.identityHashCode(ticket);
     }
   }  
+
+  void simulateError(boolean flag) {
+    simulateError = flag;
+  }
+ 
+  // If errors are being simulated, then wait.
+  private void waitForEndSimulation() {
+    while (simulateError) {
+      try {
+        LOG.info("RPC Client waiting for simulation to end");
+        Thread.sleep(1000);
+      } catch (InterruptedException e) {
+      }
+    }
+  }
 }

src/java/org/apache/hadoop/ipc/RPC.java

@@ -30,6 +30,7 @@ import java.net.SocketTimeoutException;
 import java.io.*;
 import java.util.Map;
 import java.util.HashMap;
+import java.util.Collection;
 
 import javax.net.SocketFactory;
 
@@ -168,6 +169,17 @@ public class RPC {
     CLIENTS.clear();
   }
 
+  /*
+   * remove specified client from the list of clients.
+   */
+  static synchronized void removeClients() {
+    CLIENTS.clear();
+  }
+
+  static synchronized Collection allClients() {
+    return CLIENTS.values();
+  }
+
   private static class Invoker implements InvocationHandler {
     private InetSocketAddress address;
     private UserGroupInformation ticket;
@@ -415,5 +427,4 @@ public class RPC {
       value = value.substring(0, 55)+"...";
     LOG.info(value);
   }
-  
 }

src/java/org/apache/hadoop/ipc/Server.java

@@ -23,6 +23,7 @@ import java.io.DataInputStream;
 import java.io.DataOutputStream;
 import java.io.BufferedOutputStream;
 import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
 
 import java.nio.ByteBuffer;
 import java.nio.channels.SelectionKey;
@@ -44,12 +45,13 @@ import java.util.List;
 import java.util.Iterator;
 import java.util.Random;
 
-import org.apache.commons.logging.*;
-
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.io.ObjectWritable;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.io.WritableUtils;
+import org.apache.hadoop.util.ReflectionUtils;
 import org.apache.hadoop.ipc.SocketChannelOutputStream;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.util.*;
@@ -145,6 +147,7 @@ public abstract class Server {
   private int timeout;
   private long maxCallStartAge;
   private int maxQueueSize;
+  private int socketSendBufferSize;
 
   volatile private boolean running = true;         // true while server runs
   private LinkedList<Call> callQueue = new LinkedList<Call>(); // queued calls
@@ -154,6 +157,7 @@ public abstract class Server {
   //maintain a list
   //of client connections
   private Listener listener = null;
+  private Responder responder = null;
   private int numConnections = 0;
   private Handler[] handlers = null;
 
@@ -191,17 +195,24 @@ public abstract class Server {
     private Writable param;                       // the parameter passed
     private Connection connection;                // connection to client
     private long receivedTime;                    // the time received
+    private ByteBuffer response;                      // the response for this call
 
     public Call(int id, Writable param, Connection connection) {
       this.id = id;
       this.param = param;
       this.connection = connection;
       this.receivedTime = System.currentTimeMillis();
+      this.response = null;
     }
     
+    @Override
     public String toString() {
       return param.toString() + " from " + connection.toString();
     }
+
+    public void setResponse(ByteBuffer response) {
+      this.response = response;
+    }
   }
 
   /** Listens on the socket. Creates jobs for the handler threads*/
@@ -288,6 +299,7 @@ public abstract class Server {
       }
     }
 
+    @Override
     public void run() {
       LOG.info(getName() + ": starting");
       SERVER.set(Server.this);
@@ -428,6 +440,234 @@ public abstract class Server {
     }
   }
 
+  // Sends responses of RPC back to clients.
+  private class Responder extends Thread {
+    private Selector writeSelector;
+    private boolean pending;         // call waiting to be enqueued
+
+    Responder() throws IOException {
+      this.setName("IPC Server Responder");
+      this.setDaemon(true);
+      writeSelector = Selector.open(); // create a selector
+      pending = false;
+    }
+
+    @Override
+    public void run() {
+      LOG.info(getName() + ": starting");
+      SERVER.set(Server.this);
+      long lastPurgeTime = 0;   // last check for old calls.
+
+      while (running) {
+        SelectionKey key = null;
+        try {
+          waitPending();     // If a channel is being registered, wait.
+          writeSelector.select(maxCallStartAge);
+          Iterator iter = writeSelector.selectedKeys().iterator();
+          while (iter.hasNext()) {
+            key = (SelectionKey)iter.next();
+            iter.remove();
+            try {
+              if (key.isValid() && key.isWritable()) {
+                  doAsyncWrite(key);
+              }
+            } catch (IOException e) {
+              LOG.info(getName() + ": doAsyncWrite threw exception " + e);
+              key.cancel();
+            }
+            key = null;
+          }
+          long now = System.currentTimeMillis();
+          if (now < lastPurgeTime + maxCallStartAge) {
+            continue;
+          }
+          lastPurgeTime = now;
+          //
+          // If there were some calls that have not been sent out for a
+          // long time, discard them.
+          //
+          LOG.debug("Checking for old call responses.");
+          iter = writeSelector.keys().iterator();
+          while (iter.hasNext()) {
+            key = (SelectionKey)iter.next();
+            try {
+              doPurge(key, now);
+            } catch (IOException e) {
+              LOG.warn("Error in purging old calls " + e);
+            }
+          }
+        } catch (OutOfMemoryError e) {
+          //
+          // we can run out of memory if we have too many threads
+          // log the event and sleep for a minute and give
+          // some thread(s) a chance to finish
+          //
+          LOG.warn("Out of Memory in server select", e);
+          try { Thread.sleep(60000); } catch (Exception ie) {}
+        } catch (Exception e) {
+          LOG.warn("Exception in Responder " + e);
+        }
+      }
+      LOG.info("Stopping " + this.getName());
+    }
+
+    private void doAsyncWrite(SelectionKey key) throws IOException {
+      Call call = (Call)key.attachment();
+      if (call == null) {
+        return;
+      }
+      if (key.channel() != call.connection.channel) {
+        throw new IOException("doAsyncWrite: bad channel");
+      }
+      if (processResponse(call.connection.responseQueue)) {
+        key.cancel();          // remove item from selector.
+      }
+    }
+
+    //
+    // Remove calls that have been pending in the responseQueue 
+    // for a long time.
+    //
+    private void doPurge(SelectionKey key, long now) throws IOException {
+      Call call = (Call)key.attachment();
+      if (call == null) {
+        return;
+      }
+      if (key.channel() != call.connection.channel) {
+        LOG.info("doPurge: bad channel");
+        return;
+      }
+      LinkedList<Call> responseQueue = call.connection.responseQueue;
+      synchronized (responseQueue) {
+        Iterator iter = responseQueue.listIterator(0);
+        while (iter.hasNext()) {
+          call = (Call)iter.next();
+          if (now > call.receivedTime + maxCallStartAge) {
+            LOG.info(getName() + ", call " + call +
+                     ": response discarded for being too old (" +
+                     (now - call.receivedTime) + ")");
+            iter.remove();
+          }
+        }
+
+        // If all the calls for this channel were removed, then 
+        // remove this channel from the selector
+        if (responseQueue.size() == 0) {
+          key.cancel();
+        } 
+      }
+    }
+
+    // Processes one response. Returns true if there are no more pending
+    // data for this channel.
+    //
+    private boolean processResponse(LinkedList<Call> responseQueue) throws IOException {
+      boolean error = true;
+      boolean done = false;       // there is more data for this channel.
+      int numElements = 0;
+      Call call = null;
+      try {
+        synchronized (responseQueue) {
+          //
+          // If there are no items for this channel, then we are done
+          //
+          numElements = responseQueue.size();
+          if (numElements == 0) {
+            error = false;
+            return true;              // no more data for this channel.
+          }
+          //
+          // Extract the first call
+          //
+          int numBytes = 0;
+          call = responseQueue.removeFirst();
+          SocketChannel channel = call.connection.channel;
+          if (LOG.isDebugEnabled()) {
+            LOG.debug(getName() + ": responding to #" + call.id + " from " +
+                      call.connection);
+          }
+          //
+          // Send as much data as we can in the non-blocking fashion
+          //
+          numBytes = channel.write(call.response);
+          if (!call.response.hasRemaining()) {
+            if (numElements == 1) {    // last call fully processes.
+              done = true;             // no more data for this channel.
+            } else {
+              done = false;            // more calls pending to be sent.
+            }
+            if (LOG.isDebugEnabled()) {
+              LOG.debug(getName() + ": responding to #" + call.id + " from " +
+                        call.connection + " Wrote " + numBytes + " bytes.");
+            }
+          } else {
+            //
+            // If we were unable to write the entire response out, then 
+            // insert in Selector queue. 
+            //
+            call.connection.responseQueue.addFirst(call); 
+            setPending();
+            try {
+              // Wakeup the thread blocked on select, only then can the call 
+              // to channel.register() complete.
+              writeSelector.wakeup();
+              SelectionKey readKey = channel.register(writeSelector, 
+                                                      SelectionKey.OP_WRITE);
+              readKey.attach(call);
+            } finally {
+              clearPending();
+            }
+            if (LOG.isDebugEnabled()) {
+              LOG.debug(getName() + ": responding to #" + call.id + " from " +
+                        call.connection + " Wrote partial " + numBytes + 
+                        " bytes.");
+            }
+            done = false;             // this call not fully processed.
+          }
+          error = false;              // everything went off well
+        }
+      } finally {
+        if (error && call != null) {
+          LOG.warn(getName()+", call " + call + ": output error");
+          done = true;               // error. no more data for this channel.
+          synchronized (connectionList) {
+            if (connectionList.remove(call.connection))
+              numConnections--;
+          }
+          call.connection.close();
+        }
+      }
+      return done;
+    }
+
+    //
+    // Enqueue a response from the application.
+    //
+    void doRespond(Call call) throws IOException {
+      synchronized (call.connection.responseQueue) {
+        call.connection.responseQueue.addLast(call);
+        if (call.connection.responseQueue.size() == 1) {
+          processResponse(call.connection.responseQueue);
+        }
+      }
+    }
+
+    private synchronized void setPending() {   // call waiting to be enqueued.
+      pending = true;
+    }
+
+    private synchronized void clearPending() { // call done enqueueing.
+      pending = false;
+      notify();
+    }
+
+    private synchronized void waitPending() throws InterruptedException {
+      while (pending) {
+        wait();
+      }
+    }
+  }
+
   /** Reads calls from a connection and queues them for handling. */
   private class Connection {
     private boolean versionRead = false; //if initial signature and
@@ -438,8 +678,7 @@ public abstract class Server {
     private SelectionKey key;
     private ByteBuffer data;
     private ByteBuffer dataLengthBuffer;
-    private DataOutputStream out;
-    private SocketChannelOutputStream channelOut;
+    private LinkedList<Call> responseQueue;
     private long lastContact;
     private int dataLength;
     private Socket socket;
@@ -457,9 +696,6 @@ public abstract class Server {
       this.data = null;
       this.dataLengthBuffer = ByteBuffer.allocate(4);
       this.socket = channel.socket();
-      this.out = new DataOutputStream
-        (new BufferedOutputStream(
-          this.channelOut = new SocketChannelOutputStream(channel)));
       InetAddress addr = socket.getInetAddress();
       if (addr == null) {
         this.hostAddress = "*Unknown*";
@@ -467,8 +703,18 @@ public abstract class Server {
         this.hostAddress = addr.getHostAddress();
       }
       this.remotePort = socket.getPort();
+      this.responseQueue = new LinkedList<Call>();
+      if (socketSendBufferSize != 0) {
+        try {
+          socket.setSendBufferSize(socketSendBufferSize);
+        } catch (IOException e) {
+          LOG.warn("Connection: unable to set socket send buffer size to " +
+                   socketSendBufferSize);
+        }
+      }
     }   
 
+    @Override
     public String toString() {
       return getHostAddress() + ":" + remotePort; 
     }
@@ -516,7 +762,9 @@ public abstract class Server {
           if (!HEADER.equals(dataLengthBuffer) || version != CURRENT_VERSION) {
             //Warning is ok since this is not supposed to happen.
             LOG.warn("Incorrect header or version mismatch from " + 
-                     hostAddress + ":" + remotePort);
+                     hostAddress + ":" + remotePort +
+                     " got version " + version + 
+                     " expected version " + CURRENT_VERSION);
             return -1;
           }
           dataLengthBuffer.clear();
@@ -589,8 +837,6 @@ public abstract class Server {
       if (!channel.isOpen())
         return;
       try {socket.shutdownOutput();} catch(Exception e) {}
-      try {out.close();} catch(Exception e) {}
-      try {channelOut.destroy();} catch(Exception e) {}
       if (channel.isOpen()) {
         try {channel.close();} catch(Exception e) {}
       }
@@ -607,9 +853,11 @@ public abstract class Server {
       this.setName("IPC Server handler "+ instanceNumber + " on " + port);
     }
 
+    @Override
     public void run() {
       LOG.info(getName() + ": starting");
       SERVER.set(Server.this);
+      ByteArrayOutputStream buf = new ByteArrayOutputStream(10240);
       while (running) {
         try {
           Call call;
@@ -648,28 +896,20 @@ public abstract class Server {
             error = StringUtils.stringifyException(e);
           }
           CurCall.set(null);
-            
-          DataOutputStream out = call.connection.out;
-          synchronized (out) {
-            try {
-              out.writeInt(call.id);                // write call id
-              out.writeBoolean(error!=null);        // write error flag
-              if (error == null) {
-                value.write(out);
-              } else {
-                WritableUtils.writeString(out, errorClass);
-                WritableUtils.writeString(out, error);
-              }
-              out.flush();
-            } catch (Exception e) {
-              LOG.warn(getName()+", call "+call+": output error", e);
-              synchronized (connectionList) {
-                if (connectionList.remove(call.connection))
-                  numConnections--;
-              }
-              call.connection.close();
-            }
+
+          buf.reset();
+          DataOutputStream out = new DataOutputStream(buf);
+          out.writeInt(call.id);                // write call id
+          out.writeBoolean(error != null);      // write error flag
+
+          if (error == null) {
+            value.write(out);
+          } else {
+            WritableUtils.writeString(out, errorClass);
+            WritableUtils.writeString(out, error);
           }
+          call.setResponse(ByteBuffer.wrap(buf.toByteArray()));
+          responder.doRespond(call);
         } catch (InterruptedException e) {
           if (running) {                          // unexpected -- log it
             LOG.info(getName() + " caught: " + e, e);
@@ -695,6 +935,7 @@ public abstract class Server {
     this.paramClass = paramClass;
     this.handlerCount = handlerCount;
     this.timeout = conf.getInt("ipc.client.timeout", 10000);
+    this.socketSendBufferSize = 0;
     maxCallStartAge = (long) (timeout * MAX_CALL_QUEUE_TIME);
     maxQueueSize = handlerCount * MAX_QUEUE_SIZE_PER_HANDLER;
     this.maxIdleTime = conf.getInt("ipc.client.maxidletime", 120000);
@@ -704,13 +945,20 @@ public abstract class Server {
     // Start the listener here and let it bind to the port
     listener = new Listener();
     this.port = listener.getAddress().getPort();    
+
+    // Create the responder here
+    responder = new Responder();
   }
 
   /** Sets the timeout used for network i/o. */
   public void setTimeout(int timeout) { this.timeout = timeout; }
 
+  /** Sets the socket buffer size used for responding to RPCs */
+  public void setSocketSendBufSize(int size) { this.socketSendBufferSize = size; }
+
   /** Starts the service.  Must be called before any calls will be handled. */
   public synchronized void start() throws IOException {
+    responder.start();
     listener.start();
     handlers = new Handler[handlerCount];
     
@@ -733,6 +981,7 @@ public abstract class Server {
     }
     listener.interrupt();
     listener.doStop();
+    responder.interrupt();
     notifyAll();
   }
 

src/test/org/apache/hadoop/ipc/TestRPC.java

@@ -25,6 +25,8 @@ import java.lang.reflect.Method;
 import junit.framework.TestCase;
 
 import java.util.Arrays;
+import java.util.Collection;
+import java.util.Iterator;
 
 import org.apache.commons.logging.*;
 
@@ -43,6 +45,9 @@ public class TestRPC extends TestCase {
   
   private static Configuration conf = new Configuration();
 
+  int datasize = 1024*100;
+  int numThreads = 50;
+
   public TestRPC(String name) { super(name); }
 	
   public interface TestProtocol extends VersionedProtocol {
@@ -56,6 +61,7 @@ public class TestRPC extends TestCase {
     int add(int[] values) throws IOException;
     int error() throws IOException;
     void testServerGet() throws IOException;
+    int[] exchange(int[] values) throws IOException;
   }
 
   public class TestImpl implements TestProtocol {
@@ -95,8 +101,117 @@ public class TestRPC extends TestCase {
       }
     }
 
+    public int[] exchange(int[] values) {
+      int sum = 0;
+      for (int i = 0; i < values.length; i++) {
+        values[i] = i;
+      }
+      return values;
+    }
+  }
+
+  //
+  // an object that does a bunch of transactions
+  //
+  static class Transactions implements Runnable {
+    int datasize;
+    TestProtocol proxy;
+
+    Transactions(TestProtocol proxy, int datasize) {
+      this.proxy = proxy;
+      this.datasize = datasize;
+    }
+
+    // do two RPC that transfers data.
+    public void run() {
+      int[] indata = new int[datasize];
+      int[] outdata = null;
+      int val = 0;
+      try {
+        outdata = proxy.exchange(indata);
+        val = proxy.add(1,2);
+      } catch (IOException e) {
+        assertTrue("Exception from RPC exchange() "  + e, false);
+      }
+      assertEquals(indata.length, outdata.length);
+      assertEquals(val, 3);
+      for (int i = 0; i < outdata.length; i++) {
+        assertEquals(outdata[i], i);
+      }
+    }
   }
 
+  //
+  // A class that does an RPC but does not read its response.
+  //
+  static class SlowRPC implements Runnable {
+    private TestProtocol proxy;
+    private volatile boolean done;
+   
+    SlowRPC(TestProtocol proxy) {
+      this.proxy = proxy;
+      done = false;
+    }
+
+    boolean isDone() {
+      return done;
+    }
+
+    public void run() {
+      try {
+        proxy.ping();           // this would hang until simulateError is false
+        done = true;
+      } catch (IOException e) {
+        assertTrue("SlowRPC ping exception " + e, false);
+      }
+    }
+  }
+
+  public void testSlowRpc() throws Exception {
+    System.out.println("Testing Slow RPC");
+    Server server = RPC.getServer(new TestImpl(), ADDRESS, 0, conf);
+    server.start();
+
+    InetSocketAddress addr = server.getListenerAddress();
+
+    // create a client and make an RPC that does not read its response
+    //
+    TestProtocol proxy1 =
+      (TestProtocol)RPC.getProxy(TestProtocol.class, TestProtocol.versionID, addr, conf);
+    Collection collection = RPC.allClients();
+    assertTrue("There should be only one client.", collection.size() == 1);
+    Iterator iter = collection.iterator();
+    Client client = (Client) iter.next();
+
+    client.simulateError(true);
+    RPC.removeClients();
+    SlowRPC slowrpc = new SlowRPC(proxy1);
+    Thread thread = new Thread(slowrpc, "SlowRPC");
+    thread.start();
+    assertTrue("Slow RPC should not have finished1.", !slowrpc.isDone());
+
+    // create another client and make another RPC to the same server. This
+    // should complete even though the first one is still hanging.
+    //
+    TestProtocol proxy2 =
+      (TestProtocol)RPC.getProxy(TestProtocol.class, TestProtocol.versionID, addr, conf);
+    proxy2.ping();
+
+    // verify that the first RPC is still stuck
+    assertTrue("Slow RPC should not have finished2.", !slowrpc.isDone());
+
+    // Make the first RPC process its response. 
+    client.simulateError(false);
+    while (!slowrpc.isDone()) {
+      System.out.println("Waiting for slow RPC to get done.");
+      try {
+        Thread.sleep(1000);
+      } catch (InterruptedException e) {}
+    }
+    server.stop();
+  }
+
+
   public void testCalls() throws Exception {
     Server server = RPC.getServer(new TestImpl(), ADDRESS, 0, conf);
     server.start();
@@ -142,6 +257,26 @@ public class TestRPC extends TestCase {
 
     proxy.testServerGet();
 
+    // create multiple threads and make them do large data transfers
+    System.out.println("Starting multi-threaded RPC test...");
+    server.setSocketSendBufSize(1024);
+    Thread threadId[] = new Thread[numThreads];
+    for (int i = 0; i < numThreads; i++) {
+      Transactions trans = new Transactions(proxy, datasize);
+      threadId[i] = new Thread(trans, "TransactionThread-" + i);
+      threadId[i].start();
+    }
+
+    // wait for all transactions to get over
+    System.out.println("Waiting for all threads to finish RPCs...");
+    for (int i = 0; i < numThreads; i++) {
+      try {
+        threadId[i].join();
+      } catch (InterruptedException e) {
+        i--;      // retry
+      }
+    }
+
     // try some multi-calls
     Method echo =
       TestProtocol.class.getMethod("echo", new Class[] { String.class });
@@ -161,5 +296,4 @@ public class TestRPC extends TestCase {
     new TestRPC("test").testCalls();
 
   }
-
 }