Repository: agronholm/anyio

Issue:
to/from_thread and contextvars
The behavior of how contextvars are propagated through `to`/`from_thread` differs depending on which backend you use. With `asyncio`, they inherit the context of the calling thread; in `trio` they inherit the context of the system task group.
```py
import contextvars
import anyio

var = contextvars.ContextVar("var", default=None)

async def get():
	return var.get()

def anyio_from_thread():
	var.set("b")
	return anyio.from_thread.run(get)

async def anyio_to_from_thread():
	var.set("a")
	return await anyio.to_thread.run_sync(anyio_from_thread)

print("asyncio", anyio.run(anyio_to_from_thread, backend="asyncio"))
print("trio   ", anyio.run(anyio_to_from_thread, backend="trio"))
```

On anyio 3.3.0, this prints:
```
asyncio b
trio    None
```

This is easy (but inconvenient) to work around for `run_sync` by adding a `copy_context().run`, but I think any similar workaround for `run` would require special-casing for trio.

Also, `asyncio.to_thread` copies the context whereas `anyio.to_thread.run_sync` does not:
```py
import contextvars
import anyio
import asyncio

var = contextvars.ContextVar("var", default=None)

async def asyncio_to_thread():
	var.set("b")
	return await asyncio.to_thread(var.get)

async def anyio_to_thread():
	var.set("b")
	return await anyio.to_thread.run_sync(var.get)

print("asyncio", asyncio.run(asyncio_to_thread()))
print("anyio  ", anyio.run(anyio_to_thread, backend="asyncio"))
```
```
asyncio c
anyio   None
```

This is not as important since they are different functions, but it's still annoying. Same workaround as above works here, of course.

Discussion:
tiangolo: Awesome! 🚀🎉
agronholm: It might take some time for the trio folks to get these changes out in a release. As such, I see three options for us here:

1. Wait for trio
2. Merge anyway and document that this won't work properly on trio yet
3. Add (potentially clunky) workarounds where we can and merge
agronholm: @tiangolo I'm going to make a decision on this soon, but I'd like to hear your opinion first.
tiangolo: Thanks for having me into account and asking!

For me personally, options 2 or 3 would be ideal, as I want to use this in some projects, and they don't depend on Trio.

I knew it was important for AnyIO to have support in Trio (eventually, at least), and because the API design is inspired by Trio, so I started the PR there first, but I'm currently not depending on Trio itself.
agronholm: I went with option 3 so we have full functionality with either trio or asyncio@py37+, without requiring the trio PR.

Patch:
diff --git a/docs/subprocesses.rst b/docs/subprocesses.rst
index 0b38821..cb3cd82 100644
--- a/docs/subprocesses.rst
+++ b/docs/subprocesses.rst
@@ -108,7 +108,7 @@ Other considerations:
 * If a cancellable call is cancelled during execution on the worker process, the worker process
   will be killed
 * The worker process imports the parent's ``__main__`` module, so guarding for any import time side
-  effects using ``if __name__ == '__main__':`` is required to avoid inifinite recursion
+  effects using ``if __name__ == '__main__':`` is required to avoid infinite recursion
 * ``sys.stdin`` and ``sys.stdout``, ``sys.stderr`` are redirected to ``/dev/null`` so :func:`print`
   and :func:`input` won't work
 * Worker processes terminate after 5 minutes of inactivity, or when the event loop is finished
diff --git a/docs/tasks.rst b/docs/tasks.rst
index 6da0cc2..b1b695b 100644
--- a/docs/tasks.rst
+++ b/docs/tasks.rst
@@ -90,3 +90,17 @@ the task group context manager? The answer is "both". In practice this means tha
 exception, :exc:`~ExceptionGroup` is raised which contains both exception objects.
 Unfortunately this complicates any code that wishes to catch a specific exception because it could
 be wrapped in an :exc:`~ExceptionGroup`.
+
+Context propagation
+-------------------
+
+Whenever a new task is spawned, `context`_ will be copied to the new task. It is important to note
+*which* content will be copied to the newly spawned task. It is not the context of the task group's
+host task that will be copied, but the context of the task that calls
+:meth:`TaskGroup.start() <.abc.TaskGroup.start>` or
+:meth:`TaskGroup.start_soon() <.abc.TaskGroup.start_soon>`.
+
+.. note:: Context propagation **does not work** on asyncio when using Python 3.6, as asyncio
+    support for this only landed in v3.7.
+
+.. _context: https://docs.python.org/3/library/contextvars.html
diff --git a/docs/threads.rst b/docs/threads.rst
index 88defe0..a9d09a3 100644
--- a/docs/threads.rst
+++ b/docs/threads.rst
@@ -177,3 +177,15 @@ managers as a synchronous one::
 
 .. note:: You cannot use wrapped async context managers in synchronous callbacks inside the event
           loop thread.
+
+Context propagation
+-------------------
+
+When running functions in worker threads, the current context is copied to the worker thread.
+Therefore any context variables available on the task will also be available to the code running
+on the thread. As always with context variables, any changes made to them will not propagate back
+to the calling asynchronous task.
+
+When calling asynchronous code from worker threads, context is again copied to the task that calls
+the target function in the event loop thread. Note, however, that this **does not work** on asyncio
+when running on Python 3.6.
diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst
index d43a3f1..0535725 100644
--- a/docs/versionhistory.rst
+++ b/docs/versionhistory.rst
@@ -5,6 +5,12 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
 
 **UNRELEASED**
 
+- Added context propagation to/from worker threads in ``to_thread.run_sync()``,
+  ``from_thread.run()`` and ``from_thread.run_sync()``
+  (`#363 <https://github.com/agronholm/anyio/issues/363>`_; partially based on a PR by Sebastián
+  Ramírez)
+
+  **NOTE**: Requires Python 3.7 to work properly on asyncio!
 - Fixed race condition in ``Lock`` and ``Semaphore`` classes when a task waiting on ``acquire()``
   is cancelled while another task is waiting to acquire the same primitive
   (`#387 <https://github.com/agronholm/anyio/issues/387>`_)
@@ -12,6 +18,7 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
   ``BlockingPortal.wrap_async_context_manager()`` if the host task is cancelled
   (`#381 <https://github.com/agronholm/anyio/issues/381>`_; PR by Jonathan Slenders)
 - Fixed worker threads being marked as being event loop threads in sniffio
+- Fixed task parent ID not getting set to the correct value on asyncio
 
 **3.3.4**
 
diff --git a/setup.cfg b/setup.cfg
index d85f3af..65a06ec 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -28,6 +28,7 @@ packages = find:
 python_requires = >= 3.6.2
 zip_safe = False
 install_requires =
+    contextvars; python_version < '3.7'
     dataclasses; python_version < '3.7'
     idna >= 2.8
     sniffio >= 1.1
diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py
index 834dd91..82df30a 100644
--- a/src/anyio/_backends/_asyncio.py
+++ b/src/anyio/_backends/_asyncio.py
@@ -7,6 +7,7 @@ import sys
 from asyncio.base_events import _run_until_complete_cb  # type: ignore
 from collections import OrderedDict, deque
 from concurrent.futures import Future
+from contextvars import Context, copy_context
 from dataclasses import dataclass
 from functools import partial, wraps
 from inspect import (
@@ -22,6 +23,8 @@ from typing import (
     Mapping, Optional, Sequence, Set, Tuple, Type, TypeVar, Union, cast)
 from weakref import WeakKeyDictionary
 
+import sniffio
+
 from .. import CapacityLimiterStatistics, EventStatistics, TaskInfo, abc
 from .._core._compat import DeprecatedAsyncContextManager, DeprecatedAwaitable
 from .._core._eventloop import claim_worker_thread, threadlocals
@@ -515,8 +518,9 @@ class ExceptionGroup(BaseExceptionGroup):
 
 
 class _AsyncioTaskStatus(abc.TaskStatus):
-    def __init__(self, future: asyncio.Future):
+    def __init__(self, future: asyncio.Future, parent_id: int):
         self._future = future
+        self._parent_id = parent_id
 
     def started(self, value: object = None) -> None:
         try:
@@ -524,6 +528,9 @@ class _AsyncioTaskStatus(abc.TaskStatus):
         except asyncio.InvalidStateError:
             raise RuntimeError("called 'started' twice on the same task status") from None
 
+        task = cast(asyncio.Task, current_task())
+        _task_states[task].parent_id = self._parent_id
+
 
 class TaskGroup(abc.TaskGroup):
     def __init__(self) -> None:
@@ -653,7 +660,11 @@ class TaskGroup(abc.TaskGroup):
 
         kwargs = {}
         if task_status_future:
-            kwargs['task_status'] = _AsyncioTaskStatus(task_status_future)
+            parent_id = id(current_task())
+            kwargs['task_status'] = _AsyncioTaskStatus(task_status_future,
+                                                       id(self.cancel_scope._host_task))
+        else:
+            parent_id = id(self.cancel_scope._host_task)
 
         coro = func(*args, **kwargs)
         if not asyncio.iscoroutine(coro):
@@ -668,7 +679,7 @@ class TaskGroup(abc.TaskGroup):
             task.add_done_callback(task_done)
 
         # Make the spawned task inherit the task group's cancel scope
-        _task_states[task] = TaskState(parent_id=id(current_task()), name=name,
+        _task_states[task] = TaskState(parent_id=parent_id, name=name,
                                        cancel_scope=self.cancel_scope)
         self.cancel_scope._tasks.add(task)
         return task
@@ -710,7 +721,7 @@ class WorkerThread(Thread):
         self.workers = workers
         self.idle_workers = idle_workers
         self.loop = root_task._loop
-        self.queue: Queue[Union[Tuple[Callable, tuple, asyncio.Future], None]] = Queue(2)
+        self.queue: Queue[Union[Tuple[Context, Callable, tuple, asyncio.Future], None]] = Queue(2)
         self.idle_since = current_time()
         self.stopping = False
 
@@ -735,12 +746,12 @@ class WorkerThread(Thread):
                     # Shutdown command received
                     return
 
-                func, args, future = item
+                context, func, args, future = item
                 if not future.cancelled():
                     result = None
                     exception: Optional[BaseException] = None
                     try:
-                        result = func(*args)
+                        result = context.run(func, *args)
                     except BaseException as exc:
                         exception = exc
 
@@ -801,7 +812,9 @@ async def run_sync_in_worker_thread(
                     expired_worker.root_task.remove_done_callback(expired_worker.stop)
                     expired_worker.stop()
 
-            worker.queue.put_nowait((func, args, future))
+            context = copy_context()
+            context.run(sniffio.current_async_library_cvar.set, None)
+            worker.queue.put_nowait((context, func, args, future))
             return await future
 
 
@@ -818,7 +831,11 @@ def run_sync_from_thread(func: Callable[..., T_Retval], *args: object,
 
     f: concurrent.futures.Future[T_Retval] = Future()
     loop = loop or threadlocals.loop
-    loop.call_soon_threadsafe(wrapper)
+    if sys.version_info < (3, 7):
+        loop.call_soon_threadsafe(copy_context().run, wrapper)
+    else:
+        loop.call_soon_threadsafe(wrapper)
+
     return f.result()
 
 
diff --git a/src/anyio/_backends/_trio.py b/src/anyio/_backends/_trio.py
index aa4e0e2..4ce13ca 100644
--- a/src/anyio/_backends/_trio.py
+++ b/src/anyio/_backends/_trio.py
@@ -2,6 +2,7 @@ import array
 import math
 import socket
 from concurrent.futures import Future
+from contextvars import copy_context
 from dataclasses import dataclass
 from functools import partial
 from io import IOBase
@@ -11,6 +12,7 @@ from typing import (
     Any, Awaitable, Callable, Collection, ContextManager, Coroutine, Deque, Dict, Generic, List,
     Mapping, NoReturn, Optional, Sequence, Set, Tuple, Type, TypeVar, Union)
 
+import sniffio
 import trio.from_thread
 from outcome import Error, Outcome, Value
 from trio.socket import SocketType as TrioSocketType
@@ -37,6 +39,10 @@ except ImportError:
 else:
     from trio.lowlevel import wait_readable, wait_writable
 
+try:
+    from trio.lowlevel import open_process as trio_open_process
+except ImportError:
+    from trio import open_process as trio_open_process
 
 T_Retval = TypeVar('T_Retval')
 T_SockAddr = TypeVar('T_SockAddr', str, IPSockAddrType)
@@ -167,10 +173,36 @@ async def run_sync_in_worker_thread(
         with claim_worker_thread('trio'):
             return func(*args)
 
-    return await run_sync(wrapper, cancellable=cancellable, limiter=limiter)
+    # TODO: remove explicit context copying when trio 0.20 is the minimum requirement
+    context = copy_context()
+    context.run(sniffio.current_async_library_cvar.set, None)
+    return await run_sync(context.run, wrapper, cancellable=cancellable, limiter=limiter)
+
+
+# TODO: remove this workaround when trio 0.20 is the minimum requirement
+def run_async_from_thread(fn: Callable[..., Awaitable[T_Retval]], *args: Any) -> T_Retval:
+    async def wrapper() -> Optional[T_Retval]:
+        retval: T_Retval
+
+        async def inner() -> None:
+            nonlocal retval
+            __tracebackhide__ = True
+            retval = await fn(*args)
+
+        async with trio.open_nursery() as n:
+            context.run(n.start_soon, inner)
+
+        __tracebackhide__ = True
+        return retval
+
+    context = copy_context()
+    context.run(sniffio.current_async_library_cvar.set, 'trio')
+    return trio.from_thread.run(wrapper)
+
 
-run_async_from_thread = trio.from_thread.run
-run_sync_from_thread = trio.from_thread.run_sync
+def run_sync_from_thread(fn: Callable[..., T_Retval], *args: Any) -> T_Retval:
+    # TODO: remove explicit context copying when trio 0.20 is the minimum requirement
+    return trio.from_thread.run_sync(copy_context().run, fn, *args)
 
 
 class BlockingPortal(abc.BlockingPortal):
@@ -183,9 +215,11 @@ class BlockingPortal(abc.BlockingPortal):
 
     def _spawn_task_from_thread(self, func: Callable, args: tuple, kwargs: Dict[str, Any],
                                 name: object, future: Future) -> None:
+        context = copy_context()
+        context.run(sniffio.current_async_library_cvar.set, 'trio')
         return trio.from_thread.run_sync(
-            partial(self._task_group.start_soon, name=name), self._call_func, func, args, kwargs,
-            future, trio_token=self._token)
+            context.run, partial(self._task_group.start_soon, name=name), self._call_func, func,
+            args, kwargs, future, trio_token=self._token)
 
 
 #
@@ -283,7 +317,7 @@ async def open_process(command: Union[str, Sequence[str]], *, shell: bool,
                        stdin: int, stdout: int, stderr: int,
                        cwd: Union[str, bytes, PathLike, None] = None,
                        env: Optional[Mapping[str, str]] = None) -> Process:
-    process = await trio.open_process(command, stdin=stdin, stdout=stdout, stderr=stderr,
+    process = await trio_open_process(command, stdin=stdin, stdout=stdout, stderr=stderr,
                                       shell=shell, cwd=cwd, env=env)
     stdin_stream = SendStreamWrapper(process.stdin) if process.stdin else None
     stdout_stream = ReceiveStreamWrapper(process.stdout) if process.stdout else None
diff --git a/src/anyio/_core/_exceptions.py b/src/anyio/_core/_exceptions.py
index f7d020c..418c003 100644
--- a/src/anyio/_core/_exceptions.py
+++ b/src/anyio/_core/_exceptions.py
@@ -4,7 +4,7 @@ from typing import Sequence
 
 class BrokenResourceError(Exception):
     """
-    Raised when trying to use a resource that has been rendered unusuable due to external causes
+    Raised when trying to use a resource that has been rendered unusable due to external causes
     (e.g. a send stream whose peer has disconnected).
     """
 


Test patch:
diff --git a/tests/test_from_thread.py b/tests/test_from_thread.py
index 017abf5..4a61edf 100644
--- a/tests/test_from_thread.py
+++ b/tests/test_from_thread.py
@@ -3,6 +3,7 @@ import threading
 import time
 from concurrent.futures import CancelledError
 from contextlib import suppress
+from contextvars import ContextVar
 from typing import Any, AsyncGenerator, Dict, List, NoReturn, Optional
 
 import pytest
@@ -13,6 +14,7 @@ from anyio import (
     to_thread, wait_all_tasks_blocked)
 from anyio.abc import TaskStatus
 from anyio.from_thread import BlockingPortal, start_blocking_portal
+from anyio.lowlevel import checkpoint
 
 if sys.version_info >= (3, 8):
     from typing import Literal
@@ -122,6 +124,22 @@ class TestRunAsyncFromThread:
         exc = pytest.raises(RuntimeError, from_thread.run, foo)
         exc.match('This function can only be run from an AnyIO worker thread')
 
+    async def test_contextvar_propagation(self, anyio_backend_name: str) -> None:
+        if anyio_backend_name == 'asyncio' and sys.version_info < (3, 7):
+            pytest.skip('Asyncio does not propagate context before Python 3.7')
+
+        var = ContextVar('var', default=1)
+
+        async def async_func() -> int:
+            await checkpoint()
+            return var.get()
+
+        def worker() -> int:
+            var.set(6)
+            return from_thread.run(async_func)
+
+        assert await to_thread.run_sync(worker) == 6
+
 
 class TestRunSyncFromThread:
     def test_run_sync_from_unclaimed_thread(self) -> None:
@@ -131,6 +149,15 @@ class TestRunSyncFromThread:
         exc = pytest.raises(RuntimeError, from_thread.run_sync, foo)
         exc.match('This function can only be run from an AnyIO worker thread')
 
+    async def test_contextvar_propagation(self) -> None:
+        var = ContextVar('var', default=1)
+
+        def worker() -> int:
+            var.set(6)
+            return from_thread.run_sync(var.get)
+
+        assert await to_thread.run_sync(worker) == 6
+
 
 class TestBlockingPortal:
     class AsyncCM:
@@ -400,6 +427,35 @@ class TestBlockingPortal:
                 taskfunc, name='testname')  # type: ignore[arg-type]
             assert start_value == 'testname'
 
+    def test_contextvar_propagation_sync(self, anyio_backend_name: str,
+                                         anyio_backend_options: Dict[str, Any]) -> None:
+        if anyio_backend_name == 'asyncio' and sys.version_info < (3, 7):
+            pytest.skip('Asyncio does not propagate context before Python 3.7')
+
+        var = ContextVar('var', default=1)
+        var.set(6)
+        with start_blocking_portal(anyio_backend_name, anyio_backend_options) as portal:
+            propagated_value = portal.call(var.get)
+
+        assert propagated_value == 6
+
+    def test_contextvar_propagation_async(self, anyio_backend_name: str,
+                                          anyio_backend_options: Dict[str, Any]) -> None:
+        if anyio_backend_name == 'asyncio' and sys.version_info < (3, 7):
+            pytest.skip('Asyncio does not propagate context before Python 3.7')
+
+        var = ContextVar('var', default=1)
+        var.set(6)
+
+        async def get_var() -> int:
+            await checkpoint()
+            return var.get()
+
+        with start_blocking_portal(anyio_backend_name, anyio_backend_options) as portal:
+            propagated_value = portal.call(get_var)
+
+        assert propagated_value == 6
+
     @pytest.mark.parametrize('anyio_backend', ['asyncio'])
     async def test_asyncio_run_sync_called(self, caplog: LogCaptureFixture) -> None:
         """Regression test for #357."""
diff --git a/tests/test_taskgroups.py b/tests/test_taskgroups.py
index 1980470..fd60344 100644
--- a/tests/test_taskgroups.py
+++ b/tests/test_taskgroups.py
@@ -2,7 +2,7 @@ import asyncio
 import re
 import sys
 import time
-from typing import Any, AsyncGenerator, Coroutine, Dict, Generator, NoReturn, Set
+from typing import Any, AsyncGenerator, Coroutine, Dict, Generator, NoReturn, Optional, Set
 
 import pytest
 import trio
@@ -10,7 +10,8 @@ import trio
 import anyio
 from anyio import (
     CancelScope, ExceptionGroup, create_task_group, current_effective_deadline, current_time,
-    fail_after, get_cancelled_exc_class, move_on_after, sleep, wait_all_tasks_blocked)
+    fail_after, get_cancelled_exc_class, get_current_task, move_on_after, sleep,
+    wait_all_tasks_blocked)
 from anyio.abc import TaskGroup, TaskStatus
 from anyio.lowlevel import checkpoint
 
@@ -1004,3 +1005,45 @@ async def test_cancellederror_combination_with_message() -> None:
             await wait_all_tasks_blocked()
             assert isinstance(task, asyncio.Task)
             task.cancel('blah')
+
+
+async def test_start_soon_parent_id() -> None:
+    root_task_id = get_current_task().id
+    parent_id: Optional[int] = None
+
+    async def subtask() -> None:
+        nonlocal parent_id
+        parent_id = get_current_task().parent_id
+
+    async def starter_task() -> None:
+        tg.start_soon(subtask)
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(starter_task)
+
+    assert parent_id == root_task_id
+
+
+async def test_start_parent_id() -> None:
+    root_task_id = get_current_task().id
+    starter_task_id: Optional[int] = None
+    initial_parent_id: Optional[int] = None
+    permanent_parent_id: Optional[int] = None
+
+    async def subtask(*, task_status: TaskStatus) -> None:
+        nonlocal initial_parent_id, permanent_parent_id
+        initial_parent_id = get_current_task().parent_id
+        task_status.started()
+        permanent_parent_id = get_current_task().parent_id
+
+    async def starter_task() -> None:
+        nonlocal starter_task_id
+        starter_task_id = get_current_task().id
+        await tg.start(subtask)
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(starter_task)
+
+    assert initial_parent_id != permanent_parent_id
+    assert initial_parent_id == starter_task_id
+    assert permanent_parent_id == root_task_id
diff --git a/tests/test_to_thread.py b/tests/test_to_thread.py
index b17d230..5d0a31c 100644
--- a/tests/test_to_thread.py
+++ b/tests/test_to_thread.py
@@ -3,10 +3,12 @@ import sys
 import threading
 import time
 from concurrent.futures import Future
+from contextvars import ContextVar
 from functools import partial
 from typing import Any, List, NoReturn, Optional
 
 import pytest
+import sniffio
 
 import anyio.to_thread
 from anyio import (
@@ -132,6 +134,17 @@ async def test_cancel_wait_on_thread() -> None:
     assert future.result(1)
 
 
+async def test_contextvar_propagation() -> None:
+    var = ContextVar('var', default=1)
+    var.set(6)
+    assert await to_thread.run_sync(var.get) == 6
+
+
+async def test_asynclib_detection() -> None:
+    with pytest.raises(sniffio.AsyncLibraryNotFoundError):
+        await to_thread.run_sync(sniffio.current_async_library)
+
+
 @pytest.mark.parametrize('anyio_backend', ['asyncio'])
 async def test_asyncio_cancel_native_task() -> None:
     task: "Optional[asyncio.Task[None]]" = None