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

mitogen/core.py

@@ -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,22 +805,23 @@ 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
+                raise TimeoutError()
-            if _tls.rsock in rfds:
+
+            assert _tls.rsock in rfds
             _tls.rsock.recv(1)
             return self.queue.pop(0)
         finally:

mitogen/master.py

@@ -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.put(value)
-        self._latch.wake()
         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._latch.get(timeout=timeout)
-            recv = self._queue.get()
             try:
                 msg = recv.get(block=False)
                 if self._oneshot: