Merge pull request #1156 from njsmith/add-to-thread

Add new trio.to_thread module for running worker threads

Author: epellis

docs/source/history.rst

@@ -655,7 +655,7 @@ CPython, or PyPy3 5.9+.
 Other changes
 ~~~~~~~~~~~~~
 
-* :func:`run_sync_in_thread` now has a :ref:`robust mechanism
+* ``run_sync_in_worker_thread`` now has a :ref:`robust mechanism
   for applying capacity limits to the number of concurrent threads
   <worker-thread-limiting>` (`#10
   <https://github.com/python-trio/trio/issues/170>`__, `#57

docs/source/reference-core.rst

@@ -1471,7 +1471,7 @@ In acknowledgment of this reality, Trio provides two useful utilities
 for working with real, operating-system level,
 :mod:`threading`\-module-style threads. First, if you're in Trio but
 need to push some blocking I/O into a thread, there's
-:func:`run_sync_in_thread`. And if you're in a thread and need
+`trio.to_thread.run_sync`. And if you're in a thread and need
 to communicate back with Trio, you can use
 :func:`trio.from_thread.run` and :func:`trio.from_thread.run_sync`.
 
@@ -1494,7 +1494,7 @@ are spawned and the system gets overloaded and crashes. Instead, the N
 threads start executing the first N jobs, while the other
 (100,000 - N) jobs sit in a queue and wait their turn. Which is
 generally what you want, and this is how
-:func:`trio.run_sync_in_thread` works by default.
+:func:`trio.to_thread.run_sync` works by default.
 
 The downside of this kind of thread pool is that sometimes, you need
 more sophisticated logic for controlling how many threads are run at
@@ -1541,16 +1541,16 @@ re-using threads, but has no admission control policy: if you give it
 responsible for providing the policy to make sure that this doesn't
 happen – but since it *only* has to worry about policy, it can be much
 simpler. In fact, all there is to it is the ``limiter=`` argument
-passed to :func:`run_sync_in_thread`. This defaults to a global
+passed to :func:`trio.to_thread.run_sync`. This defaults to a global
 :class:`CapacityLimiter` object, which gives us the classic fixed-size
 thread pool behavior. (See
-:func:`current_default_thread_limiter`.) But if you want to use
-"separate pools" for type A jobs and type B jobs, then it's just a
-matter of creating two separate :class:`CapacityLimiter` objects and
-passing them in when running these jobs. Or here's an example of
-defining a custom policy that respects the global thread limit, while
-making sure that no individual user can use more than 3 threads at a
-time::
+:func:`trio.to_thread.current_default_thread_limiter`.) But if you
+want to use "separate pools" for type A jobs and type B jobs, then
+it's just a matter of creating two separate :class:`CapacityLimiter`
+objects and passing them in when running these jobs. Or here's an
+example of defining a custom policy that respects the global thread
+limit, while making sure that no individual user can use more than 3
+threads at a time::
 
    class CombinedLimiter:
         def __init__(self, first, second):
@@ -1594,19 +1594,24 @@ time::
            return combined_limiter
 
 
-   async def run_in_worker_thread_for_user(user_id, async_fn, *args, **kwargs):
-       # *args belong to async_fn; **kwargs belong to run_sync_in_thread
+   async def run_sync_in_thread_for_user(user_id, sync_fn, *args):
        kwargs["limiter"] = get_user_limiter(user_id)
-       return await trio.run_sync_in_thread(asycn_fn, *args, **kwargs)
+       return await trio.to_thread.run_sync(asycn_fn, *args)
 
 
+.. module:: trio.to_thread
+.. currentmodule:: trio
+
 Putting blocking I/O into worker threads
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
-.. autofunction:: run_sync_in_thread
+.. autofunction:: trio.to_thread.run_sync
+
+.. autofunction:: trio.to_thread.current_default_thread_limiter
 
-.. autofunction:: current_default_thread_limiter
 
+.. module:: trio.from_thread
+.. currentmodule:: trio
 
 Getting back into the Trio thread from another thread
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
@@ -1615,6 +1620,7 @@ Getting back into the Trio thread from another thread
 
 .. autofunction:: trio.from_thread.run_sync
 
+
 This will probably be clearer with an example. Here we demonstrate how
 to spawn a child thread, and then use a :ref:`memory channel
 <channels>` to send messages between the thread and a Trio task:

docs/source/reference-core/from-thread-example.py

@@ -23,7 +23,7 @@ async def main():
         # In a background thread, run:
         #   thread_fn(portal, receive_from_trio, send_to_trio)
         nursery.start_soon(
-            trio.run_sync_in_thread, thread_fn, receive_from_trio, send_to_trio
+            trio.to_thread.run_sync, thread_fn, receive_from_trio, send_to_trio
         )
 
         # prints "1"

docs/source/reference-hazmat.rst

@@ -403,8 +403,8 @@ This logic is a bit convoluted, but accomplishes all of the following:
   loop outside of the ``except BlockingIOError:`` block.
 
 These functions can also be useful in other situations. For example,
-when :func:`trio.run_sync_in_thread` schedules some work to run
-in a worker thread, it blocks until the work is finished (so it's a
+when :func:`trio.to_thread.run_sync` schedules some work to run in a
+worker thread, it blocks until the work is finished (so it's a
 schedule point), but by default it doesn't allow cancellation. So to
 make sure that the call always acts as a checkpoint, it calls
 :func:`checkpoint_if_cancelled` before starting the thread.

docs/source/reference-io.rst

@@ -492,7 +492,7 @@ To understand why, you need to know two things.
 First, right now no mainstream operating system offers a generic,
 reliable, native API for async file or filesystem operations, so we
 have to fake it by using threads (specifically,
-:func:`run_sync_in_thread`). This is cheap but isn't free: on a
+:func:`trio.to_thread.run_sync`). This is cheap but isn't free: on a
 typical PC, dispatching to a worker thread adds something like ~100 µs
 of overhead to each operation. ("µs" is pronounced "microseconds", and
 there are 1,000,000 µs in a second. Note that all the numbers here are

newsfragments/810.removal.rst

@@ -1,5 +1,5 @@
-``run_sync_in_worker_thread`` was too much of a mouthful – now it's
-just called `run_sync_in_thread` (though the old name still works with
-a deprecation warning, for now). Similarly,
-``current_default_worker_thread_limiter`` is becoming
-`current_default_thread_limiter`.
+``run_sync_in_worker_thread`` has become `trio.to_thread.run_sync`, in
+order to make it shorter, and more consistent with the new
+``trio.from_thread``. And ``current_default_worker_thread_limiter`` is
+now `trio.to_thread.current_default_thread_limiter`. (Of course the
+old names still work with a deprecation warning, for now.)

notes-to-self/blocking-read-hack.py

@@ -29,7 +29,7 @@ async def kill_it_after_timeout(new_fd):
         async with trio.open_nursery() as nursery:
             nursery.start_soon(kill_it_after_timeout, new_fd)
             try:
-                data = await trio.run_sync_in_thread(os.read, new_fd, count)
+                data = await trio.to_thread.run_sync(os.read, new_fd, count)
             except OSError as exc:
                 if cancel_requested and exc.errno == errno.ENOTCONN:
                     # Call was successfully cancelled. In a real version we'd

notes-to-self/thread-dispatch-bench.py

@@ -2,7 +2,7 @@
 # minimal a fashion as possible.
 #
 # This is useful to get a sense of the *lower-bound* cost of
-# run_sync_in_thread
+# trio.to_thread.run_sync
 
 import threading
 from queue import Queue

trio/__init__.py

@@ -32,7 +32,6 @@
     Event, CapacityLimiter, Semaphore, Lock, StrictFIFOLock, Condition
 )
 
-from ._threads import (run_sync_in_thread, current_default_thread_limiter)
 from ._threads import BlockingTrioPortal as _BlockingTrioPortal
 
 from ._highlevel_generic import aclose_forcefully, StapledStream
@@ -67,12 +66,14 @@
 
 from ._deprecate import TrioDeprecationWarning
 
-# Imported by default
+# Submodules imported by default
 from . import hazmat
 from . import socket
 from . import abc
 from . import from_thread
-# Not imported by default: testing
+from . import to_thread
+# Not imported by default, but mentioned here so static analysis tools like
+# pylint will know that it exists.
 if False:
     from . import testing
 
@@ -104,13 +105,13 @@
         ),
     "run_sync_in_worker_thread":
         _deprecate.DeprecatedAttribute(
-            run_sync_in_thread,
+            to_thread.run_sync,
             "0.12.0",
             issue=810,
         ),
     "current_default_worker_thread_limiter":
         _deprecate.DeprecatedAttribute(
-            current_default_thread_limiter,
+            to_thread.current_default_thread_limiter,
             "0.12.0",
             issue=810,
         ),
@@ -163,6 +164,7 @@
 fixup_module_metadata(socket.__name__, socket.__dict__)
 fixup_module_metadata(abc.__name__, abc.__dict__)
 fixup_module_metadata(from_thread.__name__, from_thread.__dict__)
+fixup_module_metadata(to_thread.__name__, to_thread.__dict__)
 fixup_module_metadata(__name__ + ".ssl", _deprecated_ssl_reexports.__dict__)
 fixup_module_metadata(
     __name__ + ".subprocess", _deprecated_subprocess_reexports.__dict__

trio/_core/_entry_queue.py

@@ -131,10 +131,10 @@ class TrioToken:
     This object has two uses:
 
     1. It lets you re-enter the Trio run loop from external threads or signal
-       handlers. This is the low-level primitive that
-       :func:`trio.run_sync_in_thread` uses to receive results from
-       worker threads, that :func:`trio.open_signal_receiver` uses to receive
-       notifications about signals, and so forth.
+       handlers. This is the low-level primitive that :func:`trio.to_thread`
+       and `trio.from_thread` use to communicate with worker threads, that
+       `trio.open_signal_receiver` uses to receive notifications about
+       signals, and so forth.
 
     2. Each call to :func:`trio.run` has exactly one associated
        :class:`TrioToken` object, so you can use it to identify a particular