add gevent.wait(), gevent.iwait()

- remove waitall() from gevent/event.py
- reimplement joinall() with wait()
- remove joinall from __all__
- Hub.join() no longer accepts event= argument

gevent/__init__.py

@@ -18,7 +18,8 @@ __all__ = ['get_hub',
            'spawn',
            'spawn_later',
            'spawn_raw',
-           'joinall',
+           'iwait',
+           'wait',
            'killall',
            'Timeout',
            'with_timeout',
@@ -28,8 +29,7 @@ __all__ = ['get_hub',
            'kill',
            'signal',
            'fork',
-           'reinit',
-           'wait']
+           'reinit']
 
 
 import sys
@@ -39,7 +39,7 @@ if sys.platform == 'win32':
 del sys
 
 
-from gevent.hub import get_hub
+from gevent.hub import get_hub, iwait, wait
 from gevent.greenlet import Greenlet, joinall, killall
 spawn = Greenlet.spawn
 spawn_later = Greenlet.spawn_later
@@ -50,9 +50,4 @@ try:
 except ImportError:
     __all__.remove('fork')
 
-
-def wait(timeout=None, event=None):
-    return get_hub().join(timeout=timeout, event=event)
-
-
 run = wait  # XXX to be deleted (soon)

gevent/event.py

@@ -310,18 +310,3 @@ class AsyncResult(object):
             self.set(source.value)
         else:
             self.set_exception(source.exception)
-
-
-def waitall(events):
-    # QQQ add timeout?
-    from gevent.queue import Queue
-    queue = Queue()
-    put = queue.put
-    try:
-        for event in events:
-            event.rawlink(put)
-        for _ in xrange(len(events)):
-            queue.get()
-    finally:
-        for event in events:
-            event.unlink(put)

gevent/greenlet.py

 # Copyright (c) 2009-2012 Denis Bilenko. See LICENSE for details.
 
 import sys
-from gevent.hub import greenlet, getcurrent, get_hub, GreenletExit, Waiter, PY3
+from gevent.hub import greenlet, getcurrent, get_hub, GreenletExit, Waiter, PY3, iwait, wait
 from gevent.timeout import Timeout
 from collections import deque
 
@@ -396,39 +396,17 @@ def _kill(greenlet, exception, waiter):
     waiter.switch()
 
 
-try:
-    xrange
-except NameError:
-    xrange = range
-
-
 def joinall(greenlets, timeout=None, raise_error=False, count=None):
-    from gevent.queue import Queue
-    queue = Queue()
-    put = queue.put
-    if count is None:
-        count = len(greenlets)
-    timeout = Timeout.start_new(timeout)
-    try:
-        try:
-            for greenlet in greenlets:
-                greenlet.rawlink(put)
-            if raise_error:
-                for _ in xrange(count):
-                    greenlet = queue.get()
-                    if not greenlet.successful():
-                        raise greenlet.exception
-            else:
-                for _ in xrange(count):
-                    queue.get()
-        except:
-            if sys.exc_info()[1] is not timeout:
-                raise
-        finally:
-            for greenlet in greenlets:
-                greenlet.unlink(put)
-    finally:
-        timeout.cancel()
+    if not raise_error:
+        wait(greenlets, timeout=timeout)
+    else:
+        for obj in iwait(greenlets, timeout=timeout):
+            if getattr(obj, 'exception', None) is not None:
+                raise obj.exception
+            if count is not None:
+                count -= 1
+                if count <= 0:
+                    break
 
 
 def _killall3(greenlets, exception, waiter):

gevent/hub.py

@@ -408,15 +408,14 @@ class Hub(greenlet):
         # It is still possible to kill this greenlet with throw. However, in that case
         # switching to it is no longer safe, as switch will return immediatelly
 
-    def join(self, timeout=None, event=None):
+    def join(self, timeout=None):
         """Wait for the event loop to finish. Exits only when there are
         no more spawned greenlets, started servers, active timeouts or watchers.
 
         If *timeout* is provided, wait no longer for the specified number of seconds.
-        If *event* was provided, exit when it was signalled with :meth:`Event.set` method.
 
         Returns True if exited because the loop finished execution.
-        Returns False if exited because of timeout expired or event was signalled.
+        Returns False if exited because of timeout expired.
         """
         assert getcurrent() is self.parent, "only possible from the MAIN greenlet"
         if self.dead:
@@ -424,26 +423,18 @@ class Hub(greenlet):
 
         waiter = Waiter()
 
-        if event is not None:
-            switch = waiter.switch
-            event.rawlink(switch)
+        if timeout is not None:
+            timeout = self.loop.timer(timeout, ref=False)
+            timeout.start(waiter.switch)
 
         try:
-            if timeout is not None:
-                timeout = self.loop.timer(timeout, ref=False)
-                timeout.start(waiter.switch)
-
             try:
-                try:
-                    waiter.get()
-                except LoopExit:
-                    return True
-            finally:
-                if timeout is not None:
-                    timeout.stop()
+                waiter.get()
+            except LoopExit:
+                return True
         finally:
-            if event is not None:
-                event.unlink(switch)
+            if timeout is not None:
+                timeout.stop()
         return False
 
     def destroy(self, destroy_loop=None):
@@ -622,6 +613,41 @@ class Waiter(object):
     # and unwraps it in wait() thus checking that switch() was indeed called
 
 
+def iwait(objects, timeout=None):
+    # QQQ would be nice to support iterable here that can be generated slowly (why?)
+    waiter = Waiter()
+    switch = waiter.switch
+    if timeout is not None:
+        get_hub().loop.timer(timeout, priority=-1, ref=False).start(waiter.switch, _NONE)
+    try:
+        count = len(objects)
+        for obj in objects:
+            obj.rawlink(switch)
+        for _ in xrange(count):
+            item = waiter.get()
+            waiter.clear()
+            if item is _NONE:
+                return
+            yield item
+    finally:
+        for obj in objects:
+            obj.unlink(switch)
+
+
+def wait(objects=None, timeout=None, count=None):
+    if objects is None:
+        return get_hub().join(timeout=timeout)
+    result = []
+    if count is None:
+        return list(iwait(objects, timeout))
+    for obj in iwait(objects=objects, timeout=timeout):
+        result.append(obj)
+        count -= 1
+        if count <= 0:
+            break
+    return result
+
+
 class _NONE(object):
     "A special thingy you must never pass to any of gevent API"
     __slots__ = []

greentest/test_hub_join_timeout.py

@@ -32,7 +32,7 @@ for _a in xrange(2):
     for _ in xrange(2):
         x = gevent.spawn(lambda: 5)
         with no_time(SMALL):
-            result = gevent.get_hub().join(timeout=10)
+            result = gevent.wait(timeout=10)
         assert result is True, repr(result)
         assert x.dead, x
         assert x.value == 5, x
@@ -41,7 +41,7 @@ for _a in xrange(2):
     for _ in xrange(2):
         x = gevent.spawn_later(SMALL, lambda: 5)
         with expected_time(SMALL):
-            result = gevent.get_hub().join(timeout=10)
+            result = gevent.wait(timeout=10)
         assert result is True, repr(result)
         assert x.dead, x
 
@@ -49,12 +49,12 @@ for _a in xrange(2):
     for _ in xrange(2):
         x = gevent.spawn_later(10, lambda: 5)
         with expected_time(SMALL):
-            result = gevent.get_hub().join(timeout=SMALL)
+            result = gevent.wait(timeout=SMALL)
         assert result is False, repr(result)
         assert not x.dead, x
         x.kill()
         with no_time():
-            result = gevent.get_hub().join()
+            result = gevent.wait()
         assert result is True
 
     # exiting because of event (the spawned greenlet still runs)
@@ -63,21 +63,21 @@ for _a in xrange(2):
         event = Event()
         event_set = gevent.spawn_later(SMALL, event.set)
         with expected_time(SMALL):
-            result = gevent.get_hub().join(event=event)
-        assert result is False, repr(result)
+            result = gevent.wait([event])
+        assert result == [event], repr(result)
         assert not x.dead, x
         assert event_set.dead
         assert event.is_set()
         x.kill()
         with no_time():
-            result = gevent.get_hub().join()
+            result = gevent.wait()
         assert result is True
 
     # checking "ref=False" argument
     for _ in xrange(2):
         gevent.get_hub().loop.timer(10, ref=False).start(lambda: None)
         with no_time():
-            result = gevent.get_hub().join()
+            result = gevent.wait()
         assert result is True
 
     # checking "ref=False" attribute
@@ -86,5 +86,5 @@ for _a in xrange(2):
         w.start(lambda: None)
         w.ref = False
         with no_time():
-            result = gevent.get_hub().join()
+            result = gevent.wait()
         assert result is True