Fix race in Lock.release(). Still need to do other lock classes.

5 files changed, 82 insertions, 77 deletions

diff --git a/tests/locks_test.py b/tests/locks_test.py
index 65ea048..af65c97 100644
--- a/tests/locks_test.py
+++ b/tests/locks_test.py
@@ -126,6 +126,67 @@ class LockTests(unittest.TestCase):
             self.loop.run_until_complete, task)
         self.assertFalse(lock._waiters)
 
+    def test_cancel_race(self):
+        # XXX replace assert with self.assertXXX; remove dprint().
+        # Several tasks:
+        # - A acquires the lock
+        # - B is blocked in aqcuire()
+        # - C is blocked in aqcuire()
+        #
+        # Now, concurrently:
+        # - B is cancelled
+        # - A releases the lock
+        #
+        # If B's waiter is marked cancelled but not yet removed from
+        # _waiters, A's release() call will crash when trying to set
+        # B's waiter; instead, it should move on to C's waiter.
+
+        # Setup: A has the lock, b and c are waiting.
+        def dprint(*args): pass  # or 'dprint = print'
+        lock = locks.Lock(loop=self.loop)
+        @tasks.coroutine
+        def lockit(name, blocker):
+            dprint(name, 'acquiring...')
+            yield from lock.acquire()
+            dprint(name, 'acquired')
+            try:
+                if blocker is not None:
+                    dprint(name, 'blocking...')
+                    yield from blocker
+                    dprint(name, 'unlocked')
+            finally:
+                dprint(name, 'releasing...')
+                lock.release()
+                dprint(name, 'released')
+        fa = futures.Future(loop=self.loop)
+        ta = tasks.Task(lockit('A', fa), loop=self.loop)
+        test_utils.run_briefly(self.loop)
+        assert lock.locked()
+        fb = futures.Future(loop=self.loop)
+        tb = tasks.Task(lockit('B', None), loop=self.loop)
+        test_utils.run_briefly(self.loop)
+        assert len(lock._waiters) == 1
+        fc = futures.Future(loop=self.loop)
+        tc = tasks.Task(lockit('C', None), loop=self.loop)
+        test_utils.run_briefly(self.loop)
+        assert len(lock._waiters) == 2
+
+        # Create the race and check.
+        # Without the fix this failed at the last assert.
+        dprint('---create the race---')
+        fa.set_result(None)
+        tb.cancel()
+        assert lock._waiters[0].cancelled()
+        dprint(tb, lock)
+        dprint(fa, lock)
+        assert lock._waiters[0].cancelled()
+        test_utils.run_briefly(self.loop)
+        dprint(lock)
+        assert not lock.locked()
+        assert ta.done()
+        assert tb.cancelled()
+        assert tc.done()
+
     def test_release_not_acquired(self):
         lock = locks.Lock(loop=self.loop)
 
diff --git a/tests/tasks_test.py b/tests/tasks_test.py
index 022f814..8c26e3f 100644
--- a/tests/tasks_test.py
+++ b/tests/tasks_test.py
@@ -44,57 +44,6 @@ class TaskTests(unittest.TestCase):
         self.assertIs(t._loop, loop)
         loop.close()
 
-    def test_task_decorator(self):
-        @tasks.task
-        def notmuch():
-            yield from []
-            return 'ko'
-
-        try:
-            events.set_event_loop(self.loop)
-            t = notmuch()
-        finally:
-            events.set_event_loop(None)
-
-        self.assertIsInstance(t, tasks.Task)
-        self.loop.run_until_complete(t)
-        self.assertTrue(t.done())
-        self.assertEqual(t.result(), 'ko')
-
-    def test_task_decorator_func(self):
-        @tasks.task
-        def notmuch():
-            return 'ko'
-
-        try:
-            events.set_event_loop(self.loop)
-            t = notmuch()
-        finally:
-            events.set_event_loop(None)
-
-        self.assertIsInstance(t, tasks.Task)
-        self.loop.run_until_complete(t)
-        self.assertTrue(t.done())
-        self.assertEqual(t.result(), 'ko')
-
-    def test_task_decorator_fut(self):
-        @tasks.task
-        def notmuch():
-            fut = futures.Future(loop=self.loop)
-            fut.set_result('ko')
-            return fut
-
-        try:
-            events.set_event_loop(self.loop)
-            t = notmuch()
-        finally:
-            events.set_event_loop(None)
-
-        self.assertIsInstance(t, tasks.Task)
-        self.loop.run_until_complete(t)
-        self.assertTrue(t.done())
-        self.assertEqual(t.result(), 'ko')
-
     def test_async_coroutine(self):
         @tasks.coroutine
         def notmuch():
diff --git a/tulip/locks.py b/tulip/locks.py
index de7f915..87937ec 100644
--- a/tulip/locks.py
+++ b/tulip/locks.py
@@ -73,8 +73,10 @@ class Lock:
 
     def __repr__(self):
         res = super().__repr__()
-        return '<{} [{}]>'.format(
-            res[1:-1], 'locked' if self._locked else 'unlocked')
+        extra = 'locked' if self._locked else 'unlocked'
+        if self._waiters:
+            extra = '{},waiters:{}'.format(extra, len(self._waiters))
+        return '<{} [{}]>'.format(res[1:-1], extra)
 
     def locked(self):
         """Return true if lock is acquired."""
@@ -113,8 +115,11 @@ class Lock:
         """
         if self._locked:
             self._locked = False
-            if self._waiters:
-                self._waiters[0].set_result(True)
+            # Wake up the first waiter who isn't cancelled.
+            for fut in self._waiters:
+                if not fut.cancelled():
+                    fut.set_result(True)
+                    break
         else:
             raise RuntimeError('Lock is not acquired.')
 
@@ -132,6 +137,7 @@ class Lock:
         return self
 
 
+# TODO: Why not call this Event?
 class EventWaiter:
     """A EventWaiter implementation, our equivalent to threading.Event
 
@@ -150,6 +156,7 @@ class EventWaiter:
             self._loop = events.get_event_loop()
 
     def __repr__(self):
+        # TODO: add waiters:N if > 0.
         res = super().__repr__()
         return '<{} [{}]>'.format(res[1:-1], 'set' if self._value else 'unset')
 
@@ -195,6 +202,7 @@ class EventWaiter:
             self._waiters.remove(fut)
 
 
+# TODO: Why is this a Lock subclass?  threading.Condition *has* a lock.
 class Condition(Lock):
     """A Condition implementation.
 
@@ -205,9 +213,10 @@ class Condition(Lock):
 
     def __init__(self, *, loop=None):
         super().__init__(loop=loop)
-
         self._condition_waiters = collections.deque()
 
+    # TODO: Add __repr__() with len(_condition_waiters).
+
     @tasks.coroutine
     def wait(self):
         """Wait until notified.
@@ -233,6 +242,7 @@ class Condition(Lock):
                 return True
             finally:
                 self._condition_waiters.remove(fut)
+
         except GeneratorExit:
             keep_lock = False  # Prevent yield in finally clause.
             raise
@@ -321,6 +331,7 @@ class Semaphore:
             self._loop = events.get_event_loop()
 
     def __repr__(self):
+        # TODO: add waiters:N if > 0.
         res = super().__repr__()
         return '<{} [{}]>'.format(
             res[1:-1],
@@ -380,6 +391,8 @@ class Semaphore:
                 break
 
     def __enter__(self):
+        # TODO: This is questionable.  How do we know the user actually
+        # wrote "with (yield from sema)" instead of "with sema"?
         return True
 
     def __exit__(self, *args):
diff --git a/tulip/queues.py b/tulip/queues.py
index 244d856..4a46f1a 100644
--- a/tulip/queues.py
+++ b/tulip/queues.py
@@ -4,7 +4,6 @@ __all__ = ['Queue', 'PriorityQueue', 'LifoQueue', 'JoinableQueue',
            'Full', 'Empty']
 
 import collections
-import concurrent.futures
 import heapq
 import queue
 
@@ -129,7 +128,7 @@ class Queue:
             self._putters.append((item, waiter))
             try:
                 yield from waiter
-            except concurrent.futures.CancelledError:
+            except futures.CancelledError:
                 raise Full
 
         else:
@@ -185,7 +184,7 @@ class Queue:
             self._getters.append(waiter)
             try:
                 return (yield from waiter)
-            except concurrent.futures.CancelledError:
+            except futures.CancelledError:
                 raise Empty
 
     def get_nowait(self):
diff --git a/tulip/tasks.py b/tulip/tasks.py
index 4e3ea55..a51ee29 100644
--- a/tulip/tasks.py
+++ b/tulip/tasks.py
@@ -1,6 +1,6 @@
 """Support for tasks, coroutines and the scheduler."""
 
-__all__ = ['coroutine', 'task', 'Task',
+__all__ = ['coroutine', 'Task',
            'FIRST_COMPLETED', 'FIRST_EXCEPTION', 'ALL_COMPLETED',
            'wait', 'wait_for', 'as_completed', 'sleep', 'async',
            ]
@@ -49,23 +49,6 @@ def iscoroutine(obj):
     return inspect.isgenerator(obj)  # TODO: And what?
 
 
-def task(func):
-    """Decorator for a coroutine to be wrapped in a Task."""
-    if inspect.isgeneratorfunction(func):
-        coro = func
-    else:
-        def coro(*args, **kw):
-            res = func(*args, **kw)
-            if isinstance(res, futures.Future) or inspect.isgenerator(res):
-                res = yield from res
-            return res
-
-    def task_wrapper(*args, **kwds):
-        return Task(coro(*args, **kwds))
-
-    return task_wrapper
-
-
 class Task(futures.Future):
     """A coroutine wrapped in a Future."""