Use contextvars instead of TaskLocal.

In order to make this code work correctly
```
    def get_loop(loop):
        assert loop == asyncio.get_event_loop()

    async with trio.open_nursery():
        async with trio_asyncio.open_loop() as loop1:
            async with trio_asyncio.open_loop() as loop2:
                loop1.call_soon(get_loop, loop1)
```
the context values for the current task+policy are taken from the
currently-running task (as ``get_loop`` runs, this is the task running
``loop1``), instead of from the context of the caller, which would result
in ``loop2``.

Author: smurfix

tests/test_concurrent.py

@@ -46,27 +46,24 @@ async def gen_loop(i, task_status=trio.TASK_STATUS_IGNORED):
 
 
 async def _test_same_task():
-    loops = [None, None]
-    assert isinstance(asyncio.get_event_loop_policy(), trio_asyncio.TrioPolicy)
+    assert not isinstance(asyncio.get_event_loop_policy(), trio_asyncio.TrioPolicy)
 
-    def get_loop(i):
-        loops[i] = (asyncio.get_event_loop(), asyncio.get_event_loop_policy())
+    def get_loop(i, loop, policy):
+        assert loop == asyncio.get_event_loop()
+        assert policy == asyncio.get_event_loop_policy()
 
     async with trio.open_nursery():
         async with trio_asyncio.open_loop() as loop1:
+            policy = asyncio.get_event_loop_policy()
+            assert isinstance(policy, trio_asyncio.TrioPolicy)
             async with trio_asyncio.open_loop() as loop2:
-                loop1.call_later(0.1, get_loop, 0)
-                loop2.call_later(0.1, get_loop, 1)
+                assert policy == asyncio.get_event_loop_policy()
+                loop1.call_later(0.1, get_loop, 0, loop1, policy)
+                loop2.call_later(0.1, get_loop, 1, loop2, policy)
                 await trio.sleep(0.2)
 
-    assert isinstance(asyncio.get_event_loop_policy(), trio_asyncio.TrioPolicy)
+    assert not isinstance(asyncio.get_event_loop_policy(), trio_asyncio.TrioPolicy)
     assert not isinstance(asyncio._get_running_loop(), trio_asyncio.TrioEventLoop)
-    assert isinstance(loops[0][0], trio_asyncio.TrioEventLoop)
-    assert isinstance(loops[1][0], trio_asyncio.TrioEventLoop)
-    assert isinstance(loops[1][1], trio_asyncio.TrioPolicy)
-    assert loops[0][0] is not loops[1][0]
-    assert loops[0][1] is loops[1][1]
-    assert loops[0][1] is asyncio.get_event_loop_policy()
 
 
 def test_same_task(old_policy):

trio_asyncio/async_.py

@@ -104,12 +104,16 @@ def _main_loop_exit(self):
         super()._main_loop_exit()
         self._thread = None
 
-    from .loop import _current_loop
-    outer_loop = _current_loop.loop
+    from .loop import current_loop, current_policy, TrioPolicy
 
     async with trio.open_nursery() as nursery:
+        policy = current_policy.get() 
+        if not isinstance(policy, TrioPolicy):
+            policy = TrioPolicy()
+        old_policy = current_policy.set(policy)
+
         loop = TrioEventLoop(queue_len=queue_len)
-        _current_loop.loop = loop
+        old_loop = current_loop.set(loop)
         try:
             loop._closed = False
             await loop._main_loop_init(nursery)
@@ -122,4 +126,6 @@ def _main_loop_exit(self):
                 await loop._main_loop_exit()
                 loop.close()
                 nursery.cancel_scope.cancel()
-                _current_loop.loop = outer_loop
+                current_loop.reset(old_loop)
+                current_policy.reset(old_policy)
+

trio_asyncio/base.py

@@ -169,9 +169,9 @@ def __init__(self, queue_len=None):
 
     def __repr__(self):
         try:
-            return "<%s running=%s>" % (
+            return "<%s running=%s at 0x%x>" % (
                 self.__class__.__name__, "closed" if self._closed else "no"
-                if self._stopped.is_set() else "yes"
+                if self._stopped.is_set() else "yes", id(self)
             )
         except Exception as exc:
             return "<%s ?:%s>" % (self.__class__.__name__, repr(exc))

trio_asyncio/loop.py

@@ -6,6 +6,7 @@
 import asyncio
 import warnings
 import threading
+from contextvars import ContextVar
 
 from .util import run_future
 from .async_ import TrioEventLoop, open_loop
@@ -30,8 +31,8 @@
     'TrioPolicy',
 ]
 
-_current_loop = trio.TaskLocal(loop=None, policy=None)
-
+current_loop = ContextVar('trio_aio_loop', default=None)
+current_policy = ContextVar('trio_aio_policy', default=None)
 
 class _TrioPolicy(asyncio.events.BaseDefaultEventLoopPolicy):
     _loop_factory = TrioEventLoop
@@ -73,28 +74,23 @@ def get_event_loop(self):
             # this creates a new loop in the main task
             return super().get_event_loop()
         else:
-            return _current_loop.loop
+            return current_loop.get()
 
     @property
     def current_event_loop(self):
         """The currently-running event loop, if one exists."""
-        try:
-            return _current_loop.loop
-        except RuntimeError:
-            # in the main thread this would create a new loop
-            # return super().get_event_loop()
-            return super().get_event_loop()
+        loop = current_loop.get()
+        if loop is None:
+            loop = super().get_event_loop()
+        return loop
 
     def set_event_loop(self, loop):
         """Set the current event loop."""
-        try:
-            _current_loop.loop = loop
-        except RuntimeError:
-            return super().set_event_loop(loop)
+        current_loop.set(loop)
 
 
 # We need to monkey-patch asyncio's policy+loop getters to return our
-# TrioPolicy+loop whenever we are within Trio.
+# TrioPolicy and the current loop whenever we are within Trio.
 
 from asyncio import events as _aio_event
 
@@ -105,13 +101,13 @@ def set_event_loop(self, loop):
 
 def _new_policy_get():
     try:
-        policy = _current_loop.policy
+        task = trio.hazmat.current_task()
     except RuntimeError:
-        return _orig_policy_get()
-
+        policy = None
+    else:
+        policy = task.context[current_policy]
     if policy is None:
-        policy = TrioPolicy()
-        _current_loop.policy = policy
+        policy = _orig_policy_get()
     return policy
 
 
@@ -125,9 +121,15 @@ def _new_policy_get():
 
 def _new_run_get():
     try:
-        return _current_loop.loop
+        task = trio.hazmat.current_task()
     except RuntimeError:
-        return _orig_run_get()
+        loop = None
+    else:
+        loop = task.context[current_loop]
+
+    if loop is None:
+        loop = _orig_run_get()
+    return loop
 
 
 _aio_event._get_running_loop = _new_run_get
@@ -138,10 +140,10 @@ def _new_run_get():
 
 
 def _new_loop_get():
-    try:
-        return _current_loop.loop
-    except RuntimeError:
-        return _orig_loop_get()
+    loop = _new_run_get()
+    if loop is None:
+        loop = _orig_loop_get()
+    return loop
 
 
 _aio_event.get_event_loop = _new_loop_get
@@ -156,11 +158,11 @@ def _init_watcher(self):
             if self._watcher is None:  # pragma: no branch
                 self._watcher = TrioChildWatcher()
                 if isinstance(threading.current_thread(), threading._MainThread):
-                    self._watcher.attach_loop(_current_loop.loop)
+                    self._watcher.attach_loop(current_loop.get())
 
         if self._watcher is not None and \
                 isinstance(threading.current_thread(), threading._MainThread):
-            self._watcher.attach_loop(_current_loop.loop)
+            self._watcher.attach_loop(current_loop.get())
 
     def set_child_watcher(self, watcher):
         if watcher is not None: