Browse Source

HUE-550. Switching to LIFO queue for Thrift connection pool.

This was a subtle bug and a fun debugging experience.  It's not every
day you get to use YourKit, wireshark, strace, and log-based tools at
the same time!

I noticed, while doing some other testing, that the filebrowser was very
slow, but only sometimes.  In debugging, I found that every second
refresh, roughly, was slow.  Though I had originally attributed this
slowness to the UI, the logs clearly showed that some Thrift calls were
taking almost 5 seconds.

Hue has a connection pool for Thrift clients.  This pool had a default
size of 10.  Meanwhile, the default pool size on the namenode Thrift
plugin was 5.  So, what would happen is that a Thrift client would be
requested, and the pool would round robin to the next client.  Client c1
would lock up t1 on the server.  c2 would lock up t2.  And so on till c5
and t5.  Then, when c6 came around, it would wait until the server had a
spare thread.  This would free up as soon as c1 would time out (5
seconds!).

The solution is two-fold: (1) instead of FIFO for the client pool,
we switch to a LIFO solution.  Then the same socket is constantly
re-used.  (2) We up the thread pool size to be the same as the
connection pool size by default.

There were several moderately interesting things this analysis
ruled out:
 (1) The namenode is not slow. We do a handful of
 operations for every file browser page, which isn't
 great, but it's not slow.  YourKit didn't even
 sample a single stack trace; nothing was going on.
 (2) We thought there was a problem in the reconnection
 logic.  It turns out (as the comment points out) that
 select() will return a socket as readable if it's
 been closed or timed out.  This is all and good; that
 code seems to be working correctly.
 (3) Funky TCP stuff. I was worried that somehow
 connection re-use was broken. It seems fine; wireshark
 worked, but didn't say anything too interesting.

In terms of debugging, the major point was isolating
the call by writing the following script.  If I'd
done this an hour earlier (instead of hitting refresh
on a browser), this would have taken less time.
Kudos to Todd for mkaing me do it.

  import time
  import os

  os.environ["DJANGO_SETTINGS_MODULE"] = 'desktop.settings'

  from desktop.lib import fsmanager
  fs = fsmanager.get_default_hdfs()[1]
  fs.setuser("hdfs")

  i = 0
  while True:
    i += 1
    now = time.time()
    fs.stats("/")
    diff = time.time() - now
    print i, diff

This clearly showed a pattern of 5 instantaneous calls,
followed by one that took 5 seconds, followed by another 4
quick calls, etc.  After that, I recognized that adjusting
the server pool size to 50 eliminated the problem, and
adjusting it to 1 exacerbated it.

Kudos to Todd for the insight that the queue was working
backwards, and to Henry for his assistance as well.
Philip Zeyliger 14 years ago
parent
commit
51630f4298

+ 25 - 2
desktop/core/src/desktop/lib/thrift_util.py

@@ -43,6 +43,26 @@ MAX_RECURSION_DEPTH = 50
 WARN_LEVEL_CALL_DURATION_MS = 5000
 WARN_LEVEL_CALL_DURATION_MS = 5000
 INFO_LEVEL_CALL_DURATION_MS = 1000
 INFO_LEVEL_CALL_DURATION_MS = 1000
 
 
+class LifoQueue(Queue.Queue):
+    '''
+    Variant of Queue that retrieves most recently added entries first.
+
+    This LIFO Queue is included in python2.7 (or 2.6) and later,
+    but it's a simple subclass, so we "backport" it here.
+    '''
+
+    def _init(self, maxsize):
+        self.queue = []
+
+    def _qsize(self, len=len):
+        return len(self.queue)
+
+    def _put(self, item):
+        self.queue.append(item)
+
+    def _get(self):
+        return self.queue.pop()
+
 class ConnectionConfig(object):
 class ConnectionConfig(object):
   """ Struct-like class encapsulating the configuration of a Thrift client. """
   """ Struct-like class encapsulating the configuration of a Thrift client. """
   def __init__(self, klass, host, port, service_name,
   def __init__(self, klass, host, port, service_name,
@@ -117,7 +137,7 @@ class ConnectionPooler(object):
       self.dictlock.acquire()
       self.dictlock.acquire()
       try:
       try:
         if _get_pool_key(conf) not in self.pooldict:
         if _get_pool_key(conf) not in self.pooldict:
-          q = Queue.Queue(self.poolsize)
+          q = LifoQueue(self.poolsize)
           self.pooldict[_get_pool_key(conf)] = q
           self.pooldict[_get_pool_key(conf)] = q
           for i in xrange(self.poolsize):
           for i in xrange(self.poolsize):
             client = construct_superclient(conf)
             client = construct_superclient(conf)
@@ -246,7 +266,10 @@ class PooledClient(object):
               if rlist:
               if rlist:
                 # the socket is readable, meaning there is either data from a previous call
                 # the socket is readable, meaning there is either data from a previous call
                 # (i.e our protocol is out of sync), or the connection was shut down on the
                 # (i.e our protocol is out of sync), or the connection was shut down on the
-                # remote side. Either way, we need to reopen the connection
+                # remote side. Either way, we need to reopen the connection.
+                # If the socket was closed remotely, btw, socket.read() will return
+                # an empty string.  This is a fairly normal condition, btw, since
+                # there are timeouts on both the server and client sides.
                 superclient.transport.close()
                 superclient.transport.close()
                 superclient.transport.open()
                 superclient.transport.open()
 
 

+ 1 - 1
desktop/libs/hadoop/java/src/main/java/org/apache/hadoop/thriftfs/ThriftPluginServer.java

@@ -107,7 +107,7 @@ public class ThriftPluginServer implements Configurable, Runnable {
 
 
       TServerTransport transport = new TServerSocket(sock, socketTimeout);
       TServerTransport transport = new TServerSocket(sock, socketTimeout);
       SanerThreadPoolServer.Options options = new SanerThreadPoolServer.Options();
       SanerThreadPoolServer.Options options = new SanerThreadPoolServer.Options();
-      options.minWorkerThreads = conf.getInt(ThriftFsConfig.DFS_THRIFT_THREADS_MIN_KEY, 5);
+      options.minWorkerThreads = conf.getInt(ThriftFsConfig.DFS_THRIFT_THREADS_MIN_KEY, 10);
       options.maxWorkerThreads = conf.getInt(ThriftFsConfig.DFS_THRIFT_THREADS_MAX_KEY, 20);
       options.maxWorkerThreads = conf.getInt(ThriftFsConfig.DFS_THRIFT_THREADS_MAX_KEY, 20);
       options.stopTimeoutVal = conf.getInt(ThriftFsConfig.DFS_THRIFT_TIMEOUT_KEY, 60);
       options.stopTimeoutVal = conf.getInt(ThriftFsConfig.DFS_THRIFT_TIMEOUT_KEY, 60);
       options.stopTimeoutUnit = TimeUnit.SECONDS;
       options.stopTimeoutUnit = TimeUnit.SECONDS;