فهرست منبع

HUE-1239 [core] PooledClient should use poll

Fall back to select
Abraham Elmahrek 12 سال پیش
والد
کامیت
311e51a
3فایلهای تغییر یافته به همراه66 افزوده شده و 12 حذف شده
  1. 7 0
      desktop/core/src/desktop/conf.py
  2. 49 0
      desktop/core/src/desktop/lib/python_util.py
  3. 10 12
      desktop/core/src/desktop/lib/thrift_util.py

+ 7 - 0
desktop/core/src/desktop/conf.py

@@ -92,6 +92,13 @@ COLLECT_USAGE = Config(
   type=coerce_bool,
   default=True)
 
+POLL_ENABLED = Config(
+  key="poll_enabled",
+  help=_("Use poll(2) in Hue thrift pool."),
+  type=coerce_bool,
+  private=True,
+  default=True)
+
 
 def is_https_enabled():
   return bool(SSL_CERTIFICATE.get() and SSL_PRIVATE_KEY.get())

+ 49 - 0
desktop/core/src/desktop/lib/python_util.py

@@ -17,6 +17,13 @@
 # 
 # Extra python utils
 
+import select
+from django.utils.translation import ugettext as _
+from desktop import conf
+
+
+__all__ = ['CaseInsensitiveDict', 'create_synchronous_io_multiplexer']
+
 
 class CaseInsensitiveDict(dict):
   def __setitem__(self, key, value):
@@ -31,3 +38,45 @@ class CaseInsensitiveDict(dict):
   @classmethod
   def from_dict(cls, _dict):
     return CaseInsensitiveDict([(isinstance(key, basestring) and key.lower() or key, _dict[key]) for key in _dict])
+
+
+class SynchronousIOMultiplexer(object):
+  def read(self, rd):
+    raise NotImplementedError(_('"read" method is not implemented'))
+
+  def write(self, rd):
+    raise NotImplementedError(_('"write" method is not implemented'))
+
+  def error(self, rd):
+    raise NotImplementedError(_('"error" method is not implemented'))
+
+
+class SelectSynchronousIOMultiplexer(SynchronousIOMultiplexer):
+  def __init__(self, timeout=0):
+    self.timeout = 0
+
+  def read(self, fds):
+    rlist, wlist, xlist = select.select(fds, [], [], self.timeout)
+    return rlist
+
+
+class PollSynchronousIOMultiplexer(SynchronousIOMultiplexer):
+  def __init__(self, timeout=0):
+    self.timeout = 0
+
+  def read(self, fds):
+    poll_obj = select.poll()
+    for fd in fds:
+      poll_obj.register(fd, select.POLLIN)
+    event_list = poll_obj.poll(self.timeout)
+    return [fd_event_tuple[0] for fd_event_tuple in event_list]
+
+
+def create_synchronous_io_multiplexer(timeout=0):
+  if conf.POLL_ENABLED.get():
+    try:
+      from select import poll
+      return PollSynchronousIOMultiplexer(timeout)
+    except ImportError:
+      pass
+  return SelectSynchronousIOMultiplexer(timeout)

+ 10 - 12
desktop/core/src/desktop/lib/thrift_util.py

@@ -31,6 +31,7 @@ from thrift.transport.TSocket import TSocket
 from thrift.transport.TTransport import TBufferedTransport, TMemoryBuffer,\
                                         TTransportException
 from thrift.protocol.TBinaryProtocol import TBinaryProtocol
+from desktop.lib.python_util import create_synchronous_io_multiplexer
 from desktop.lib.thrift_sasl import TSaslClientTransport
 from desktop.lib.exceptions import StructuredException, StructuredThriftTransportException
 
@@ -279,17 +280,15 @@ class PooledClient(object):
             # Poke it to see if it's closed on the other end. This can happen if a connection
             # sits in the connection pool longer than the read timeout of the server.
             sock = _grab_transport_from_wrapper(superclient.transport).handle
-            if sock:
-              rlist,wlist,xlist = select.select([sock], [], [], 0)
-              if rlist:
-                # 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
-                # 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.open()
+            if sock and create_synchronous_io_multiplexer().read([sock]):
+              # 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
+              # 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.open()
 
             superclient.set_timeout(self.conf.timeout_seconds)
             return res(*args, **kwargs)
@@ -313,7 +312,6 @@ class PooledClient(object):
       return wrapper
 
 
-
 class SuperClient(object):
   """A wrapper for a Thrift Client that causes it to automatically
   reconnect on failure.