From 9b71b0da624ebcd6e3e06b4f325e0ddffd1c8f47 Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Thu, 7 Apr 2016 16:23:24 -0700 Subject: Make _wake_r socket non-blocking; drop select from _clear_wake_fd --- kafka/client_async.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) (limited to 'kafka/client_async.py') diff --git a/kafka/client_async.py b/kafka/client_async.py index e921fa4..ca51987 100644 --- a/kafka/client_async.py +++ b/kafka/client_async.py @@ -98,6 +98,7 @@ class KafkaClient(object): 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 @@ class KafkaClient(object): 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): -- cgit v1.2.1