Make _wake_r socket non-blocking; drop select from _clear_wake_fd

dpkp · dpkp · commit 9b71b0da624e · 2016-04-08T09:23:57.000-07:00
diff --git a/kafka/client_async.py b/kafka/client_async.py
@@ -98,6 +98,7 @@ def __init__(self, **configs):
         self._bootstrap_fails = 0
         self._bootstrap(collect_hosts(self.config['bootstrap_servers']))
         self._wake_r, self._wake_w = socket.socketpair()
+        self._wake_r.setblocking(False)
 
     def __del__(self):
         self._wake_r.close()
@@ -682,10 +683,10 @@ def wakeup(self):
 
     def _clear_wake_fd(self):
         while True:
-            fds, _, _ = select.select([self._wake_r], [], [], 0)
-            if not fds:
+            try:
+                self._wake_r.recv(1)
+            except:
                 break
-            self._wake_r.recv(1)
 
 
 class DelayedTaskQueue(object):
