From bc18677872477e75918cf1a8a75ef513dfffdd7d Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Sat, 16 Jul 2016 20:46:16 -0700 Subject: Protect writes to wakeup socket with threading lock --- kafka/client_async.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) (limited to 'kafka/client_async.py') diff --git a/kafka/client_async.py b/kafka/client_async.py index 6fa9434..a4c368b 100644 --- a/kafka/client_async.py +++ b/kafka/client_async.py @@ -6,6 +6,7 @@ import heapq import itertools import logging import random +import threading # selectors in stdlib as of py3.4 try: @@ -158,6 +159,7 @@ class KafkaClient(object): self._bootstrap_fails = 0 self._wake_r, self._wake_w = socket.socketpair() self._wake_r.setblocking(False) + self._wake_lock = threading.Lock() self._selector.register(self._wake_r, selectors.EVENT_READ) self._closed = False self._bootstrap(collect_hosts(self.config['bootstrap_servers'])) @@ -747,10 +749,12 @@ class KafkaClient(object): raise Errors.NoBrokersAvailable() def wakeup(self): - if self._wake_w.send(b'x') != 1: - log.warning('Unable to send to wakeup socket!') + with self._wake_lock: + if self._wake_w.send(b'x') != 1: + log.warning('Unable to send to wakeup socket!') def _clear_wake_fd(self): + # reading from wake socket should only happen in a single thread while True: try: self._wake_r.recv(1024) -- cgit v1.2.1