Repository: agronholm/anyio

Issue:
Move worker thread/process APIs to trio-like module names
Trio has the modules `trio.to_thread` and `trio.from_thread` where you can find the corresponding functions. It would be nice for AnyIO to follow this example too. For worker processes, `anyio.to_process` would be the logical choice.

Discussion:
richardsheridan: Maybe not! I think a one-sentence note in the docs would cover it.

On Mon, Mar 29, 2021 at 6:20 PM Alex Grönholm ***@***.***>
wrote:

> ***@***.**** commented on this pull request.
> ------------------------------
>
> In tests/test_to_process.py
> <https://github.com/agronholm/anyio/pull/249#discussion_r603650971>:
>
> > ***@***.***(autouse=True)
> +def check_compatibility(anyio_backend_name):
> +    if anyio_backend_name == 'asyncio':
> +        if platform.system() == 'Windows' and sys.version_info < (3, 8):
> +            pytest.skip('Python < 3.8 uses SelectorEventLoop by default and it does not support '
> +                        'subprocesses')
>
> Asyncio raises a NotImplementedError. I wonder if we need a better error
> message. Maybe we'll cross that bridge when we come to it? It may be that
> nobody even cares :)
>
> —
> You are receiving this because you commented.
> Reply to this email directly, view it on GitHub
> <https://github.com/agronholm/anyio/pull/249#discussion_r603650971>, or
> unsubscribe
> <https://github.com/notifications/unsubscribe-auth/AB5BHRWS2357XBGNUPISBDLTGEDLLANCNFSM42AOYHOQ>
> .
>


Patch:
diff --git a/docs/api.rst b/docs/api.rst
index 5d74c19..8260bd7 100644
--- a/docs/api.rst
+++ b/docs/api.rst
@@ -43,15 +43,25 @@ Task groups
 .. autoclass:: anyio.abc.TaskGroup
 .. autoclass:: anyio.abc.TaskStatus
 
-Threads
--------
-
-.. autofunction:: anyio.run_sync_in_worker_thread
-.. autofunction:: anyio.run_async_from_thread
-.. autofunction:: anyio.run_sync_from_thread
-.. autofunction:: anyio.current_default_worker_thread_limiter
-.. autofunction:: anyio.create_blocking_portal
-.. autofunction:: anyio.start_blocking_portal
+Running code in worker threads
+------------------------------
+
+.. autofunction:: anyio.to_thread.run_sync
+.. autofunction:: anyio.to_thread.current_default_thread_limiter
+
+Running code in worker processes
+--------------------------------
+
+.. autofunction:: anyio.to_process.run_sync
+.. autofunction:: anyio.to_process.current_default_process_limiter
+
+Running asynchronous code from other threads
+--------------------------------------------
+
+.. autofunction:: anyio.from_thread.run
+.. autofunction:: anyio.from_thread.run_sync
+.. autofunction:: anyio.from_thread.create_blocking_portal
+.. autofunction:: anyio.from_thread.start_blocking_portal
 
 .. autoclass:: anyio.abc.BlockingPortal
 
diff --git a/docs/basics.rst b/docs/basics.rst
index 46de364..7121444 100644
--- a/docs/basics.rst
+++ b/docs/basics.rst
@@ -83,8 +83,7 @@ of your choice. There are a few rules to keep in mind however:
   use a library written for trio together with a library written for asyncio.
 * Tasks spawned by these "native" libraries on backends other than trio_ are not subject to the
   cancellation rules enforced by AnyIO
-* Threads spawned outside of AnyIO cannot use :func:`~run_async_from_thread` to call
-  asynchronous code
+* Threads spawned outside of AnyIO cannot use :func:`.from_thread.run` to call asynchronous code
 
 .. _virtualenv: https://docs.python-guide.org/dev/virtualenvs/
 .. _trio: https://github.com/python-trio/trio
diff --git a/docs/migration.rst b/docs/migration.rst
index 537c8b3..aeedd47 100644
--- a/docs/migration.rst
+++ b/docs/migration.rst
@@ -110,3 +110,16 @@ or, if you need to work with both AnyIO 2 and 3::
 
     async def foo() -> Event:
         return create_event()
+
+Threading functions moved
+-------------------------
+
+Threading functions were restructured to submodules, following the example of trio:
+
+* ``current_default_worker_thread_limiter`` → :func:`.to_thread.current_default_thread_limiter`
+  (NOTE: the function was renamed too!)
+* ``run_sync_in_worker_thread()`` → :func:`.to_thread.run_sync`
+* ``run_async_from_thread()`` → :func:`.from_thread.run`
+* ``run_sync_from_thread()`` → :func:`.from_thread.run_sync`
+
+The old versions are still in place but emit deprecation warnings when called.
diff --git a/docs/subprocesses.rst b/docs/subprocesses.rst
index c71e1fe..34d9529 100644
--- a/docs/subprocesses.rst
+++ b/docs/subprocesses.rst
@@ -11,6 +11,10 @@ You can either give the command as a string, in which case it is passed to your
 (``shell=False``) in which the executable is the first item in the sequence and the rest are
 arguments passed to it.
 
+.. note:: On Windows and Python 3.7 and earlier, asyncio uses :class:`~asyncio.SelectorEventLoop`
+    by default which does not support subprocesses. It is recommended to upgrade to at least Python
+    3.8 to overcome this limitation.
+
 Running one-shot commands
 -------------------------
 
@@ -74,7 +78,7 @@ This is done by using :func:`.run_sync_in_process`::
 
     import time
 
-    from anyio import run, run_sync_in_process
+    from anyio import run, to_process
 
 
     def cpu_intensive_function(arg1, arg2):
@@ -82,7 +86,7 @@ This is done by using :func:`.run_sync_in_process`::
         return arg1 + arg2
 
     async def main():
-        result = await run_sync_in_process(cpu_intensive_function, 'Hello, ', 'world!')
+        result = await to_process.run_sync(cpu_intensive_function, 'Hello, ', 'world!')
         print(result)
 
     # This check is important when the application uses run_sync_in_process()
diff --git a/docs/threads.rst b/docs/threads.rst
index 1367be5..ab5a5cb 100644
--- a/docs/threads.rst
+++ b/docs/threads.rst
@@ -19,11 +19,11 @@ To run a (synchronous) callable in a worker thread::
 
     import time
 
-    from anyio import run_sync_in_worker_thread, run
+    from anyio import to_thread, run
 
 
     async def main():
-        await run_sync_in_worker_thread(time.sleep, 5)
+        await to_thread.run_sync(time.sleep, 5)
 
     run(main)
 
@@ -38,15 +38,15 @@ Calling asynchronous code from a worker thread
 
 If you need to call a coroutine function from a worker thread, you can do this::
 
-    from anyio import run_async_from_thread, sleep, run_sync_in_worker_thread, run
+    from anyio import from_thread, sleep, to_thread, run
 
 
     def blocking_function():
-        run_async_from_thread(sleep, 5)
+        from_thread.run(sleep, 5)
 
 
     async def main():
-        await run_sync_in_worker_thread(blocking_function)
+        await to_thread.run_sync(blocking_function)
 
     run(main)
 
diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst
index 42ba6d4..5d04d68 100644
--- a/docs/versionhistory.rst
+++ b/docs/versionhistory.rst
@@ -3,6 +3,15 @@ Version history
 
 This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
 
+**UNRELEASED**
+
+- The following functions and methods have been moved:
+
+  * ``current_default_worker_thread_limiter`` → ``anyio.to_thread.current_default_thread_limiter``
+  * ``run_sync_in_worker_thread()`` → ``anyio.to_thread.run_sync()``
+  * ``run_async_from_thread()`` → ``anyio.from_thread.run()``
+  * ``run_sync_from_thread()`` → ``anyio.from_thread.run_sync()``
+
 **3.0.0rc2**
 
 - Changed the asyncio ``CancelScope`` to raise a ``RuntimeError`` if cancel scopes are being exited
diff --git a/src/anyio/__init__.py b/src/anyio/__init__.py
index 344529b..4530c58 100644
--- a/src/anyio/__init__.py
+++ b/src/anyio/__init__.py
@@ -6,17 +6,16 @@ __all__ = ('maybe_async', 'maybe_async_cm', 'run', 'sleep', 'current_time', 'get
            'connect_unix', 'create_tcp_listener', 'create_unix_listener', 'create_udp_socket',
            'create_connected_udp_socket', 'getaddrinfo', 'getnameinfo', 'wait_socket_readable',
            'wait_socket_writable', 'create_memory_object_stream', 'run_process', 'open_process',
-           'current_default_worker_process_limiter', 'run_sync_in_process', 'create_lock',
-           'CapacityLimiter', 'CapacityLimiterStatistics', 'Condition', 'ConditionStatistics',
-           'Event', 'EventStatistics', 'Lock', 'LockStatistics', 'Semaphore',
-           'SemaphoreStatistics', 'create_condition', 'create_event', 'create_semaphore',
-           'create_capacity_limiter', 'open_cancel_scope', 'fail_after', 'move_on_after',
-           'current_effective_deadline', 'TASK_STATUS_IGNORED', 'CancelScope', 'create_task_group',
-           'TaskInfo', 'get_current_task', 'get_running_tasks', 'wait_all_tasks_blocked',
-           'run_sync_in_worker_thread', 'run_async_from_thread', 'run_sync_from_thread',
-           'current_default_worker_thread_limiter', 'create_blocking_portal',
-           'start_blocking_portal', 'typed_attribute', 'TypedAttributeSet',
-           'TypedAttributeProvider')
+           'create_lock', 'CapacityLimiter', 'CapacityLimiterStatistics', 'Condition',
+           'ConditionStatistics', 'Event', 'EventStatistics', 'Lock', 'LockStatistics',
+           'Semaphore', 'SemaphoreStatistics', 'create_condition', 'create_event',
+           'create_semaphore', 'create_capacity_limiter', 'open_cancel_scope', 'fail_after',
+           'move_on_after', 'current_effective_deadline', 'TASK_STATUS_IGNORED', 'CancelScope',
+           'create_task_group', 'TaskInfo', 'get_current_task', 'get_running_tasks',
+           'wait_all_tasks_blocked', 'run_sync_in_worker_thread', 'run_async_from_thread',
+           'run_sync_from_thread', 'current_default_worker_thread_limiter',
+           'create_blocking_portal', 'start_blocking_portal', 'typed_attribute',
+           'TypedAttributeSet', 'TypedAttributeProvider')
 
 from ._core._compat import maybe_async, maybe_async_cm
 from ._core._eventloop import current_time, get_all_backends, get_cancelled_exc_class, run, sleep
@@ -31,8 +30,7 @@ from ._core._sockets import (
     connect_tcp, connect_unix, create_connected_udp_socket, create_tcp_listener, create_udp_socket,
     create_unix_listener, getaddrinfo, getnameinfo, wait_socket_readable, wait_socket_writable)
 from ._core._streams import create_memory_object_stream
-from ._core._subprocesses import (
-    current_default_worker_process_limiter, open_process, run_process, run_sync_in_process)
+from ._core._subprocesses import open_process, run_process
 from ._core._synchronization import (
     CapacityLimiter, CapacityLimiterStatistics, Condition, ConditionStatistics, Event,
     EventStatistics, Lock, LockStatistics, Semaphore, SemaphoreStatistics, create_capacity_limiter,
@@ -41,11 +39,14 @@ from ._core._tasks import (
     TASK_STATUS_IGNORED, CancelScope, create_task_group, current_effective_deadline, fail_after,
     move_on_after, open_cancel_scope)
 from ._core._testing import TaskInfo, get_current_task, get_running_tasks, wait_all_tasks_blocked
-from ._core._threads import (
-    create_blocking_portal, current_default_worker_thread_limiter, run_async_from_thread,
-    run_sync_from_thread, run_sync_in_worker_thread, start_blocking_portal)
 from ._core._typedattr import TypedAttributeProvider, TypedAttributeSet, typed_attribute
 
+# Re-exported here, for backwards compatibility
+# isort: off
+from .to_thread import current_default_worker_thread_limiter, run_sync_in_worker_thread
+from .from_thread import (
+    create_blocking_portal, run_async_from_thread, run_sync_from_thread, start_blocking_portal)
+
 # Re-export imports so they look like they live directly in this package
 for key, value in list(locals().items()):
     if getattr(value, '__module__', '').startswith('anyio.'):
diff --git a/src/anyio/_backends/_trio.py b/src/anyio/_backends/_trio.py
index df5d9e1..737e44d 100644
--- a/src/anyio/_backends/_trio.py
+++ b/src/anyio/_backends/_trio.py
@@ -651,8 +651,16 @@ class CapacityLimiter(BaseCapacityLimiter):
         return self.__original.statistics()
 
 
-def current_default_thread_limiter():
-    return CapacityLimiter(original=trio.to_thread.current_default_thread_limiter())
+_capacity_limiter_wrapper = RunVar('_capacity_limiter_wrapper')
+
+
+def current_default_thread_limiter() -> CapacityLimiter:
+    try:
+        return _capacity_limiter_wrapper.get()
+    except LookupError:
+        limiter = CapacityLimiter(original=trio.to_thread.current_default_thread_limiter())
+        _capacity_limiter_wrapper.set(limiter)
+        return limiter
 
 
 #
diff --git a/src/anyio/_core/_fileio.py b/src/anyio/_core/_fileio.py
index 7e0f0ac..fe2b53d 100644
--- a/src/anyio/_core/_fileio.py
+++ b/src/anyio/_core/_fileio.py
@@ -2,8 +2,8 @@ import os
 from os import PathLike
 from typing import Callable, Optional, Union
 
+from .. import to_thread
 from ..abc import AsyncResource
-from ._threads import run_sync_in_worker_thread
 
 
 class AsyncFile(AsyncResource):
@@ -58,43 +58,43 @@ class AsyncFile(AsyncResource):
                 break
 
     async def aclose(self) -> None:
-        return await run_sync_in_worker_thread(self._fp.close)
+        return await to_thread.run_sync(self._fp.close)
 
     async def read(self, size: int = -1) -> Union[bytes, str]:
-        return await run_sync_in_worker_thread(self._fp.read, size)
+        return await to_thread.run_sync(self._fp.read, size)
 
     async def read1(self, size: int = -1) -> Union[bytes, str]:
-        return await run_sync_in_worker_thread(self._fp.read1, size)
+        return await to_thread.run_sync(self._fp.read1, size)
 
     async def readline(self) -> bytes:
-        return await run_sync_in_worker_thread(self._fp.readline)
+        return await to_thread.run_sync(self._fp.readline)
 
     async def readlines(self) -> bytes:
-        return await run_sync_in_worker_thread(self._fp.readlines)
+        return await to_thread.run_sync(self._fp.readlines)
 
     async def readinto(self, b: Union[bytes, memoryview]) -> bytes:
-        return await run_sync_in_worker_thread(self._fp.readinto, b)
+        return await to_thread.run_sync(self._fp.readinto, b)
 
     async def readinto1(self, b: Union[bytes, memoryview]) -> bytes:
-        return await run_sync_in_worker_thread(self._fp.readinto1, b)
+        return await to_thread.run_sync(self._fp.readinto1, b)
 
     async def write(self, b: bytes) -> None:
-        return await run_sync_in_worker_thread(self._fp.write, b)
+        return await to_thread.run_sync(self._fp.write, b)
 
     async def writelines(self, lines: bytes) -> None:
-        return await run_sync_in_worker_thread(self._fp.writelines, lines)
+        return await to_thread.run_sync(self._fp.writelines, lines)
 
     async def truncate(self, size: Optional[int] = None) -> int:
-        return await run_sync_in_worker_thread(self._fp.truncate, size)
+        return await to_thread.run_sync(self._fp.truncate, size)
 
     async def seek(self, offset: int, whence: Optional[int] = os.SEEK_SET) -> int:
-        return await run_sync_in_worker_thread(self._fp.seek, offset, whence)
+        return await to_thread.run_sync(self._fp.seek, offset, whence)
 
     async def tell(self) -> int:
-        return await run_sync_in_worker_thread(self._fp.tell)
+        return await to_thread.run_sync(self._fp.tell)
 
     async def flush(self) -> None:
-        return await run_sync_in_worker_thread(self._fp.flush)
+        return await to_thread.run_sync(self._fp.flush)
 
 
 async def open_file(file: Union[str, PathLike, int], mode: str = 'r', buffering: int = -1,
@@ -109,6 +109,6 @@ async def open_file(file: Union[str, PathLike, int], mode: str = 'r', buffering:
     :return: an asynchronous file object
 
     """
-    fp = await run_sync_in_worker_thread(open, file, mode, buffering, encoding, errors, newline,
-                                         closefd, opener)
+    fp = await to_thread.run_sync(open, file, mode, buffering, encoding, errors, newline,
+                                  closefd, opener)
     return AsyncFile(fp)
diff --git a/src/anyio/_core/_sockets.py b/src/anyio/_core/_sockets.py
index d1346d2..48d2f25 100644
--- a/src/anyio/_core/_sockets.py
+++ b/src/anyio/_core/_sockets.py
@@ -7,6 +7,7 @@ from pathlib import Path
 from socket import AddressFamily, SocketKind
 from typing import Awaitable, List, Optional, Tuple, Union, cast, overload
 
+from .. import to_thread
 from ..abc import (
     ConnectedUDPSocket, IPAddressType, IPSockAddrType, SocketListener, SocketStream, UDPSocket,
     UNIXSocketStream)
@@ -16,7 +17,6 @@ from ._eventloop import get_asynclib
 from ._resources import aclose_forcefully
 from ._synchronization import Event
 from ._tasks import create_task_group, move_on_after
-from ._threads import run_sync_in_worker_thread
 
 if sys.version_info >= (3, 8):
     from typing import Literal
@@ -295,9 +295,9 @@ async def create_unix_listener(
     raw_socket = socket.socket(socket.AF_UNIX)
     raw_socket.setblocking(False)
     try:
-        await run_sync_in_worker_thread(raw_socket.bind, path_str, cancellable=True)
+        await to_thread.run_sync(raw_socket.bind, path_str, cancellable=True)
         if mode is not None:
-            await run_sync_in_worker_thread(chmod, path_str, mode, cancellable=True)
+            await to_thread.run_sync(chmod, path_str, mode, cancellable=True)
 
         raw_socket.listen(backlog)
         return get_asynclib().UNIXSocketListener(raw_socket)
diff --git a/src/anyio/_core/_subprocess_worker.py b/src/anyio/_core/_subprocess_worker.py
deleted file mode 100644
index 98654ef..0000000
--- a/src/anyio/_core/_subprocess_worker.py
+++ /dev/null
@@ -1,70 +0,0 @@
-import os
-import pickle
-import sys
-from importlib.abc import Loader
-from importlib.util import module_from_spec, spec_from_file_location
-from typing import Optional, cast
-
-
-def process_worker() -> None:
-    # Redirect standard streams to os.devnull so that user code won't interfere with the
-    # parent-worker communication
-    stdin = sys.stdin
-    stdout = sys.stdout
-    sys.stdin = open(os.devnull)
-    sys.stdout = open(os.devnull, 'w')
-
-    stdout.buffer.write(b'READY\n')
-    while True:
-        retval = exception = None
-        try:
-            command, *args = pickle.load(stdin.buffer)
-        except EOFError:
-            return
-        except BaseException as exc:
-            exception = exc
-        else:
-            if command == 'run':
-                func, args = args
-                try:
-                    retval = func(*args)
-                except BaseException as exc:
-                    exception = exc
-            elif command == 'init':
-                main_module_path: Optional[str]
-                sys.path, main_module_path = args
-                del sys.modules['__main__']
-                if main_module_path:
-                    # Load the parent's main module but as __mp_main__ instead of __main__
-                    # (like multiprocessing does) to avoid infinite recursion
-                    try:
-                        spec = spec_from_file_location('__mp_main__', main_module_path)
-                        if spec and spec.loader:
-                            main = module_from_spec(spec)
-                            cast(Loader, spec.loader).exec_module(main)
-                            sys.modules['__main__'] = main
-                    except BaseException as exc:
-                        exception = exc
-
-        try:
-            if exception is not None:
-                status = b'EXCEPTION'
-                pickled = pickle.dumps(exception, pickle.HIGHEST_PROTOCOL)
-            else:
-                status = b'RETURN'
-                pickled = pickle.dumps(retval, pickle.HIGHEST_PROTOCOL)
-        except BaseException as exc:
-            exception = exc
-            status = b'EXCEPTION'
-            pickled = pickle.dumps(exc, pickle.HIGHEST_PROTOCOL)
-
-        stdout.buffer.write(b'%s %d\n' % (status, len(pickled)))
-        stdout.buffer.write(pickled)
-
-        # Respect SIGTERM
-        if isinstance(exception, SystemExit):
-            raise exception
-
-
-if __name__ == '__main__':
-    process_worker()
diff --git a/src/anyio/_core/_subprocesses.py b/src/anyio/_core/_subprocesses.py
index 7ebb1f5..95b7f5c 100644
--- a/src/anyio/_core/_subprocesses.py
+++ b/src/anyio/_core/_subprocesses.py
@@ -1,26 +1,9 @@
-import os
-import pickle
-import subprocess
-import sys
-from collections import deque
 from subprocess import DEVNULL, PIPE, CalledProcessError, CompletedProcess
-from typing import Callable, Deque, List, Optional, Sequence, Set, Tuple, TypeVar, Union, cast
+from typing import Optional, Sequence, Union, cast
 
-from ..abc import ByteReceiveStream, ByteSendStream, Process
-from ..lowlevel import RunVar, checkpoint_if_cancelled
-from ..streams.buffered import BufferedByteReceiveStream
-from ._eventloop import current_time, get_asynclib, get_cancelled_exc_class
-from ._exceptions import BrokenWorkerProcess
-from ._synchronization import CapacityLimiter
-from ._tasks import CancelScope, create_task_group, fail_after
-
-T_Retval = TypeVar('T_Retval')
-_process_pool_workers: RunVar[Set[Process]] = RunVar('_process_pool_workers')
-_process_pool_idle_workers: RunVar[Deque[Tuple[Process, float]]] = RunVar(
-    '_process_pool_idle_workers')
-_default_process_limiter: RunVar[CapacityLimiter] = RunVar('_default_process_limiter')
-
-WORKER_MAX_IDLE_TIME = 300  # 5 minutes
+from ..abc import Process
+from ._eventloop import get_asynclib
+from ._tasks import create_task_group
 
 
 async def run_process(command: Union[str, Sequence[str]], *, input: Optional[bytes] = None,
@@ -92,145 +75,3 @@ async def open_process(command: Union[str, Sequence[str]], *, stdin: int = PIPE,
     shell = isinstance(command, str)
     return await get_asynclib().open_process(command, shell=shell, stdin=stdin, stdout=stdout,
                                              stderr=stderr)
-
-
-async def run_sync_in_process(
-        func: Callable[..., T_Retval], *args, cancellable: bool = False,
-        limiter: Optional[CapacityLimiter] = None) -> T_Retval:
-    """
-    Call the given function with the given arguments in a worker process.
-
-    If the ``cancellable`` option is enabled and the task waiting for its completion is cancelled,
-    the worker process running it will be abruptly terminated using SIGKILL (or
-    ``terminateProcess()`` on Windows).
-
-    :param func: a callable
-    :param args: positional arguments for the callable
-    :param cancellable: ``True`` to allow cancellation of the operation while it's running
-    :param limiter: capacity limiter to use to limit the total amount of processes running
-        (if omitted, the default limiter is used)
-    :return: an awaitable that yields the return value of the function.
-
-    """
-    async def send_raw_command(pickled_cmd: bytes):
-        try:
-            await stdin.send(pickled_cmd)
-            response = await buffered.receive_until(b'\n', 50)
-            status, length = response.split(b' ')
-            if status not in (b'RETURN', b'EXCEPTION'):
-                raise RuntimeError(f'Worker process returned unexpected response: {response!r}')
-
-            pickled_response = await buffered.receive_exactly(int(length))
-        except BaseException as exc:
-            workers.discard(process)
-            try:
-                process.kill()
-                with CancelScope(shield=True):
-                    await process.aclose()
-            except ProcessLookupError:
-                pass
-
-            if isinstance(exc, get_cancelled_exc_class()):
-                raise
-            else:
-                raise BrokenWorkerProcess from exc
-
-        retval = pickle.loads(pickled_response)
-        if status == b'EXCEPTION':
-            assert isinstance(retval, BaseException)
-            raise retval
-        else:
-            return retval
-
-    # First pickle the request before trying to reserve a worker process
-    await checkpoint_if_cancelled()
-    request = pickle.dumps(('run', func, args), protocol=pickle.HIGHEST_PROTOCOL)
-
-    # If this is the first run in this event loop thread, set up the necessary variables
-    try:
-        workers = _process_pool_workers.get()
-        idle_workers = _process_pool_idle_workers.get()
-    except LookupError:
-        workers = set()
-        idle_workers = deque()
-        _process_pool_workers.set(workers)
-        _process_pool_idle_workers.set(idle_workers)
-        get_asynclib().setup_process_pool_exit_at_shutdown(workers)
-
-    async with (limiter or current_default_worker_process_limiter()):
-        # Pop processes from the pool (starting from the most recently used) until we find one that
-        # hasn't exited yet
-        process: Process
-        while idle_workers:
-            process, idle_since = idle_workers.pop()
-            if process.returncode is None:
-                stdin = cast(ByteSendStream, process.stdin)
-                buffered = BufferedByteReceiveStream(cast(ByteReceiveStream, process.stdout))
-
-                # Prune any other workers that have been idle for WORKER_MAX_IDLE_TIME seconds or
-                # longer
-                now = current_time()
-                killed_processes: List[Process] = []
-                while idle_workers:
-                    if now - idle_workers[0][1] < WORKER_MAX_IDLE_TIME:
-                        break
-
-                    process, idle_since = idle_workers.popleft()
-                    process.kill()
-                    workers.remove(process)
-                    killed_processes.append(process)
-
-                with CancelScope(shield=True):
-                    for process in killed_processes:
-                        await process.aclose()
-
-                break
-
-            workers.remove(process)
-        else:
-            command = [sys.executable, '-u', '-m', 'anyio._core._subprocess_worker']
-            process = await open_process(command, stdin=subprocess.PIPE,
-                                         stdout=subprocess.PIPE)
-            try:
-                stdin = cast(ByteSendStream, process.stdin)
-                buffered = BufferedByteReceiveStream(cast(ByteReceiveStream, process.stdout))
-                with fail_after(20):
-                    message = await buffered.receive(6)
-
-                if message != b'READY\n':
-                    raise BrokenWorkerProcess(
-                        f'Worker process returned unexpected response: {message!r}')
-
-                main_module_path = getattr(sys.modules['__main__'], '__file__', None)
-                pickled = pickle.dumps(('init', sys.path, main_module_path),
-                                       protocol=pickle.HIGHEST_PROTOCOL)
-                await send_raw_command(pickled)
-            except (BrokenWorkerProcess, get_cancelled_exc_class()):
-                raise
-            except BaseException as exc:
-                process.kill()
-                raise BrokenWorkerProcess('Error during worker process initialization') from exc
-
-            workers.add(process)
-
-        with CancelScope(shield=not cancellable):
-            try:
-                return await send_raw_command(request)
-            finally:
-                if process in workers:
-                    idle_workers.append((process, current_time()))
-
-
-def current_default_worker_process_limiter() -> CapacityLimiter:
-    """
-    Return the capacity limiter that is used by default to limit the number of worker processes.
-
-    :return: a capacity limiter object
-
-    """
-    try:
-        return _default_process_limiter.get()
-    except LookupError:
-        limiter = CapacityLimiter(os.cpu_count() or 2)
-        _default_process_limiter.set(limiter)
-        return limiter
diff --git a/src/anyio/_core/_threads.py b/src/anyio/_core/_threads.py
index 4b301ef..2214ecb 100644
--- a/src/anyio/_core/_threads.py
+++ b/src/anyio/_core/_threads.py
@@ -1,9 +1,9 @@
-from concurrent.futures import FIRST_COMPLETED, Future, ThreadPoolExecutor, wait
-from contextlib import contextmanager
-from typing import Any, Callable, Coroutine, Dict, Generator, Iterable, Optional, TypeVar, cast
+from builtins import DeprecationWarning
+from typing import Callable, Optional, TypeVar
+from warnings import warn
 
-from ..abc import BlockingPortal, CapacityLimiter
-from ._eventloop import get_asynclib, run, threadlocals
+from ..abc import CapacityLimiter
+from ._eventloop import get_asynclib
 
 T_Retval = TypeVar('T_Retval')
 
@@ -26,106 +26,7 @@ async def run_sync_in_worker_thread(
     :return: an awaitable that yields the return value of the function.
 
     """
+    warn('run_sync_in_worker_thread() is deprecated - use anyio.to_thread.run_sync() instead',
+         DeprecationWarning)
     return await get_asynclib().run_sync_in_worker_thread(func, *args, cancellable=cancellable,
                                                           limiter=limiter)
-
-
-def run_sync_from_thread(func: Callable[..., T_Retval], *args) -> T_Retval:
-    """
-    Call a function in the event loop thread from a worker thread.
-
-    :param func: a callable
-    :param args: positional arguments for the callable
-    :return: the return value of the callable
-
-    """
-    try:
-        asynclib = threadlocals.current_async_module
-    except AttributeError:
-        raise RuntimeError('This function can only be run from an AnyIO worker thread')
-
-    return asynclib.run_sync_from_thread(func, *args)
-
-
-def run_async_from_thread(func: Callable[..., Coroutine[Any, Any, T_Retval]], *args) -> T_Retval:
-    """
-    Call a coroutine function from a worker thread.
-
-    :param func: a coroutine function
-    :param args: positional arguments for the callable
-    :return: the return value of the coroutine function
-
-    """
-    try:
-        asynclib = threadlocals.current_async_module
-    except AttributeError:
-        raise RuntimeError('This function can only be run from an AnyIO worker thread')
-
-    return asynclib.run_async_from_thread(func, *args)
-
-
-def current_default_worker_thread_limiter() -> CapacityLimiter:
-    """
-    Return the capacity limiter that is used by default to limit the number of concurrent threads.
-
-    :return: a capacity limiter object
-
-    """
-    return get_asynclib().current_default_thread_limiter()
-
-
-def create_blocking_portal() -> BlockingPortal:
-    """
-    Create a portal for running functions in the event loop thread from external threads.
-
-    Use this function in asynchronous code when you need to allow external threads access to the
-    event loop where your asynchronous code is currently running.
-    """
-    return get_asynclib().BlockingPortal()
-
-
-@contextmanager
-def start_blocking_portal(
-        backend: str = 'asyncio',
-        backend_options: Optional[Dict[str, Any]] = None) -> Generator[BlockingPortal, Any, None]:
-    """
-    Start a new event loop in a new thread and run a blocking portal in its main task.
-
-    The parameters are the same as for :func:`~anyio.run`.
-
-    :param backend: name of the backend
-    :param backend_options: backend options
-    :return: a context manager that yields a blocking portal
-
-    .. versionchanged:: 3.0
-        Usage as a context manager is now required.
-
-    """
-    async def run_portal():
-        async with create_blocking_portal() as portal_:
-            if future.set_running_or_notify_cancel():
-                future.set_result(portal_)
-                await portal_.sleep_until_stopped()
-
-    future: Future[BlockingPortal] = Future()
-    with ThreadPoolExecutor(1) as executor:
-        run_future = executor.submit(run, run_portal, backend=backend,
-                                     backend_options=backend_options)
-        try:
-            wait(cast(Iterable[Future], [run_future, future]), return_when=FIRST_COMPLETED)
-        except BaseException:
-            future.cancel()
-            run_future.cancel()
-            raise
-
-        if future.done():
-            portal = future.result()
-            try:
-                yield portal
-            except BaseException:
-                portal.call(portal.stop, True)
-                raise
-
-            portal.call(portal.stop, False)
-
-        run_future.result()
diff --git a/src/anyio/from_thread.py b/src/anyio/from_thread.py
new file mode 100644
index 0000000..50e643f
--- /dev/null
+++ b/src/anyio/from_thread.py
@@ -0,0 +1,352 @@
+import threading
+from abc import ABCMeta, abstractmethod
+from asyncio import iscoroutine
+from concurrent.futures import FIRST_COMPLETED, Future, ThreadPoolExecutor, wait
+from contextlib import AbstractContextManager, contextmanager
+from types import TracebackType
+from typing import (
+    Any, AsyncContextManager, Callable, ContextManager, Coroutine, Dict, Generator, Iterable,
+    Optional, Tuple, Type, TypeVar, cast, overload)
+from warnings import warn
+
+from ._core import _eventloop
+from ._core._eventloop import get_asynclib, get_cancelled_exc_class, threadlocals
+from ._core._synchronization import Event
+from ._core._tasks import CancelScope, TaskStatus, create_task_group
+
+T_Retval = TypeVar('T_Retval')
+T_co = TypeVar('T_co', covariant=True)
+
+
+def run(func: Callable[..., Coroutine[Any, Any, T_Retval]], *args) -> T_Retval:
+    """
+    Call a coroutine function from a worker thread.
+
+    :param func: a coroutine function
+    :param args: positional arguments for the callable
+    :return: the return value of the coroutine function
+
+    """
+    try:
+        asynclib = threadlocals.current_async_module
+    except AttributeError:
+        raise RuntimeError('This function can only be run from an AnyIO worker thread')
+
+    return asynclib.run_async_from_thread(func, *args)
+
+
+def run_async_from_thread(func: Callable[..., Coroutine[Any, Any, T_Retval]], *args) -> T_Retval:
+    warn('run_async_from_thread() has been deprecated, use anyio.from_thread.run() instead',
+         DeprecationWarning)
+    return run(func, *args)
+
+
+def run_sync(func: Callable[..., T_Retval], *args) -> T_Retval:
+    """
+    Call a function in the event loop thread from a worker thread.
+
+    :param func: a callable
+    :param args: positional arguments for the callable
+    :return: the return value of the callable
+
+    """
+    try:
+        asynclib = threadlocals.current_async_module
+    except AttributeError:
+        raise RuntimeError('This function can only be run from an AnyIO worker thread')
+
+    return asynclib.run_sync_from_thread(func, *args)
+
+
+def run_sync_from_thread(func: Callable[..., T_Retval], *args) -> T_Retval:
+    warn('run_sync_from_thread() has been deprecated, use anyio.from_thread.run_sync() instead',
+         DeprecationWarning)
+    return run_sync(func, *args)
+
+
+class _BlockingAsyncContextManager(AbstractContextManager):
+    _enter_future: Future
+    _exit_future: Future
+    _exit_event: Event
+    _exit_exc_info: Tuple[Optional[Type[BaseException]], Optional[BaseException],
+                          Optional[TracebackType]]
+
+    def __init__(self, async_cm: AsyncContextManager[T_co], portal: 'BlockingPortal'):
+        self._async_cm = async_cm
+        self._portal = portal
+
+    async def run_async_cm(self):
+        try:
+            self._exit_event = Event()
+            value = await self._async_cm.__aenter__()
+        except BaseException as exc:
+            self._enter_future.set_exception(exc)
+            raise
+        else:
+            self._enter_future.set_result(value)
+
+        await self._exit_event.wait()
+        return await self._async_cm.__aexit__(*self._exit_exc_info)
+
+    def __enter__(self) -> T_co:
+        self._enter_future = Future()
+        self._exit_future = self._portal.spawn_task(self.run_async_cm)
+        cm = self._enter_future.result()
+        return cast(T_co, cm)
+
+    def __exit__(self, __exc_type: Optional[Type[BaseException]],
+                 __exc_value: Optional[BaseException],
+                 __traceback: Optional[TracebackType]) -> Optional[bool]:
+        self._exit_exc_info = __exc_type, __exc_value, __traceback
+        self._portal.call(self._exit_event.set)
+        return self._exit_future.result()
+
+
+class _BlockingPortalTaskStatus(TaskStatus):
+    def __init__(self, future: Future):
+        self._future = future
+
+    def started(self, value=None) -> None:
+        self._future.set_result(value)
+
+
+class BlockingPortal(metaclass=ABCMeta):
+    """An object tied that lets external threads run code in an asynchronous event loop."""
+
+    def __init__(self):
+        self._event_loop_thread_id = threading.get_ident()
+        self._stop_event = Event()
+        self._task_group = create_task_group()
+        self._cancelled_exc_class = get_cancelled_exc_class()
+
+    async def __aenter__(self) -> 'BlockingPortal':
+        await self._task_group.__aenter__()
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        await self.stop()
+        return await self._task_group.__aexit__(exc_type, exc_val, exc_tb)
+
+    def _check_running(self) -> None:
+        if self._event_loop_thread_id is None:
+            raise RuntimeError('This portal is not running')
+        if self._event_loop_thread_id == threading.get_ident():
+            raise RuntimeError('This method cannot be called from the event loop thread')
+
+    async def sleep_until_stopped(self) -> None:
+        """Sleep until :meth:`stop` is called."""
+        await self._stop_event.wait()
+
+    async def stop(self, cancel_remaining: bool = False) -> None:
+        """
+        Signal the portal to shut down.
+
+        This marks the portal as no longer accepting new calls and exits from
+        :meth:`sleep_until_stopped`.
+
+        :param cancel_remaining: ``True`` to cancel all the remaining tasks, ``False`` to let them
+            finish before returning
+
+        """
+        self._event_loop_thread_id = None
+        self._stop_event.set()
+        if cancel_remaining:
+            self._task_group.cancel_scope.cancel()
+
+    async def _call_func(self, func: Callable, args: tuple, kwargs: Dict[str, Any],
+                         future: Future) -> None:
+        def callback(f: Future):
+            if f.cancelled():
+                self.call(scope.cancel)
+
+        try:
+            retval = func(*args, **kwargs)
+            if iscoroutine(retval):
+                with CancelScope() as scope:
+                    if future.cancelled():
+                        scope.cancel()
+                    else:
+                        future.add_done_callback(callback)
+
+                    retval = await retval
+        except self._cancelled_exc_class:
+            future.cancel()
+        except BaseException as exc:
+            if not future.cancelled():
+                future.set_exception(exc)
+
+            # Let base exceptions fall through
+            if not isinstance(exc, Exception):
+                raise
+        else:
+            if not future.cancelled():
+                future.set_result(retval)
+        finally:
+            scope = None
+
+    @abstractmethod
+    def _spawn_task_from_thread(self, func: Callable, args: tuple, kwargs: Dict[str, Any],
+                                name, future: Future) -> None:
+        """
+        Spawn a new task using the given callable.
+
+        Implementors must ensure that the future is resolved when the task finishes.
+
+        :param func: a callable
+        :param args: positional arguments to be passed to the callable
+        :param kwargs: keyword arguments to be passed to the callable
+        :param name: name of the task (will be coerced to a string if not ``None``)
+        :param future: a future that will resolve to the return value of the callable, or the
+            exception raised during its execution
+        """
+
+    @overload
+    def call(self, func: Callable[..., Coroutine[Any, Any, T_Retval]], *args) -> T_Retval:
+        ...
+
+    @overload
+    def call(self, func: Callable[..., T_Retval], *args) -> T_Retval:
+        ...
+
+    def call(self, func, *args):
+        """
+        Call the given function in the event loop thread.
+
+        If the callable returns a coroutine object, it is awaited on.
+
+        :param func: any callable
+        :raises RuntimeError: if the portal is not running or if this method is called from within
+            the event loop thread
+
+        """
+        return self.spawn_task(func, *args).result()
+
+    def spawn_task(self, func: Callable[..., Coroutine], *args, name=None) -> Future:
+        """
+        Spawn a task in the portal's task group.
+
+        The task will be run inside a cancel scope which can be cancelled by cancelling the
+        returned future.
+
+        :param func: the target coroutine function
+        :param args: positional arguments passed to ``func``
+        :param name: name of the task (will be coerced to a string if not ``None``)
+        :return: a future that resolves with the return value of the callable if the task completes
+            successfully, or with the exception raised in the task
+        :raises RuntimeError: if the portal is not running or if this method is called from within
+            the event loop thread
+
+        .. versionadded:: 2.1
+        .. versionchanged:: 3.0
+            Added the ``name`` argument.
+
+        """
+        self._check_running()
+        f: Future = Future()
+        self._spawn_task_from_thread(func, args, {}, name, f)
+        return f
+
+    def start_task(self, func: Callable[..., Coroutine], *args, name=None) -> Tuple[Future, Any]:
+        """
+        Spawn a task in the portal's task group and wait until it signals for readiness.
+
+        This method works the same way as :meth:`TaskGroup.start`.
+
+        :param func: the target coroutine function
+        :param args: positional arguments passed to ``func``
+        :param name: name of the task (will be coerced to a string if not ``None``)
+        :return: a tuple of (future, task_status_value) where the ``task_status_value`` is the
+            value passed to ``task_status.started()`` from within the target function
+
+        .. versionadded:: 3.0
+
+        """
+        def task_done(future: Future) -> None:
+            if not task_status_future.done():
+                if future.cancelled():
+                    task_status_future.cancel()
+                elif future.exception():
+                    task_status_future.set_exception(future.exception())
+                else:
+                    exc = RuntimeError('Task exited without calling task_status.started()')
+                    task_status_future.set_exception(exc)
+
+        self._check_running()
+        task_status_future: Future = Future()
+        task_status = _BlockingPortalTaskStatus(task_status_future)
+        f: Future = Future()
+        f.add_done_callback(task_done)
+        self._spawn_task_from_thread(func, args, {'task_status': task_status}, name, f)
+        return f, task_status_future.result()
+
+    def wrap_async_context_manager(self, cm: AsyncContextManager[T_co]) -> ContextManager[T_co]:
+        """
+        Wrap an async context manager as a synchronous context manager via this portal.
+
+        Spawns a task that will call both ``__aenter__()`` and ``__aexit__()``, stopping in the
+        middle until the synchronous context manager exits.
+
+        :param cm: an asynchronous context manager
+        :return: a synchronous context manager
+
+        .. versionadded:: 2.1
+
+        """
+        return _BlockingAsyncContextManager(cm, self)
+
+
+def create_blocking_portal() -> BlockingPortal:
+    """
+    Create a portal for running functions in the event loop thread from external threads.
+
+    Use this function in asynchronous code when you need to allow external threads access to the
+    event loop where your asynchronous code is currently running.
+    """
+    return get_asynclib().BlockingPortal()
+
+
+@contextmanager
+def start_blocking_portal(
+        backend: str = 'asyncio',
+        backend_options: Optional[Dict[str, Any]] = None) -> Generator[BlockingPortal, Any, None]:
+    """
+    Start a new event loop in a new thread and run a blocking portal in its main task.
+
+    The parameters are the same as for :func:`~anyio.run`.
+
+    :param backend: name of the backend
+    :param backend_options: backend options
+    :return: a context manager that yields a blocking portal
+
+    .. versionchanged:: 3.0
+        Usage as a context manager is now required.
+
+    """
+    async def run_portal():
+        async with create_blocking_portal() as portal_:
+            if future.set_running_or_notify_cancel():
+                future.set_result(portal_)
+                await portal_.sleep_until_stopped()
+
+    future: Future[BlockingPortal] = Future()
+    with ThreadPoolExecutor(1) as executor:
+        run_future = executor.submit(_eventloop.run, run_portal, backend=backend,
+                                     backend_options=backend_options)
+        try:
+            wait(cast(Iterable[Future], [run_future, future]), return_when=FIRST_COMPLETED)
+        except BaseException:
+            future.cancel()
+            run_future.cancel()
+            raise
+
+        if future.done():
+            portal = future.result()
+            try:
+                yield portal
+            except BaseException:
+                portal.call(portal.stop, True)
+                raise
+
+            portal.call(portal.stop, False)
+
+        run_future.result()
diff --git a/src/anyio/streams/file.py b/src/anyio/streams/file.py
index 0599b31..12442ad 100644
--- a/src/anyio/streams/file.py
+++ b/src/anyio/streams/file.py
@@ -2,10 +2,10 @@ from io import SEEK_SET, UnsupportedOperation
 from pathlib import Path
 from typing import BinaryIO, Union, cast
 
-from anyio import (
-    BrokenResourceError, ClosedResourceError, EndOfStream, TypedAttributeSet,
-    run_sync_in_worker_thread, typed_attribute)
-from anyio.abc import ByteReceiveStream, ByteSendStream
+from .. import (
+    BrokenResourceError, ClosedResourceError, EndOfStream, TypedAttributeSet, to_thread,
+    typed_attribute)
+from ..abc import ByteReceiveStream, ByteSendStream
 
 
 class FileStreamAttribute(TypedAttributeSet):
@@ -22,7 +22,7 @@ class _BaseFileStream:
         self._file = file
 
     async def aclose(self) -> None:
-        await run_sync_in_worker_thread(self._file.close)
+        await to_thread.run_sync(self._file.close)
 
     @property
     def extra_attributes(self):
@@ -60,12 +60,12 @@ class FileReadStream(_BaseFileStream, ByteReceiveStream):
         :param path: path of the file to read from
 
         """
-        file = await run_sync_in_worker_thread(Path(path).open, 'rb')
+        file = await to_thread.run_sync(Path(path).open, 'rb')
         return cls(cast(BinaryIO, file))
 
     async def receive(self, max_bytes: int = 65536) -> bytes:
         try:
-            data = await run_sync_in_worker_thread(self._file.read, max_bytes)
+            data = await to_thread.run_sync(self._file.read, max_bytes)
         except ValueError:
             raise ClosedResourceError from None
         except OSError as exc:
@@ -90,7 +90,7 @@ class FileReadStream(_BaseFileStream, ByteReceiveStream):
         :raises OSError: if the file is not seekable
 
         """
-        return await run_sync_in_worker_thread(self._file.seek, position, whence)
+        return await to_thread.run_sync(self._file.seek, position, whence)
 
     async def tell(self) -> int:
         """
@@ -102,7 +102,7 @@ class FileReadStream(_BaseFileStream, ByteReceiveStream):
         :raises OSError: if the file is not seekable
 
         """
-        return await run_sync_in_worker_thread(self._file.tell)
+        return await to_thread.run_sync(self._file.tell)
 
 
 class FileWriteStream(_BaseFileStream, ByteSendStream):
@@ -125,12 +125,12 @@ class FileWriteStream(_BaseFileStream, ByteSendStream):
 
         """
         mode = 'ab' if append else 'wb'
-        file = await run_sync_in_worker_thread(Path(path).open, mode)
+        file = await to_thread.run_sync(Path(path).open, mode)
         return cls(cast(BinaryIO, file))
 
     async def send(self, item: bytes) -> None:
         try:
-            await run_sync_in_worker_thread(self._file.write, item)
+            await to_thread.run_sync(self._file.write, item)
         except ValueError:
             raise ClosedResourceError from None
         except OSError as exc:
diff --git a/src/anyio/to_process.py b/src/anyio/to_process.py
new file mode 100644
index 0000000..8c18cd7
--- /dev/null
+++ b/src/anyio/to_process.py
@@ -0,0 +1,230 @@
+import os
+import pickle
+import subprocess
+import sys
+from collections import deque
+from importlib.abc import Loader
+from importlib.util import module_from_spec, spec_from_file_location
+from typing import Callable, Deque, List, Optional, Set, Tuple, TypeVar, cast
+
+from ._core._eventloop import current_time, get_asynclib, get_cancelled_exc_class
+from ._core._exceptions import BrokenWorkerProcess
+from ._core._subprocesses import open_process
+from ._core._synchronization import CapacityLimiter
+from ._core._tasks import CancelScope, fail_after
+from .abc import ByteReceiveStream, ByteSendStream, Process
+from .lowlevel import RunVar, checkpoint_if_cancelled
+from .streams.buffered import BufferedByteReceiveStream
+
+WORKER_MAX_IDLE_TIME = 300  # 5 minutes
+
+T_Retval = TypeVar('T_Retval')
+_process_pool_workers: RunVar[Set[Process]] = RunVar('_process_pool_workers')
+_process_pool_idle_workers: RunVar[Deque[Tuple[Process, float]]] = RunVar(
+    '_process_pool_idle_workers')
+_default_process_limiter: RunVar[CapacityLimiter] = RunVar('_default_process_limiter')
+
+
+async def run_sync(
+        func: Callable[..., T_Retval], *args, cancellable: bool = False,
+        limiter: Optional[CapacityLimiter] = None) -> T_Retval:
+    """
+    Call the given function with the given arguments in a worker process.
+
+    If the ``cancellable`` option is enabled and the task waiting for its completion is cancelled,
+    the worker process running it will be abruptly terminated using SIGKILL (or
+    ``terminateProcess()`` on Windows).
+
+    :param func: a callable
+    :param args: positional arguments for the callable
+    :param cancellable: ``True`` to allow cancellation of the operation while it's running
+    :param limiter: capacity limiter to use to limit the total amount of processes running
+        (if omitted, the default limiter is used)
+    :return: an awaitable that yields the return value of the function.
+
+    """
+    async def send_raw_command(pickled_cmd: bytes):
+        try:
+            await stdin.send(pickled_cmd)
+            response = await buffered.receive_until(b'\n', 50)
+            status, length = response.split(b' ')
+            if status not in (b'RETURN', b'EXCEPTION'):
+                raise RuntimeError(f'Worker process returned unexpected response: {response!r}')
+
+            pickled_response = await buffered.receive_exactly(int(length))
+        except BaseException as exc:
+            workers.discard(process)
+            try:
+                process.kill()
+                with CancelScope(shield=True):
+                    await process.aclose()
+            except ProcessLookupError:
+                pass
+
+            if isinstance(exc, get_cancelled_exc_class()):
+                raise
+            else:
+                raise BrokenWorkerProcess from exc
+
+        retval = pickle.loads(pickled_response)
+        if status == b'EXCEPTION':
+            assert isinstance(retval, BaseException)
+            raise retval
+        else:
+            return retval
+
+    # First pickle the request before trying to reserve a worker process
+    await checkpoint_if_cancelled()
+    request = pickle.dumps(('run', func, args), protocol=pickle.HIGHEST_PROTOCOL)
+
+    # If this is the first run in this event loop thread, set up the necessary variables
+    try:
+        workers = _process_pool_workers.get()
+        idle_workers = _process_pool_idle_workers.get()
+    except LookupError:
+        workers = set()
+        idle_workers = deque()
+        _process_pool_workers.set(workers)
+        _process_pool_idle_workers.set(idle_workers)
+        get_asynclib().setup_process_pool_exit_at_shutdown(workers)
+
+    async with (limiter or current_default_process_limiter()):
+        # Pop processes from the pool (starting from the most recently used) until we find one that
+        # hasn't exited yet
+        process: Process
+        while idle_workers:
+            process, idle_since = idle_workers.pop()
+            if process.returncode is None:
+                stdin = cast(ByteSendStream, process.stdin)
+                buffered = BufferedByteReceiveStream(cast(ByteReceiveStream, process.stdout))
+
+                # Prune any other workers that have been idle for WORKER_MAX_IDLE_TIME seconds or
+                # longer
+                now = current_time()
+                killed_processes: List[Process] = []
+                while idle_workers:
+                    if now - idle_workers[0][1] < WORKER_MAX_IDLE_TIME:
+                        break
+
+                    process, idle_since = idle_workers.popleft()
+                    process.kill()
+                    workers.remove(process)
+                    killed_processes.append(process)
+
+                with CancelScope(shield=True):
+                    for process in killed_processes:
+                        await process.aclose()
+
+                break
+
+            workers.remove(process)
+        else:
+            command = [sys.executable, '-u', '-m', __name__]
+            process = await open_process(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE)
+            try:
+                stdin = cast(ByteSendStream, process.stdin)
+                buffered = BufferedByteReceiveStream(cast(ByteReceiveStream, process.stdout))
+                with fail_after(20):
+                    message = await buffered.receive(6)
+
+                if message != b'READY\n':
+                    raise BrokenWorkerProcess(
+                        f'Worker process returned unexpected response: {message!r}')
+
+                main_module_path = getattr(sys.modules['__main__'], '__file__', None)
+                pickled = pickle.dumps(('init', sys.path, main_module_path),
+                                       protocol=pickle.HIGHEST_PROTOCOL)
+                await send_raw_command(pickled)
+            except (BrokenWorkerProcess, get_cancelled_exc_class()):
+                raise
+            except BaseException as exc:
+                process.kill()
+                raise BrokenWorkerProcess('Error during worker process initialization') from exc
+
+            workers.add(process)
+
+        with CancelScope(shield=not cancellable):
+            try:
+                return await send_raw_command(request)
+            finally:
+                if process in workers:
+                    idle_workers.append((process, current_time()))
+
+
+def current_default_process_limiter() -> CapacityLimiter:
+    """
+    Return the capacity limiter that is used by default to limit the number of worker processes.
+
+    :return: a capacity limiter object
+
+    """
+    try:
+        return _default_process_limiter.get()
+    except LookupError:
+        limiter = CapacityLimiter(os.cpu_count() or 2)
+        _default_process_limiter.set(limiter)
+        return limiter
+
+
+def process_worker() -> None:
+    # Redirect standard streams to os.devnull so that user code won't interfere with the
+    # parent-worker communication
+    stdin = sys.stdin
+    stdout = sys.stdout
+    sys.stdin = open(os.devnull)
+    sys.stdout = open(os.devnull, 'w')
+
+    stdout.buffer.write(b'READY\n')
+    while True:
+        retval = exception = None
+        try:
+            command, *args = pickle.load(stdin.buffer)
+        except EOFError:
+            return
+        except BaseException as exc:
+            exception = exc
+        else:
+            if command == 'run':
+                func, args = args
+                try:
+                    retval = func(*args)
+                except BaseException as exc:
+                    exception = exc
+            elif command == 'init':
+                main_module_path: Optional[str]
+                sys.path, main_module_path = args
+                del sys.modules['__main__']
+                if main_module_path:
+                    # Load the parent's main module but as __mp_main__ instead of __main__
+                    # (like multiprocessing does) to avoid infinite recursion
+                    try:
+                        spec = spec_from_file_location('__mp_main__', main_module_path)
+                        if spec and spec.loader:
+                            main = module_from_spec(spec)
+                            cast(Loader, spec.loader).exec_module(main)
+                            sys.modules['__main__'] = main
+                    except BaseException as exc:
+                        exception = exc
+
+        try:
+            if exception is not None:
+                status = b'EXCEPTION'
+                pickled = pickle.dumps(exception, pickle.HIGHEST_PROTOCOL)
+            else:
+                status = b'RETURN'
+                pickled = pickle.dumps(retval, pickle.HIGHEST_PROTOCOL)
+        except BaseException as exc:
+            exception = exc
+            status = b'EXCEPTION'
+            pickled = pickle.dumps(exc, pickle.HIGHEST_PROTOCOL)
+
+        stdout.buffer.write(b'%s %d\n' % (status, len(pickled)))
+        stdout.buffer.write(pickled)
+
+        # Respect SIGTERM
+        if isinstance(exception, SystemExit):
+            raise exception
+
+
+if __name__ == '__main__':
+    process_worker()
diff --git a/src/anyio/to_thread.py b/src/anyio/to_thread.py
new file mode 100644
index 0000000..e042877
--- /dev/null
+++ b/src/anyio/to_thread.py
@@ -0,0 +1,54 @@
+from typing import Callable, Optional, TypeVar
+from warnings import warn
+
+from ._core._eventloop import get_asynclib
+from .abc import CapacityLimiter
+
+T_Retval = TypeVar('T_Retval')
+
+
+async def run_sync(
+        func: Callable[..., T_Retval], *args, cancellable: bool = False,
+        limiter: Optional[CapacityLimiter] = None) -> T_Retval:
+    """
+    Call the given function with the given arguments in a worker thread.
+
+    If the ``cancellable`` option is enabled and the task waiting for its completion is cancelled,
+    the thread will still run its course but its return value (or any raised exception) will be
+    ignored.
+
+    :param func: a callable
+    :param args: positional arguments for the callable
+    :param cancellable: ``True`` to allow cancellation of the operation
+    :param limiter: capacity limiter to use to limit the total amount of threads running
+        (if omitted, the default limiter is used)
+    :return: an awaitable that yields the return value of the function.
+
+    """
+    return await get_asynclib().run_sync_in_worker_thread(func, *args, cancellable=cancellable,
+                                                          limiter=limiter)
+
+
+async def run_sync_in_worker_thread(
+        func: Callable[..., T_Retval], *args, cancellable: bool = False,
+        limiter: Optional[CapacityLimiter] = None) -> T_Retval:
+    warn('run_sync_in_worker_thread() has been deprecated, use anyio.to_thread.run_sync() instead',
+         DeprecationWarning)
+    return await run_sync(func, *args, cancellable=cancellable, limiter=limiter)
+
+
+def current_default_thread_limiter() -> CapacityLimiter:
+    """
+    Return the capacity limiter that is used by default to limit the number of concurrent threads.
+
+    :return: a capacity limiter object
+
+    """
+    return get_asynclib().current_default_thread_limiter()
+
+
+def current_default_worker_thread_limiter() -> CapacityLimiter:
+    warn('current_default_worker_thread_limiter() has been deprecated, '
+         'use anyio.to_thread.current_default_thread_limiter() instead',
+         DeprecationWarning)
+    return current_default_thread_limiter()


Test patch:
diff --git a/tests/test_compat.py b/tests/test_compat.py
index dbdb598..3ec6191 100644
--- a/tests/test_compat.py
+++ b/tests/test_compat.py
@@ -1,13 +1,15 @@
 import signal
 import sys
+import threading
 
 import pytest
 
 from anyio import (
     CancelScope, CapacityLimiter, Condition, Event, Lock, Semaphore, TaskInfo,
-    create_memory_object_stream, create_task_group, current_effective_deadline, current_time,
-    fail_after, get_current_task, get_running_tasks, maybe_async, maybe_async_cm, move_on_after,
-    open_signal_receiver, sleep)
+    create_memory_object_stream, create_task_group, current_default_worker_thread_limiter,
+    current_effective_deadline, current_time, fail_after, get_current_task, get_running_tasks,
+    maybe_async, maybe_async_cm, move_on_after, open_signal_receiver, run_async_from_thread,
+    run_sync_from_thread, run_sync_in_worker_thread, sleep, to_thread)
 
 pytestmark = pytest.mark.anyio
 
@@ -137,3 +139,30 @@ class TestDeprecations:
         with pytest.raises(TimeoutError), pytest.deprecated_call():
             async with fail_after(0):
                 pass
+
+    async def test_run_sync_in_worker_thread(self):
+        with pytest.deprecated_call():
+            thread_id = await run_sync_in_worker_thread(threading.get_ident)
+            assert thread_id != threading.get_ident()
+
+    async def test_run_async_from_thread(self):
+        async def get_ident():
+            return threading.get_ident()
+
+        def thread_func():
+            with pytest.deprecated_call():
+                return run_async_from_thread(get_ident)
+
+        assert await to_thread.run_sync(thread_func) == threading.get_ident()
+
+    async def test_run_sync_from_thread(self):
+        def thread_func():
+            with pytest.deprecated_call():
+                return run_sync_from_thread(threading.get_ident)
+
+        assert await to_thread.run_sync(thread_func) == threading.get_ident()
+
+    async def test_current_default_worker_thread_limiter(self):
+        with pytest.deprecated_call():
+            default_limiter = to_thread.current_default_thread_limiter()
+            assert current_default_worker_thread_limiter() is default_limiter
diff --git a/tests/test_threads.py b/tests/test_from_thread.py
similarity index 58%
rename from tests/test_threads.py
rename to tests/test_from_thread.py
index d45dc2a..badc413 100644
--- a/tests/test_threads.py
+++ b/tests/test_from_thread.py
@@ -1,230 +1,111 @@
-import asyncio
-import sys
 import threading
-import time
 from concurrent.futures import CancelledError
 from contextlib import suppress
 
 import pytest
 
 from anyio import (
-    CapacityLimiter, Event, create_blocking_portal, create_task_group, get_cancelled_exc_class,
-    get_current_task, run, run_async_from_thread, run_sync_from_thread, run_sync_in_worker_thread,
-    sleep, start_blocking_portal, wait_all_tasks_blocked)
-
-if sys.version_info < (3, 7):
-    current_task = asyncio.Task.current_task
-else:
-    current_task = asyncio.current_task
+    Event, create_blocking_portal, from_thread, get_cancelled_exc_class, get_current_task, run,
+    sleep, start_blocking_portal, to_thread, wait_all_tasks_blocked)
 
 pytestmark = pytest.mark.anyio
 
 
-async def test_run_async_from_thread():
-    async def add(a, b):
-        assert threading.get_ident() == event_loop_thread_id
-        return a + b
-
-    def worker(a, b):
-        assert threading.get_ident() != event_loop_thread_id
-        return run_async_from_thread(add, a, b)
-
-    event_loop_thread_id = threading.get_ident()
-    result = await run_sync_in_worker_thread(worker, 1, 2)
-    assert result == 3
-
-
-async def test_run_sync_from_thread():
-    def add(a, b):
-        assert threading.get_ident() == event_loop_thread_id
-        return a + b
-
-    def worker(a, b):
-        assert threading.get_ident() != event_loop_thread_id
-        return run_sync_from_thread(add, a, b)
-
-    event_loop_thread_id = threading.get_ident()
-    result = await run_sync_in_worker_thread(worker, 1, 2)
-    assert result == 3
-
-
-def test_run_sync_from_thread_pooling():
-    async def main():
-        thread_ids = set()
-        for _ in range(5):
-            thread_ids.add(await run_sync_in_worker_thread(threading.get_ident))
-
-        # Expects that all the work has been done in the same worker thread
-        assert len(thread_ids) == 1
-        assert thread_ids.pop() != threading.get_ident()
-        assert threading.active_count() == initial_count + 1
-
-    # The thread should not exist after the event loop has been closed
-    initial_count = threading.active_count()
-    run(main, backend='asyncio')
-    assert threading.active_count() == initial_count
-
-
-async def test_run_async_from_thread_exception():
-    async def add(a, b):
-        assert threading.get_ident() == event_loop_thread_id
-        return a + b
-
-    def worker(a, b):
-        assert threading.get_ident() != event_loop_thread_id
-        return run_async_from_thread(add, a, b)
-
-    event_loop_thread_id = threading.get_ident()
-    with pytest.raises(TypeError) as exc:
-        await run_sync_in_worker_thread(worker, 1, 'foo')
-
-    exc.match("unsupported operand type")
-
+class TestRunAsyncFromThread:
+    async def test_run_async_from_thread(self):
+        async def add(a, b):
+            assert threading.get_ident() == event_loop_thread_id
+            return a + b
 
-async def test_run_sync_from_thread_exception():
-    def add(a, b):
-        assert threading.get_ident() == event_loop_thread_id
-        return a + b
+        def worker(a, b):
+            assert threading.get_ident() != event_loop_thread_id
+            return from_thread.run(add, a, b)
 
-    def worker(a, b):
-        assert threading.get_ident() != event_loop_thread_id
-        return run_sync_from_thread(add, a, b)
+        event_loop_thread_id = threading.get_ident()
+        result = await to_thread.run_sync(worker, 1, 2)
+        assert result == 3
 
-    event_loop_thread_id = threading.get_ident()
-    with pytest.raises(TypeError) as exc:
-        await run_sync_in_worker_thread(worker, 1, 'foo')
+    async def test_run_sync_from_thread(self):
+        def add(a, b):
+            assert threading.get_ident() == event_loop_thread_id
+            return a + b
 
-    exc.match("unsupported operand type")
+        def worker(a, b):
+            assert threading.get_ident() != event_loop_thread_id
+            return from_thread.run_sync(add, a, b)
 
+        event_loop_thread_id = threading.get_ident()
+        result = await to_thread.run_sync(worker, 1, 2)
+        assert result == 3
 
-async def test_run_anyio_async_func_from_thread():
-    def worker(*args):
-        run_async_from_thread(sleep, *args)
-        return True
+    def test_run_sync_from_thread_pooling(self):
+        async def main():
+            thread_ids = set()
+            for _ in range(5):
+                thread_ids.add(await to_thread.run_sync(threading.get_ident))
 
-    assert await run_sync_in_worker_thread(worker, 0)
+            # Expects that all the work has been done in the same worker thread
+            assert len(thread_ids) == 1
+            assert thread_ids.pop() != threading.get_ident()
+            assert threading.active_count() == initial_count + 1
 
+        # The thread should not exist after the event loop has been closed
+        initial_count = threading.active_count()
+        run(main, backend='asyncio')
+        assert threading.active_count() == initial_count
 
-async def test_run_in_thread_cancelled():
-    def thread_worker():
-        nonlocal state
-        state = 2
+    async def test_run_async_from_thread_exception(self):
+        async def add(a, b):
+            assert threading.get_ident() == event_loop_thread_id
+            return a + b
 
-    async def worker():
-        nonlocal state
-        state = 1
-        await run_sync_in_worker_thread(thread_worker)
-        state = 3
+        def worker(a, b):
+            assert threading.get_ident() != event_loop_thread_id
+            return from_thread.run(add, a, b)
 
-    state = 0
-    async with create_task_group() as tg:
-        tg.spawn(worker)
-        tg.cancel_scope.cancel()
+        event_loop_thread_id = threading.get_ident()
+        with pytest.raises(TypeError) as exc:
+            await to_thread.run_sync(worker, 1, 'foo')
 
-    assert state == 1
+        exc.match("unsupported operand type")
 
+    async def test_run_sync_from_thread_exception(self):
+        def add(a, b):
+            assert threading.get_ident() == event_loop_thread_id
+            return a + b
 
-async def test_run_in_thread_exception():
-    def thread_worker():
-        raise ValueError('foo')
+        def worker(a, b):
+            assert threading.get_ident() != event_loop_thread_id
+            return from_thread.run_sync(add, a, b)
 
-    with pytest.raises(ValueError) as exc:
-        await run_sync_in_worker_thread(thread_worker)
+        event_loop_thread_id = threading.get_ident()
+        with pytest.raises(TypeError) as exc:
+            await to_thread.run_sync(worker, 1, 'foo')
 
-    exc.match('^foo$')
+        exc.match("unsupported operand type")
 
+    async def test_run_anyio_async_func_from_thread(self):
+        def worker(*args):
+            from_thread.run(sleep, *args)
+            return True
 
-async def test_run_in_custom_limiter():
-    def thread_worker():
-        nonlocal num_active_threads, max_active_threads
-        num_active_threads += 1
-        max_active_threads = max(num_active_threads, max_active_threads)
-        event.wait(1)
-        num_active_threads -= 1
+        assert await to_thread.run_sync(worker, 0)
 
-    async def task_worker():
-        await run_sync_in_worker_thread(thread_worker, limiter=limiter)
-
-    event = threading.Event()
-    num_active_threads = max_active_threads = 0
-    limiter = CapacityLimiter(3)
-    async with create_task_group() as tg:
-        for _ in range(4):
-            tg.spawn(task_worker)
-
-        await sleep(0.1)
-        assert num_active_threads == 3
-        assert limiter.borrowed_tokens == 3
-        event.set()
-
-    assert num_active_threads == 0
-    assert max_active_threads == 3
-
-
-def test_run_async_from_unclaimed_thread():
-    async def foo():
-        pass
-
-    exc = pytest.raises(RuntimeError, run_async_from_thread, foo)
-    exc.match('This function can only be run from an AnyIO worker thread')
-
-
-def test_run_sync_from_unclaimed_thread():
-    def foo():
-        pass
-
-    exc = pytest.raises(RuntimeError, run_sync_from_thread, foo)
-    exc.match('This function can only be run from an AnyIO worker thread')
-
-
-@pytest.mark.parametrize('cancellable, expected_last_active', [
-    (False, 'task'),
-    (True, 'thread')
-], ids=['uncancellable', 'cancellable'])
-async def test_cancel_worker_thread(cancellable, expected_last_active):
-    """
-    Test that when a task running a worker thread is cancelled, the cancellation is not acted on
-    until the thread finishes.
-
-    """
-    def thread_worker():
-        nonlocal last_active
-        run_sync_from_thread(sleep_event.set)
-        time.sleep(0.2)
-        last_active = 'thread'
-        run_sync_from_thread(finish_event.set)
-
-    async def task_worker():
-        nonlocal last_active
-        try:
-            await run_sync_in_worker_thread(thread_worker, cancellable=cancellable)
-        finally:
-            last_active = 'task'
-
-    sleep_event = Event()
-    finish_event = Event()
-    last_active = None
-    async with create_task_group() as tg:
-        tg.spawn(task_worker)
-        await sleep_event.wait()
-        tg.cancel_scope.cancel()
+    def test_run_async_from_unclaimed_thread(self):
+        async def foo():
+            pass
 
-    await finish_event.wait()
-    assert last_active == expected_last_active
+        exc = pytest.raises(RuntimeError, from_thread.run, foo)
+        exc.match('This function can only be run from an AnyIO worker thread')
 
 
-@pytest.mark.parametrize('anyio_backend', ['asyncio'])
-async def test_cancel_asyncio_native_task():
-    async def run_in_thread():
-        nonlocal task
-        task = current_task()
-        await run_sync_in_worker_thread(time.sleep, 1, cancellable=True)
+class TestRunSyncFromThread:
+    def test_run_sync_from_unclaimed_thread(self):
+        def foo():
+            pass
 
-    task = None
-    async with create_task_group() as tg:
-        tg.spawn(run_in_thread)
-        await wait_all_tasks_blocked()
-        task.cancel()
+        exc = pytest.raises(RuntimeError, from_thread.run_sync, foo)
+        exc.match('This function can only be run from an AnyIO worker thread')
 
 
 class TestBlockingPortal:
@@ -250,7 +131,7 @@ class TestBlockingPortal:
         async with create_blocking_portal() as portal:
             thread = threading.Thread(target=external_thread)
             thread.start()
-            await run_sync_in_worker_thread(thread.join)
+            await to_thread.run_sync(thread.join)
 
         for thread_id in thread_ids:
             assert thread_id == threading.get_ident()
@@ -276,8 +157,8 @@ class TestBlockingPortal:
                 assert not results
                 raise Exception
 
-        await run_sync_in_worker_thread(thread1.join)
-        await run_sync_in_worker_thread(thread2.join)
+        await to_thread.run_sync(thread1.join)
+        await to_thread.run_sync(thread2.join)
 
         assert len(results) == 2
         assert isinstance(results[0], CancelledError)
@@ -302,8 +183,8 @@ class TestBlockingPortal:
             await sleep(0.1)
             assert not results
 
-        await run_sync_in_worker_thread(thread1.join)
-        await run_sync_in_worker_thread(thread2.join)
+        await to_thread.run_sync(thread1.join)
+        await to_thread.run_sync(thread2.join)
 
         assert results == [None, None]
 
diff --git a/tests/test_signals.py b/tests/test_signals.py
index b8adc4e..554c03a 100644
--- a/tests/test_signals.py
+++ b/tests/test_signals.py
@@ -4,7 +4,7 @@ import sys
 
 import pytest
 
-from anyio import create_task_group, fail_after, open_signal_receiver, run_sync_in_worker_thread
+from anyio import create_task_group, fail_after, open_signal_receiver, to_thread
 
 pytestmark = [
     pytest.mark.anyio,
@@ -17,8 +17,8 @@ pytestmark = [
 
 async def test_receive_signals():
     with open_signal_receiver(signal.SIGUSR1, signal.SIGUSR2) as sigiter:
-        await run_sync_in_worker_thread(os.kill, os.getpid(), signal.SIGUSR1)
-        await run_sync_in_worker_thread(os.kill, os.getpid(), signal.SIGUSR2)
+        await to_thread.run_sync(os.kill, os.getpid(), signal.SIGUSR1)
+        await to_thread.run_sync(os.kill, os.getpid(), signal.SIGUSR2)
         with fail_after(1):
             assert await sigiter.__anext__() == signal.SIGUSR1
             assert await sigiter.__anext__() == signal.SIGUSR2
diff --git a/tests/test_subprocesses.py b/tests/test_subprocesses.py
index 6ef88bd..42a6cc6 100644
--- a/tests/test_subprocesses.py
+++ b/tests/test_subprocesses.py
@@ -1,17 +1,11 @@
-import os
 import platform
 import sys
-import time
-from functools import partial
 from subprocess import CalledProcessError
 from textwrap import dedent
 
 import pytest
 
-from anyio import (
-    CancelScope, create_task_group, fail_after, open_process, run_process, run_sync_in_process,
-    wait_all_tasks_blocked)
-from anyio._core._subprocesses import _process_pool_workers
+from anyio import open_process, run_process
 from anyio.streams.buffered import BufferedByteReceiveStream
 
 pytestmark = pytest.mark.anyio
@@ -69,61 +63,3 @@ async def test_terminate(tmp_path):
 
         process.terminate()
         assert await process.wait() == 2
-
-
-class TestProcessPool:
-    async def test_run_sync_in_process_pool(self):
-        """
-        Test that the function runs in a different process, and the same process in both calls.
-
-        """
-        worker_pid = await run_sync_in_process(os.getpid)
-        assert worker_pid != os.getpid()
-        assert await run_sync_in_process(os.getpid) == worker_pid
-
-    async def test_identical_sys_path(self):
-        """Test that partial() can be used to pass keyword arguments."""
-        assert await run_sync_in_process(eval, 'sys.path') == sys.path
-
-    async def test_partial(self):
-        """Test that partial() can be used to pass keyword arguments."""
-        assert await run_sync_in_process(partial(sorted, reverse=True), ['a', 'b']) == ['b', 'a']
-
-    async def test_exception(self):
-        """Test that exceptions are delivered properly."""
-        with pytest.raises(ValueError, match='invalid literal for int'):
-            assert await run_sync_in_process(int, 'a')
-
-    async def test_print(self):
-        """Test that print() won't interfere with parent-worker communication."""
-        worker_pid = await run_sync_in_process(os.getpid)
-        await run_sync_in_process(print, 'hello')
-        await run_sync_in_process(print, 'world')
-        assert await run_sync_in_process(os.getpid) == worker_pid
-
-    async def test_cancel_before(self):
-        """
-        Test that starting run_sync_in_process() in a cancelled scope does not cause a worker
-        process to be reserved.
-
-        """
-        with CancelScope() as scope:
-            scope.cancel()
-            await run_sync_in_process(os.getpid)
-
-        pytest.raises(LookupError, _process_pool_workers.get)
-
-    async def test_cancel_during(self):
-        """
-        Test that cancelling an operation on the worker process causes the process to be killed.
-
-        """
-        worker_pid = await run_sync_in_process(os.getpid)
-        with fail_after(4):
-            async with create_task_group() as tg:
-                tg.spawn(partial(run_sync_in_process, cancellable=True), time.sleep, 5)
-                await wait_all_tasks_blocked()
-                tg.cancel_scope.cancel()
-
-        # The previous worker was killed so we should get a new one now
-        assert await run_sync_in_process(os.getpid) != worker_pid
diff --git a/tests/test_synchronization.py b/tests/test_synchronization.py
index baa1ccf..10b4ed0 100644
--- a/tests/test_synchronization.py
+++ b/tests/test_synchronization.py
@@ -1,8 +1,8 @@
 import pytest
 
 from anyio import (
-    CancelScope, Condition, Event, Lock, Semaphore, WouldBlock, create_task_group,
-    current_default_worker_thread_limiter, wait_all_tasks_blocked)
+    CancelScope, Condition, Event, Lock, Semaphore, WouldBlock, create_task_group, to_thread,
+    wait_all_tasks_blocked)
 from anyio.abc import CapacityLimiter
 
 pytestmark = pytest.mark.anyio
@@ -398,7 +398,7 @@ class TestCapacityLimiter:
         assert event2.is_set()
 
     async def test_current_default_thread_limiter(self):
-        limiter = current_default_worker_thread_limiter()
+        limiter = to_thread.current_default_thread_limiter()
         assert isinstance(limiter, CapacityLimiter)
         assert limiter.total_tokens == 40
 
diff --git a/tests/test_to_process.py b/tests/test_to_process.py
new file mode 100644
index 0000000..fa65fb9
--- /dev/null
+++ b/tests/test_to_process.py
@@ -0,0 +1,82 @@
+import os
+import platform
+import sys
+import time
+from functools import partial
+
+import pytest
+
+from anyio import CancelScope, create_task_group, fail_after, to_process, wait_all_tasks_blocked
+
+pytestmark = pytest.mark.anyio
+
+
+@pytest.fixture(autouse=True)
+def check_compatibility(anyio_backend_name):
+    if anyio_backend_name == 'asyncio':
+        if platform.system() == 'Windows' and sys.version_info < (3, 8):
+            pytest.skip('Python < 3.8 uses SelectorEventLoop by default and it does not support '
+                        'subprocesses')
+
+
+async def test_run_sync_in_process_pool():
+    """
+    Test that the function runs in a different process, and the same process in both calls.
+
+    """
+    worker_pid = await to_process.run_sync(os.getpid)
+    assert worker_pid != os.getpid()
+    assert await to_process.run_sync(os.getpid) == worker_pid
+
+
+async def test_identical_sys_path():
+    """Test that partial() can be used to pass keyword arguments."""
+    assert await to_process.run_sync(eval, 'sys.path') == sys.path
+
+
+async def test_partial():
+    """Test that partial() can be used to pass keyword arguments."""
+    assert await to_process.run_sync(partial(sorted, reverse=True), ['a', 'b']) == ['b', 'a']
+
+
+async def test_exception():
+    """Test that exceptions are delivered properly."""
+    with pytest.raises(ValueError, match='invalid literal for int'):
+        assert await to_process.run_sync(int, 'a')
+
+
+async def test_print():
+    """Test that print() won't interfere with parent-worker communication."""
+    worker_pid = await to_process.run_sync(os.getpid)
+    await to_process.run_sync(print, 'hello')
+    await to_process.run_sync(print, 'world')
+    assert await to_process.run_sync(os.getpid) == worker_pid
+
+
+async def test_cancel_before():
+    """
+    Test that starting to_process.run_sync() in a cancelled scope does not cause a worker
+    process to be reserved.
+
+    """
+    with CancelScope() as scope:
+        scope.cancel()
+        await to_process.run_sync(os.getpid)
+
+    pytest.raises(LookupError, to_process._process_pool_workers.get)
+
+
+async def test_cancel_during():
+    """
+    Test that cancelling an operation on the worker process causes the process to be killed.
+
+    """
+    worker_pid = await to_process.run_sync(os.getpid)
+    with fail_after(4):
+        async with create_task_group() as tg:
+            tg.spawn(partial(to_process.run_sync, cancellable=True), time.sleep, 5)
+            await wait_all_tasks_blocked()
+            tg.cancel_scope.cancel()
+
+    # The previous worker was killed so we should get a new one now
+    assert await to_process.run_sync(os.getpid) != worker_pid
diff --git a/tests/test_to_thread.py b/tests/test_to_thread.py
new file mode 100644
index 0000000..322d411
--- /dev/null
+++ b/tests/test_to_thread.py
@@ -0,0 +1,123 @@
+import asyncio
+import sys
+import threading
+import time
+
+import pytest
+
+from anyio import (
+    CapacityLimiter, Event, create_task_group, from_thread, sleep, to_thread,
+    wait_all_tasks_blocked)
+
+if sys.version_info < (3, 7):
+    current_task = asyncio.Task.current_task
+else:
+    current_task = asyncio.current_task
+
+pytestmark = pytest.mark.anyio
+
+
+async def test_run_in_thread_cancelled():
+    def thread_worker():
+        nonlocal state
+        state = 2
+
+    async def worker():
+        nonlocal state
+        state = 1
+        await to_thread.run_sync(thread_worker)
+        state = 3
+
+    state = 0
+    async with create_task_group() as tg:
+        tg.spawn(worker)
+        tg.cancel_scope.cancel()
+
+    assert state == 1
+
+
+async def test_run_in_thread_exception():
+    def thread_worker():
+        raise ValueError('foo')
+
+    with pytest.raises(ValueError) as exc:
+        await to_thread.run_sync(thread_worker)
+
+    exc.match('^foo$')
+
+
+async def test_run_in_custom_limiter():
+    def thread_worker():
+        nonlocal num_active_threads, max_active_threads
+        num_active_threads += 1
+        max_active_threads = max(num_active_threads, max_active_threads)
+        event.wait(1)
+        num_active_threads -= 1
+
+    async def task_worker():
+        await to_thread.run_sync(thread_worker, limiter=limiter)
+
+    event = threading.Event()
+    num_active_threads = max_active_threads = 0
+    limiter = CapacityLimiter(3)
+    async with create_task_group() as tg:
+        for _ in range(4):
+            tg.spawn(task_worker)
+
+        await sleep(0.1)
+        assert num_active_threads == 3
+        assert limiter.borrowed_tokens == 3
+        event.set()
+
+    assert num_active_threads == 0
+    assert max_active_threads == 3
+
+
+@pytest.mark.parametrize('cancellable, expected_last_active', [
+    (False, 'task'),
+    (True, 'thread')
+], ids=['uncancellable', 'cancellable'])
+async def test_cancel_worker_thread(cancellable, expected_last_active):
+    """
+    Test that when a task running a worker thread is cancelled, the cancellation is not acted on
+    until the thread finishes.
+
+    """
+    def thread_worker():
+        nonlocal last_active
+        from_thread.run_sync(sleep_event.set)
+        time.sleep(0.2)
+        last_active = 'thread'
+        from_thread.run_sync(finish_event.set)
+
+    async def task_worker():
+        nonlocal last_active
+        try:
+            await to_thread.run_sync(thread_worker, cancellable=cancellable)
+        finally:
+            last_active = 'task'
+
+    sleep_event = Event()
+    finish_event = Event()
+    last_active = None
+    async with create_task_group() as tg:
+        tg.spawn(task_worker)
+        await sleep_event.wait()
+        tg.cancel_scope.cancel()
+
+    await finish_event.wait()
+    assert last_active == expected_last_active
+
+
+@pytest.mark.parametrize('anyio_backend', ['asyncio'])
+async def test_cancel_asyncio_native_task():
+    async def run_in_thread():
+        nonlocal task
+        task = current_task()
+        await to_thread.run_sync(time.sleep, 1, cancellable=True)
+
+    task = None
+    async with create_task_group() as tg:
+        tg.spawn(run_in_thread)
+        await wait_all_tasks_blocked()
+        task.cancel()