master: use queue-like Latch in Select() too.

pull/87/head
David Wilson 7 years ago
parent 371a259a5e
commit 0d57afc914

@ -78,7 +78,7 @@ else:
class Error(Exception):
def __init__(self, fmt, *args):
def __init__(self, fmt=None, *args):
if args:
fmt %= args
Exception.__init__(self, fmt)
@ -113,7 +113,7 @@ class StreamError(Error):
pass
class TimeoutError(StreamError):
class TimeoutError(Error):
pass
@ -805,24 +805,25 @@ class Latch(object):
if self.queue:
return self.queue.pop(0)
if not block:
return
raise TimeoutError()
self._tls_init()
self.wake_socks.append(_tls.wsock)
finally:
self.lock.release()
rfds, _, _ = select.select([_tls.rsock], [], [], timeout)
assert len(rfds) or timeout is None
assert len(rfds) or timeout is not None
self.lock.acquire()
try:
if _tls.wsock in self.wake_socks:
# Nothing woke us, remove stale entry.
self.wake_socks.remove(_tls.wsock)
return
if _tls.rsock in rfds:
_tls.rsock.recv(1)
return self.queue.pop(0)
raise TimeoutError()
assert _tls.rsock in rfds
_tls.rsock.recv(1)
return self.queue.pop(0)
finally:
self.lock.release()

@ -126,14 +126,12 @@ class Select(object):
def __init__(self, receivers=(), oneshot=True):
self._receivers = []
self._oneshot = oneshot
self._queue = Queue.Queue()
self._latch = mitogen.core.Latch()
for recv in receivers:
self.add(recv)
def _put(self, value):
self._queue.put(value)
self._latch.wake()
self._latch.put(value)
if self.notify:
self.notify(self)
@ -193,7 +191,7 @@ class Select(object):
self.remove(recv)
def empty(self):
return self._queue.empty()
return self._latch.empty()
empty_msg = 'Cannot get(), Select instance is empty'
@ -202,8 +200,7 @@ class Select(object):
raise SelectError(self.empty_msg)
while True:
self._latch.wait()
recv = self._queue.get()
recv = self._latch.get(timeout=timeout)
try:
msg = recv.get(block=False)
if self._oneshot:

Loading…
Cancel
Save