Added TaskGroup.start_soon() and changed spawn() back into a coroutine function (#254)

* Added TaskGroup.start_soon() and changed spawn() back into a coroutine function
* Renamed BlockingPortal.spawn_task to start_task_soon

One more step towards a trio-like API.

Author: agronholm

docs/cancellation.rst

@@ -69,7 +69,7 @@ To accomplish this, open a new cancel scope with the ``shield=True`` argument::
     async def main():
         async with create_task_group() as tg:
             with CancelScope(shield=True) as scope:
-                tg.spawn(external_task)
+                tg.start_soon(external_task)
                 tg.cancel_scope.cancel()
                 print('Started sleeping in the host task')
                 await sleep(1)
@@ -146,7 +146,7 @@ Therefore, do **NOT** do this::
 
     async def some_generator():
         async with create_task_group() as tg:
-            tg.spawn(foo)
+            tg.start_soon(foo)
             yield
 
 The problem with this code is that it violates structural concurrency: what happens if the spawned

docs/migration.rst

@@ -4,6 +4,7 @@ Migrating from AnyIO 2 to AnyIO 3
 .. py:currentmodule:: anyio
 
 AnyIO 3 changed some functions and methods in a way that needs some adaptation in your code.
+All deprecated functions and methods will be removed in AnyIO 4.
 
 Asynchronous functions converted to synchronous
 -----------------------------------------------
@@ -31,7 +32,6 @@ The following functions and methods were changed:
 * :meth:`MemoryObjectSendStream.send_nowait() <.streams.memory.MemoryObjectSendStream.send_nowait>`
 * :func:`open_signal_receiver`
 * :meth:`Semaphore.release`
-* :meth:`TaskGroup.spawn() <.abc.TaskGroup.spawn>`
 
 When migrating to AnyIO 3, simply remove the ``await`` from each call to these.
 
@@ -81,6 +81,26 @@ Example 2 – opening a cancel scope::
 
 .. _trio: https://github.com/python-trio/trio
 
+Starting tasks
+--------------
+
+The :meth:`TaskGroup.spawn` coroutine method has been deprecated in favor of the synchronous
+method :meth:`TaskGroup.start_soon` (which mirrors ``start_soon()`` in trio's nurseries). If you're
+fully migrating to AnyIO 3, simply switch to calling the new method (and remove the ``await``).
+
+The :meth:`BlockingPortal.spawn_task` method has also been renamed to
+:meth:`~BlockingPortal.start_task_soon`, so as to be consistent with task groups.
+
+If your code needs to work with both AnyIO 2 and 3, you can keep using :meth:`~TaskGroup.spawn`
+(until AnyIO 4) and suppress the deprecation warning::
+
+    import warnings
+
+    async def foo():
+        async with create_task_group() as tg:
+            with warnings.catch_warnings():
+                await tg.spawn(otherfunc)
+
 Synchronization primitives
 --------------------------
 

docs/signals.rst

@@ -60,7 +60,7 @@ signal handler::
 
     async def main():
         async with create_task_group() as tg:
-            tg.spawn(signal_handler, tg.cancel_scope)
+            tg.start_soon(signal_handler, tg.cancel_scope)
             ...  # proceed with starting the actual application logic
 
     run(main)

docs/streams.rst

@@ -60,7 +60,7 @@ Example::
     async def main():
         send_stream, receive_stream = create_memory_object_stream()
         async with create_task_group() as tg:
-            tg.spawn(process_items, receive_stream)
+            tg.start_soon(process_items, receive_stream)
             async with send_stream:
                 for num in range(10):
                     await send_stream.send(f'number {num}')

docs/synchronization.rst

@@ -25,7 +25,7 @@ Example::
     async def main():
         event = Event()
         async with create_task_group() as tg:
-            tg.spawn(notify, event)
+            tg.start_soon(notify, event)
             await event.wait()
             print('Received notification!')
 
@@ -58,7 +58,7 @@ Example::
         semaphore = Semaphore(2)
         async with create_task_group() as tg:
             for num in range(10):
-                tg.spawn(use_resource, num, semaphore)
+                tg.start_soon(use_resource, num, semaphore)
 
     run(main)
 
@@ -84,7 +84,7 @@ Example::
         lock = Lock()
         async with create_task_group() as tg:
             for num in range(4):
-                tg.spawn(use_resource, num, lock)
+                tg.start_soon(use_resource, num, lock)
 
     run(main)
 
@@ -115,7 +115,7 @@ Example::
         condition = Condition()
         async with create_task_group() as tg:
             for tasknum in range(6):
-                tg.spawn(listen, tasknum, condition)
+                tg.start_soon(listen, tasknum, condition)
 
             await sleep(1)
             async with condition:
@@ -153,7 +153,7 @@ Example::
         limiter = CapacityLimiter(2)
         async with create_task_group() as tg:
             for num in range(10):
-                tg.spawn(use_resource, num, limiter)
+                tg.start_soon(use_resource, num, limiter)
 
     run(main)
 

docs/tasks.rst

@@ -30,7 +30,7 @@ Here's a demonstration::
     async def main():
         async with create_task_group() as tg:
             for num in range(5):
-                tg.spawn(sometask, num)
+                tg.start_soon(sometask, num)
 
         print('All tasks finished!')
 
@@ -77,7 +77,7 @@ calling will :meth:`TaskGroup.start() <.abc.TaskGroup.start>` will be blocked un
 spawned task never calls it, then the :meth:`TaskGroup.start() <.abc.TaskGroup.start>` call will
 raise a ``RuntimeError``.
 
-.. note:: Unlike :meth:`~.abc.TaskGroup.spawn`, :meth:`~.abc.TaskGroup.start` needs an ``await``.
+.. note:: Unlike :meth:`~.abc.TaskGroup.start_soon`, :meth:`~.abc.TaskGroup.start` needs an ``await``.
 
 Handling multiple errors in a task group
 ----------------------------------------

docs/threads.rst

@@ -85,7 +85,7 @@ Spawning tasks from worker threads
 ----------------------------------
 
 When you need to spawn a task to be run in the background, you can do so using
-:meth:`~.BlockingPortal.spawn_task`::
+:meth:`~.BlockingPortal.start_task_soon`::
 
     from concurrent.futures import as_completed
 
@@ -100,7 +100,7 @@ When you need to spawn a task to be run in the background, you can do so using
 
 
     with start_blocking_portal() as portal:
-        futures = [portal.spawn_task(long_running_task, i) for i in range(1, 5)]
+        futures = [portal.start_task_soon(long_running_task, i) for i in range(1, 5)]
         for future in as_completed(futures):
             print(future.result())
 

docs/versionhistory.rst

@@ -6,6 +6,9 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
 **UNRELEASED**
 
 - Fixed ``Semaphore.acquire()`` raising ``WouldBlock`` when a race condition occurs
+- Changed ``TaskGroup.spawn()`` back into a coroutine function and added
+  ``TaskGroup.start_soon`` as the replacement
+- Renamed ``BlockingPortal.spawn_task`` into ``BlockingPortal.start_task_soon``
 
 **3.0.0rc3**
 

src/anyio/_backends/_asyncio.py

@@ -623,9 +623,8 @@ def task_done(_task: asyncio.Task) -> None:
         self.cancel_scope._tasks.add(task)
         return task
 
-    def spawn(self, func: Callable[..., Coroutine], *args, name=None) -> DeprecatedAwaitable:
+    def start_soon(self, func: Callable[..., Coroutine], *args, name=None) -> None:
         self._spawn(func, args, name)
-        return DeprecatedAwaitable(self.spawn)
 
     async def start(self, func: Callable[..., Coroutine], *args, name=None) -> None:
         future: asyncio.Future = asyncio.Future()
@@ -766,7 +765,7 @@ def __init__(self):
     def _spawn_task_from_thread(self, func: Callable, args: tuple, kwargs: Dict[str, Any],
                                 name, future: Future) -> None:
         run_sync_from_thread(
-            partial(self._task_group.spawn, name=name), self._call_func, func, args, kwargs,
+            partial(self._task_group.start_soon, name=name), self._call_func, func, args, kwargs,
             future, loop=self._loop)
 
 

src/anyio/_backends/_trio.py

@@ -132,12 +132,11 @@ async def __aexit__(self, exc_type: Optional[Type[BaseException]],
         finally:
             self._active = False
 
-    def spawn(self, func: Callable, *args, name=None) -> DeprecatedAwaitable:
+    def start_soon(self, func: Callable, *args, name=None) -> None:
         if not self._active:
             raise RuntimeError('This task group is not active; no new tasks can be spawned.')
 
         self._nursery.start_soon(func, *args, name=name)
-        return DeprecatedAwaitable(self.spawn)
 
     async def start(self, func: Callable[..., Coroutine], *args, name=None):
         if not self._active:
@@ -171,7 +170,7 @@ def __init__(self):
     def _spawn_task_from_thread(self, func: Callable, args: tuple, kwargs: Dict[str, Any],
                                 name, future: Future) -> None:
         return trio.from_thread.run_sync(
-            partial(self._task_group.spawn, name=name), self._call_func, func, args, kwargs,
+            partial(self._task_group.start_soon, name=name), self._call_func, func, args, kwargs,
             future, trio_token=self._token)
 
 

src/anyio/_core/_sockets.py

@@ -173,7 +173,7 @@ async def try_connect(remote_host: str, event: Event):
     async with create_task_group() as tg:
         for i, (af, addr) in enumerate(target_addrs):
             event = Event()
-            tg.spawn(try_connect, addr, event)
+            tg.start_soon(try_connect, addr, event)
             with move_on_after(happy_eyeballs_delay):
                 await event.wait()
 

src/anyio/_core/_subprocesses.py

@@ -37,9 +37,9 @@ async def drain_stream(stream, index):
         try:
             async with create_task_group() as tg:
                 if process.stdout:
-                    tg.spawn(drain_stream, process.stdout, 0)
+                    tg.start_soon(drain_stream, process.stdout, 0)
                 if process.stderr:
-                    tg.spawn(drain_stream, process.stderr, 1)
+                    tg.start_soon(drain_stream, process.stderr, 1)
                 if process.stdin and input:
                     await process.stdin.send(input)
                     await process.stdin.aclose()

src/anyio/abc/_sockets.py

@@ -127,7 +127,7 @@ async def serve(self, handler: Callable[[T_Stream], Any],
         async with context_manager:
             while True:
                 stream = await self.accept()
-                task_group.spawn(handler, stream)
+                task_group.start_soon(handler, stream)
 
 
 class UDPSocket(UnreliableObjectStream[UDPPacketType], _SocketProvider):

src/anyio/abc/_tasks.py

@@ -2,8 +2,7 @@
 from abc import ABCMeta, abstractmethod
 from types import TracebackType
 from typing import Callable, Coroutine, Optional, Type, TypeVar
-
-from anyio._core._compat import DeprecatedAwaitable
+from warnings import warn
 
 if typing.TYPE_CHECKING:
     from anyio._core._tasks import CancelScope
@@ -31,14 +30,33 @@ class TaskGroup(metaclass=ABCMeta):
 
     cancel_scope: 'CancelScope'
 
+    async def spawn(self, func: Callable[..., Coroutine], *args, name=None) -> None:
+        """
+        Deprecated alios for :meth:`start_soon`.
+
+        :param func: a coroutine function
+        :param args: positional arguments to call the function with
+        :param name: name of the task, for the purposes of introspection and debugging
+
+        .. deprecated:: 3.0
+           Use :meth:`start_soon` instead. If your code needs AnyIO 2 compatibility, you
+           can keep using this until AnyIO 4.
+
+        """
+        warn('spawn() is deprecated -- use start_soon() (without the "await") instead',
+             DeprecationWarning)
+        self.start_soon(func, *args, name=name)
+
     @abstractmethod
-    def spawn(self, func: Callable[..., Coroutine], *args, name=None) -> DeprecatedAwaitable:
+    def start_soon(self, func: Callable[..., Coroutine], *args, name=None) -> None:
         """
         Launch a new task in this task group.
 
         :param func: a coroutine function
         :param args: positional arguments to call the function with
         :param name: name of the task, for the purposes of introspection and debugging
+
+        .. versionadded:: 3.0
         """
 
     @abstractmethod

src/anyio/from_thread.py

@@ -90,7 +90,7 @@ async def run_async_cm(self):
 
     def __enter__(self) -> T_co:
         self._enter_future = Future()
-        self._exit_future = self._portal.spawn_task(self.run_async_cm)
+        self._exit_future = self._portal.start_task_soon(self.run_async_cm)
         cm = self._enter_future.result()
         return cast(T_co, cm)
 
@@ -219,9 +219,30 @@ def call(self, func, *args):
             the event loop thread
 
         """
-        return self.spawn_task(func, *args).result()
+        return self.start_task_soon(func, *args).result()
 
     def spawn_task(self, func: Callable[..., Coroutine], *args, name=None) -> Future:
+        """
+        Deprecated alias for :meth:`start_task_soon`.
+
+        :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
+        .. deprecated:: 3.0
+           Use :meth:`start_task_soon` instead. If your code needs AnyIO 2 compatibility, you
+           can keep using this until AnyIO 4.
+
+        """
+        warn('spawn_task() is deprecated -- use start_task_soon() instead', DeprecationWarning)
+        return self.start_task_soon(func, *args, name=name)
+
+    def start_task_soon(self, func: Callable[..., Coroutine], *args, name=None) -> Future:
         """
         Spawn a task in the portal's task group.
 
@@ -236,9 +257,7 @@ def spawn_task(self, func: Callable[..., Coroutine], *args, name=None) -> Future
         :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.
+        .. versionadded:: 3.0
 
         """
         self._check_running()

src/anyio/streams/stapled.py

@@ -109,7 +109,7 @@ async def serve(self, handler: Callable[[T_Stream], Any],
 
         async with create_task_group() as tg:
             for listener in self.listeners:
-                tg.spawn(listener.serve, handler, task_group)
+                tg.start_soon(listener.serve, handler, task_group)
 
     async def aclose(self) -> None:
         for listener in self.listeners: