Expand cancellation usability from native trio threads

docs/source/reference-core.rst

@@ -1823,8 +1823,22 @@ to spawn a child thread, and then use a :ref:`memory channel
 
 .. literalinclude:: reference-core/from-thread-example.py
 
-You can also perform a non-blocking check for cancellation from threads spawned
-by `trio.to_thread.run_sync`.
+.. note::
+
+ The ``from_thread.run*`` functions reuse the host task that called
+ :func:`trio.to_thread.run_sync` to run your provided function in the typical case,
+ namely when ``cancellable=False`` so Trio can be sure that the task will always be
+ around to perform the work. If you pass ``cancellable=True`` at the outset, or if
+ you provide a :class:`~trio.lowlevel.TrioToken` when calling back in to Trio, your
+ functions will be executed in a new system task. Therefore, the
+ :func:`~trio.lowlevel.current_task`, :func:`current_effective_deadline`, or other
+ task-tree specific values may differ depending on keyword argument values.
+
+You can also use :func:`trio.from_thread.check_cancelled` to check for cancellation from
+a thread that was spawned by :func:`trio.to_thread.run_sync`. If the call to
+:func:`~trio.to_thread.run_sync` was cancelled (even if ``cancellable=False``!), then
+:func:`~trio.from_thread.check_cancelled` will raise :func:`trio.Cancelled`.
+It's like ``trio.from_thread.run(trio.sleep, 0)``, but much faster.
 
 .. autofunction:: trio.from_thread.check_cancelled
 

newsfragments/2392.feature.rst

@@ -1 +1,5 @@
-If called from a thread spawned by `trio.to_thread.run_sync`, `trio.from_thread.run` and `trio.from_thread.run_sync` now reuse the task and cancellation status of the host task and have `trio.from_thread.check_cancelled` to efficiently poll for cancellation.
+If called from a thread spawned by `trio.to_thread.run_sync`, `trio.from_thread.run` and
+`trio.from_thread.run_sync` now reuse the task and cancellation status of the host task;
+this means that context variables and cancel scopes naturally propagate 'through'
+threads spawned by Trio. You can also use `trio.from_thread.check_cancelled`
+to efficiently check for cancellation without reentering the Trio thread.

trio/_tests/test_threads.py

@@ -644,7 +644,7 @@ async def async_fn(): # pragma: no cover
  def thread_fn():
  from_thread_run_sync(async_fn)
 
- with pytest.raises(TypeError, match="expected a sync function"):
+ with pytest.raises(TypeError, match="expected a synchronous function"):
  await to_thread_run_sync(thread_fn)
 
 
@@ -813,19 +813,23 @@ async def agen(token):
  try:
  yield
  finally:
- with pytest.raises(_core.RunFinishedError), _core.CancelScope(shield=True):
- await to_thread_run_sync(
- partial(from_thread_run, sleep, 0, trio_token=token)
- )
- record.append("ok")
+ with _core.CancelScope(shield=True):
+ try:
+ await to_thread_run_sync(
+ partial(from_thread_run, sleep, 0, trio_token=token)
+ )
+ except _core.RunFinishedError:
+ record.append("finished")
+ else:
+ record.append("clean")
 
  async def main(use_system_task):
  save.append(agen(_core.current_trio_token() if use_system_task else None))
  await save[-1].asend(None)
 
  _core.run(main, True) # System nursery will be closed and raise RunFinishedError
- _core.run(main, False) # host task will not be rescheduled
- assert record == ["ok"]
+ _core.run(main, False) # host task will be rescheduled as normal
+ assert record == ["finished", "clean"]
 
 
 async def test_trio_token_weak_referenceable():
@@ -874,8 +878,8 @@ def sync_check():
  from_thread_run_sync(cancel_scope.cancel)
  try:
  from_thread_run_sync(bool)
- except _core.Cancelled:
- queue.put(True)
+ except _core.Cancelled: # pragma: no cover
+ queue.put(True) # sync functions don't raise Cancelled
  else:
  queue.put(False)
 
@@ -889,7 +893,7 @@ def sync_check():
  await to_thread_run_sync(sync_check, cancellable=True)
 
  assert cancel_scope.cancelled_caught
- assert await to_thread_run_sync(partial(queue.get, timeout=1))
+ assert not await to_thread_run_sync(partial(queue.get, timeout=1))
 
  async def no_checkpoint():
  return True
@@ -898,8 +902,8 @@ def async_check():
  from_thread_run_sync(cancel_scope.cancel)
  try:
  assert from_thread_run(no_checkpoint)
- except _core.Cancelled:
- queue.put(True)
+ except _core.Cancelled: # pragma: no cover
+ queue.put(True) # async functions raise Cancelled at checkpoints
  else:
  queue.put(False)
 
@@ -913,7 +917,7 @@ def async_check():
  await to_thread_run_sync(async_check, cancellable=True)
 
  assert cancel_scope.cancelled_caught
- assert await to_thread_run_sync(partial(queue.get, timeout=1))
+ assert not await to_thread_run_sync(partial(queue.get, timeout=1))
 
  async def async_time_bomb():
  cancel_scope.cancel()
@@ -929,14 +933,16 @@ async def async_time_bomb():
 async def test_from_thread_check_cancelled():
  q = stdlib_queue.Queue()
 
- async def child(cancellable):
- record.append("start")
- try:
- return await to_thread_run_sync(f, cancellable=cancellable)
- except _core.Cancelled:
- record.append("cancel")
- finally:
- record.append("exit")
+ async def child(cancellable, scope):
+ with scope:
+ record.append("start")
+ try:
+ return await to_thread_run_sync(f, cancellable=cancellable)
+ except _core.Cancelled:
+ record.append("cancel")
+ raise
+ finally:
+ record.append("exit")
 
  def f():
  try:
@@ -952,7 +958,7 @@ def f():
  record = []
  ev = threading.Event()
  async with _core.open_nursery() as nursery:
- nursery.start_soon(child, False)
+ nursery.start_soon(child, False, _core.CancelScope())
  await wait_all_tasks_blocked()
  assert record[0] == "start"
  assert q.get(timeout=1) == "Not Cancelled"
@@ -964,14 +970,15 @@ def f():
  # the appropriate cancel scope
  record = []
  ev = threading.Event()
+ scope = _core.CancelScope() # Nursery cancel scope gives false positives
  async with _core.open_nursery() as nursery:
- nursery.start_soon(child, False)
+ nursery.start_soon(child, False, scope)
  await wait_all_tasks_blocked()
  assert record[0] == "start"
  assert q.get(timeout=1) == "Not Cancelled"
- nursery.cancel_scope.cancel()
+ scope.cancel()
  ev.set()
- assert nursery.cancel_scope.cancelled_caught
+ assert scope.cancelled_caught
  assert "cancel" in record
  assert record[-1] == "exit"
 
@@ -988,13 +995,14 @@ def f(): # noqa: F811
 
  record = []
  ev = threading.Event()
+ scope = _core.CancelScope()
  async with _core.open_nursery() as nursery:
- nursery.start_soon(child, True)
+ nursery.start_soon(child, True, scope)
  await wait_all_tasks_blocked()
  assert record[0] == "start"
- nursery.cancel_scope.cancel()
+ scope.cancel()
  ev.set()
- assert nursery.cancel_scope.cancelled_caught
+ assert scope.cancelled_caught
  assert "cancel" in record
  assert record[-1] == "exit"
  assert q.get(timeout=1) == "Cancelled"

trio/_tests/verify_types_darwin.json

@@ -40,7 +40,7 @@
  ],
  "exportedSymbolCounts": {
  "withAmbiguousType": 0,
- "withKnownType": 632,
+ "withKnownType": 631,
  "withUnknownType": 0
  },
  "ignoreUnknownTypesFromImports": true,

trio/_tests/verify_types_linux.json

@@ -28,7 +28,7 @@
  ],
  "exportedSymbolCounts": {
  "withAmbiguousType": 0,
- "withKnownType": 629,
+ "withKnownType": 628,
  "withUnknownType": 0
  },
  "ignoreUnknownTypesFromImports": true,

trio/_tests/verify_types_windows.json

@@ -7,7 +7,7 @@
  "warningCount": 0
  },
  "typeCompleteness": {
- "completenessScore": 0.9857594936708861,
+ "completenessScore": 1,
  "diagnostics": [
  {
  "message": "No docstring found for function \"trio.lowlevel.current_iocp\"",
@@ -64,8 +64,8 @@
  ],
  "exportedSymbolCounts": {
  "withAmbiguousType": 0,
- "withKnownType": 623,
- "withUnknownType": 9
+ "withKnownType": 631,
+ "withUnknownType": 0
  },
  "ignoreUnknownTypesFromImports": true,
  "missingClassDocStringCount": 1,