Repository: ReactiveX/RxPY

Issue:
Deadlock in AsyncIOThreadSafeScheduler
Fallowing code produces deadlock:

```python
import rx
import rx.operators as ops
from rx.subject.subject import Subject
from rx.scheduler.eventloop import AsyncIOThreadSafeScheduler
import asyncio
import logging

def print_threads_traces(_):
    import traceback
    print("\n*** STACKTRACE - START ***\n")
    code = []
    for threadId, stack in sys._current_frames().items():
        code.append("\n# ThreadID: %s" % threadId)
        for filename, lineno, name, line in traceback.extract_stack(stack):
            code.append('File: "%s", line %d, in %s' % (filename,
                                                        lineno, name))
            if line:
                code.append("  %s" % (line.strip()))

    for line in code:
        print(line)
    print("\n*** STACKTRACE - END ***\n")

logging.basicConfig(
    level=logging.DEBUG,
    format='[%(threadName)s|%(levelname).5s|%(name).32s:%(lineno)d] %(message)s'
)

logger = logging.getLogger('repro')

loop = asyncio.get_event_loop()
scheduler = AsyncIOThreadSafeScheduler(loop)

src = Subject()

def log_and_push_another_one(msg):
    logger.info(msg)
    src.on_next(msg + "0")
    logger.info('Done')

# Two delays are needed to break the cycle of action self-rescheduling in delay.py.
# That will cause setting cancelable.disposable = mad, which will cause previous
# disposable to be disposed, which will cause scheduler to wait for confirmation,
# which causes deadlock.
obs = src.pipe(ops.delay(1.0, scheduler=scheduler), ops.delay(1.0, scheduler=scheduler))
obs.subscribe(log_and_push_another_one)

# After few seconds print stack trace of each thread, in another thread.
rx.of(1).pipe(ops.delay(6.0)).subscribe(print_threads_traces)

loop.call_soon(src.on_next, "Message0")

loop.run_forever()

```

This is where thread waits:

```
File: "main.py", line 58, in <module>
  loop.run_forever()
File: "/usr/lib/python3.6/asyncio/base_events.py", line 438, in run_forever
  self._run_once()
File: "/usr/lib/python3.6/asyncio/base_events.py", line 1451, in _run_once
  handle._run()
File: "/usr/lib/python3.6/asyncio/events.py", line 145, in _run
  self._callback(*self._args)
File: "RxPY/rx/scheduler/eventloop/asynciothreadsafescheduler.py", line 87, in interval
  sad.disposable = self.invoke_action(action, state=state)
File: "RxPY/rx/scheduler/scheduler.py", line 103, in invoke_action
  ret = action(self, state)
File: "RxPY/rx/core/operators/delay.py", line 69, in action
  result.accept(observer)
File: "RxPY/rx/core/notification.py", line 37, in accept
  return self._accept_observer(on_next)
File: "RxPY/rx/core/notification.py", line 99, in _accept_observer
  return observer.on_next(self.value)
File: "RxPY/rx/core/observer/autodetachobserver.py", line 26, in on_next
  self._on_next(value)
File: "mamicu.py", line 43, in log_and_push_another_one
  src.on_next(msg + "0")
File: "RxPY/rx/subject/subject.py", line 55, in on_next
  super().on_next(value)
File: "RxPY/rx/core/observer/observer.py", line 26, in on_next
  self._on_next_core(value)
File: "RxPY/rx/subject/subject.py", line 62, in _on_next_core
  observer.on_next(value)
File: "RxPY/rx/core/observer/autodetachobserver.py", line 26, in on_next
  self._on_next(value)
File: "RxPY/rx/core/operators/materialize.py", line 23, in on_next
  observer.on_next(OnNext(value))
File: "RxPY/rx/core/observer/autodetachobserver.py", line 26, in on_next
  self._on_next(value)
File: "RxPY/rx/core/operators/map.py", line 41, in on_next
  obv.on_next(result)
File: "RxPY/rx/core/observer/autodetachobserver.py", line 26, in on_next
  self._on_next(value)
File: "RxPY/rx/core/observer/autodetachobserver.py", line 26, in on_next
  self._on_next(value)
File: "RxPY/rx/core/operators/delay.py", line 55, in on_next
  cancelable.disposable = mad
File: "RxPY/rx/disposable/serialdisposable.py", line 39, in set_disposable
  old.dispose()
File: "RxPY/rx/disposable/multipleassignmentdisposable.py", line 47, in dispose
  old.dispose()
File: "RxPY/rx/disposable/compositedisposable.py", line 68, in dispose
  disp.dispose()
File: "RxPY/rx/disposable/disposable.py", line 41, in dispose
  self.action()
File: "RxPY/rx/scheduler/eventloop/asynciothreadsafescheduler.py", line 110, in dispose
  future.result()
File: "/usr/lib/python3.6/concurrent/futures/_base.py", line 427, in result
  self._condition.wait(timeout)
File: "/usr/lib/python3.6/threading.py", line 295, in wait
  waiter.acquire()
```

It does not have to be two delays in a row, it can be observe_on after delay, or anything that prevents next delay action to be rescheduled immediately (see comment in code example).

It may seem to you that it is not real world use case to push item into observable from the same thread and asyncio loop as observer, but it is really important sometimes to be able to do everything in one thread, including pushing new items sometimes.

It may seem to you that we should use basic  AsyncIOScheduler, but if sometimes we still need to do something from other threads, it becomes impossible, I have tried.

To me it seems that scheduler shouldn't really wait for that future, because:
1. Canceling in asyncio is best effort anyway.
2. Other scheduler don't seem to behave like that.
3. If I remove future, all tests pass anyway.


Discussion:
coveralls: 
[![Coverage Status](https://coveralls.io/builds/38512726/badge)](https://coveralls.io/builds/38512726)

Coverage increased (+0.01%) to 93.498% when pulling **e9c802b98d866ccb1dc3c75c0a82ceb67c760daa on kormang:deadlock** into **1bd44fe68183dcaad1c2a243c2fc4a255e986fbd on ReactiveX:master**.

kormang: @erikkemperman You wrote this code, can you please tell me did you do it on purpose or it just seemed right thing to do?
Is this OK?
Can you, please, review?

Thanks.
erikkemperman: @kormang I did not actually write this code, although it seems that way on github history because I moved this file in a new directory. You might be right about the deadlock, but looks to me like it was a deliberate choice to make the cancel call block until the future is fulfilled. I'm not entirely sure if anything would break if that is removed. @dbrattli Could you take a look at this PR?
kormang: OK, wait then. I'll try to detect if we're running on the same loop using https://docs.python.org/3/library/asyncio-eventloop.html#asyncio.get_running_loop, and in such case  call cancel immediately just like AsyncIOScheduler. Not sure if that is possible, but I'll try later today.
kormang: I've made it wait for future, when possible. Necessary conditions to be able to wait without deadlock are, `loop_` is running, and it belongs to another thread.
MainRo: @kormang I just commented on #566 

Can you confirm that your need is to cancel subscriptions that are distributed between several asyncio event loops?
In such a case, I agree we must remove the wait for the future (This was not the intended usage of the ThreadsafeScheduler). However I would expect some strange behaviors due to the fact that the disposal becomes asynchronous: You may have some pending tasks that will execute after the call to "dispose" returns.

More generally, RxPy provides some integration with Asyncio, but I would not try to split observable chains between several event loops: The RxPY APIs are all synchronous, and there are some situations there going back and forth between RxPY functions and asyncio couroutines is not possible.
 


kormang: @MainRo I'll continue discussion there. Let here be the place to comment this particular PR.  

Patch:
diff --git a/rx/scheduler/eventloop/asynciothreadsafescheduler.py b/rx/scheduler/eventloop/asynciothreadsafescheduler.py
index 52f6e30d..79880d41 100644
--- a/rx/scheduler/eventloop/asynciothreadsafescheduler.py
+++ b/rx/scheduler/eventloop/asynciothreadsafescheduler.py
@@ -50,6 +50,10 @@ class AsyncIOThreadSafeScheduler(AsyncIOScheduler):
         handle = self._loop.call_soon_threadsafe(interval)
 
         def dispose() -> None:
+            if self._on_self_loop_or_not_running():
+                handle.cancel()
+                return
+
             future: Future = Future()
 
             def cancel_handle() -> None:
@@ -96,14 +100,21 @@ class AsyncIOThreadSafeScheduler(AsyncIOScheduler):
         handle.append(self._loop.call_soon_threadsafe(stage2))
 
         def dispose() -> None:
-            future: Future = Future()
-
-            def cancel_handle() -> None:
+            def do_cancel_handles():
                 try:
                     handle.pop().cancel()
                     handle.pop().cancel()
                 except Exception:
                     pass
+
+            if self._on_self_loop_or_not_running():
+                do_cancel_handles()
+                return
+
+            future: Future = Future()
+
+            def cancel_handle() -> None:
+                do_cancel_handles()
                 future.set_result(0)
 
             self._loop.call_soon_threadsafe(cancel_handle)
@@ -130,3 +141,22 @@ class AsyncIOThreadSafeScheduler(AsyncIOScheduler):
 
         duetime = self.to_datetime(duetime)
         return self.schedule_relative(duetime - self.now, action, state=state)
+
+    def _on_self_loop_or_not_running(self):
+        """
+            Returns True if either self._loop is not running, or we're currently
+            executing on self._loop. In both cases, waiting for a future to be
+            resolved on the loop would result in a deadlock.
+        """
+        if not self._loop.is_running():
+            return True
+        current_loop = None
+        try:
+            # In python 3.7 there asyncio.get_running_loop() is prefered.
+            current_loop = asyncio.get_event_loop()
+        except RuntimeError:
+            # If there is no loop in current thread at all, and it is not main
+            # thread, we get error like:
+            # RuntimeError: There is no current event loop in thread 'Thread-1'
+            pass
+        return self._loop == current_loop


Test patch:
diff --git a/tests/test_scheduler/test_eventloop/test_asynciothreadsafescheduler.py b/tests/test_scheduler/test_eventloop/test_asynciothreadsafescheduler.py
index 849b0d3e..e09af2ca 100644
--- a/tests/test_scheduler/test_eventloop/test_asynciothreadsafescheduler.py
+++ b/tests/test_scheduler/test_eventloop/test_asynciothreadsafescheduler.py
@@ -92,3 +92,71 @@ class TestAsyncIOThreadSafeScheduler(unittest.TestCase):
             assert ran is False
 
         loop.run_until_complete(go())
+
+    def cancel_same_thread_common(self, test_body):
+        update_state = {
+            'ran': False,
+            'dispose_completed': False
+        }
+
+        def action(scheduler, state):
+            update_state['ran'] = True
+
+        # Make the actual test body run in deamon thread, so that in case of
+        # failure it doesn't hang indefinitely.
+        def thread_target():
+            loop = asyncio.new_event_loop()
+            scheduler = AsyncIOThreadSafeScheduler(loop)
+
+            test_body(scheduler, action, update_state)
+
+            @asyncio.coroutine
+            def go():
+                yield from asyncio.sleep(0.2, loop=loop)
+
+            loop.run_until_complete(go())
+
+        thread = threading.Thread(target=thread_target)
+        thread.daemon = True
+        thread.start()
+        thread.join(0.3)
+        assert update_state['dispose_completed'] is True
+        assert update_state['ran'] is False
+
+
+    def test_asyncio_threadsafe_cancel_non_relative_same_thread(self):
+        def test_body(scheduler, action, update_state):
+            d = scheduler.schedule(action)
+
+            # Test case when dispose is called on thread on which loop is not
+            # yet running, and non-relative schedele is used.
+            d.dispose()
+            update_state['dispose_completed'] = True
+
+        self.cancel_same_thread_common(test_body)
+
+
+    def test_asyncio_threadsafe_schedule_action_cancel_same_thread(self):
+        def test_body(scheduler, action, update_state):
+            d = scheduler.schedule_relative(0.05, action)
+
+            # Test case when dispose is called on thread on which loop is not
+            # yet running, and relative schedule is used.
+            d.dispose()
+            update_state['dispose_completed'] = True
+
+        self.cancel_same_thread_common(test_body)
+
+
+    def test_asyncio_threadsafe_schedule_action_cancel_same_loop(self):
+        def test_body(scheduler, action, update_state):
+            d = scheduler.schedule_relative(0.1, action)
+
+            def do_dispose():
+                d.dispose()
+                update_state['dispose_completed'] = True
+
+            # Test case when dispose is called in loop's callback.
+            scheduler._loop.call_soon(do_dispose)
+
+        self.cancel_same_thread_common(test_body)