Merge pull request #395 from jacksontj/system_call_interrupts

System call interrupts

Author: bbangert

kazoo/handlers/threading.py

@@ -143,17 +143,26 @@ def stop(self):
             python2atexit.unregister(self.stop)
 
     def select(self, *args, **kwargs):
-        try:
-            return select.select(*args, **kwargs)
-        except select.error as ex:
-            # if the system call was interrupted, we'll return as a timeout
-            # in Python 3, system call interruptions are a native exception
-            # in Python 2, they are not
-            errnum = ex.errno if isinstance(ex, OSError) else ex[0]
-            # to mimic a timeout, we return the same thing select would
-            if errnum == errno.EINTR:
-                return ([], [], [])
-            raise
+        # select() takes no kwargs, so it will be in args
+        timeout = args[3] if len(args) == 4 else None
+        # either the time to give up, or None
+        end = (time.time() + timeout) if timeout else None
+        while end is None or time.time() < end:
+            if end is not None:
+                args = list(args)  # make a list, since tuples aren't mutable
+                args[3] = end - time.time()  # set the timeout to the remaining time
+            try:
+                return select.select(*args, **kwargs)
+            except select.error as ex:
+                # if the system call was interrupted, we'll retry until timeout
+                # in Python 3, system call interruptions are a native exception
+                # in Python 2, they are not
+                errnum = ex.errno if isinstance(ex, OSError) else ex[0]
+                if errnum == errno.EINTR:
+                    continue
+                raise
+        # if we hit our timeout, lets return as a timeout
+        return ([], [], [])
 
     def socket(self):
         return utils.create_tcp_socket(socket)

kazoo/handlers/utils.py

@@ -3,6 +3,7 @@
 import errno
 import functools
 import select
+import time
 
 HAS_FNCTL = True
 try:
@@ -182,12 +183,31 @@ def create_tcp_socket(module):
 
 
 def create_tcp_connection(module, address, timeout=None):
+    end = None
     if timeout is None:
         # thanks to create_connection() developers for
         # this ugliness...
-        timeout = module._GLOBAL_DEFAULT_TIMEOUT
+        timeout = module.getdefaulttimeout()
+    if timeout is not None:
+        end = time.time() + timeout
+    sock = None
+
+    while end is None or time.time() < end:
+        try:
+            # if we got a timeout, lets ensure that we decrement the time
+            # otherwise there is no timeout set and we'll call it as such
+            timeout_at = end if end is None else end - time.time()
+            sock = module.create_connection(address, timeout_at)
+            break
+        except Exception as ex:
+            errnum = ex.errno if isinstance(ex, OSError) else ex[0]
+            if errnum == errno.EINTR:
+                continue
+            raise
+
+    if sock is None:
+        raise module.error
 
-    sock = module.create_connection(address, timeout)
     _set_default_tcpsock_options(module, sock)
     return sock