Sharing a dynamically started worker among several consumers

user_3212865 0
Viewed 126

I am building a worker class that connects to an external event stream using asyncio. It is a single stream, but several consumer may enable it. The goal is to only maintain the connection while one or more consumer requires it.

My requirements are as follow:

  • The worker instance is created dynamically first time a consumer requires it.
  • When other consumers then require it, they re-use the same worker instance.
  • When the last consumer closes the stream, it cleans up its resources.

This sounds easy enough. However, the startup sequence is causing me issues, because it is itself asynchronous. Thus, assuming this interface:

class Stream:
    async def start(self, *, timeout=DEFAULT_TIMEOUT):
        pass
    async def stop(self):
        pass

I have the following scenarios:

Scenario 1 - exception at startup

  • Consumer 1 requests worker to start.
  • Worker startup sequence begins
  • Consumer 2 requests worker to start.
  • Worker startup sequence raises an exception.
  • Both consumer should see the exception as the result of their call to start().

Scenario 2 - partial asynchronous cancellation

  • Consumer 1 requests worker to start.
  • Worker startup sequence begins
  • Consumer 2 requests worker to start.
  • Consumer 1 gets cancelled.
  • Worker startup sequence completes.
  • Consumer 2 should see a successful start.

Scenario 3 - complete asynchronous cancellation

  • Consumer 1 requests worker to start.
  • Worker startup sequence begins
  • Consumer 2 requests worker to start.
  • Consumer 1 gets cancelled.
  • Consumer 2 gets cancelled.
  • Worker startup sequence must be cancelled as a result.

I struggle to cover all scenarios without getting any race condition and a spaghetti mess of either bare Future or Event objects.

Here is an attempt at writing start(). It relies on _worker() setting an asyncio.Event named self._worker_ready when it completes the startup sequence:

async def start(self, timeout=None):
    assert not self.closing
    if not self._task:
        self._task = asyncio.ensure_future(self._worker())

    # Wait until worker is ready, has failed, or timeout triggers
    try:
        self._waiting_start += 1
        wait_ready = asyncio.ensure_future(self._worker_ready.wait())

        done, pending = await asyncio.wait(
            [self._task, wait_ready],
            return_when=asyncio.FIRST_COMPLETED, timeout=timeout
        )
    except asyncio.CancelledError:
        wait_ready.cancel()
        if self._waiting_start == 1:
            self.closing = True
            self._task.cancel()
            with suppress(asyncio.CancelledError):
                await self._task    # let worker shutdown
        raise
    finally:
        self._waiting_start -= 1

    # worker failed to start - either throwing or timeout triggering
    if not self._worker_ready.is_set():
        self.closing = True
        self._task.cancel()
        wait_ready.cancel()

        try:
            await self._task        # let worker shutdown
        except asyncio.CancelledError:
            raise FeedTimeoutError('stream failed to start within %ss' % timeout)
        else:
            assert False, 'worker must propagate the exception'

That seems to work, but it seems too complex, and is really hard to test: the worker has many await points, leading to combinatoric explosion if I am to try all possible cancellation points and execution orders.

I need a better way. I am thus wondering:

  • Are my requirements reasonable?
  • Is there a common pattern to do this?
  • Does my question raise some code smell?
python-asyncio python-3.x
2 Answers

Your requirements sound reasonable. I would try to simplify start by replacing Event with a future (in this case a task), using it to both wait for the startup to finish and to propagate exceptions that occur during its course, if any. Something like:

class Stream:
    async def start(self, *, timeout=DEFAULT_TIMEOUT):
        loop = asyncio.get_event_loop()
        if self._worker_startup_task is None:
            self._worker_startup_task = \
                loop.create_task(self._worker_startup())

        self._add_user()
        try:
            await asyncio.shield(asyncio.wait_for(
                self._worker_startup_task, timeout))
        except:
            self._rm_user()
            raise

    async def _worker_startup(self):
        loop = asyncio.get_event_loop()
        await asyncio.sleep(1)      # ...
        self._worker_task = loop.create_task(self._worker())

In this code the worker startup is separated from the worker coroutine, and is also moved to a separate task. This separate task can be awaited and removes the need for a dedicated Event, but more importantly, it allows scenarios 1 and 2 to be handled by the same code. Even if someone cancels the first consumer, the worker startup task will not be canceled - the cancellation just means that there is one less consumer waiting for it.

Thus in case of consumer cancellation, await self._worker_startup_task will work just fine for other consumers, whereas in case of an actual exception in worker startup, all other waiters will see the same exception because the task will have completed.

Scenario 3 should work automatically because we always cancel the startup that can no longer be observed by a consumer, regardless of the reason. If the consumers are gone because the startup itself has failed, then self._worker_startup_task will have completed (with an exception) and its cancellation will be a no-op. If it is because all consumers have been themselves canceled while awaiting the startup, then self._worker_startup_task.cancel() will cancel the startup sequence, as required by scenario 3.

The rest of the code would look like this (untested):

    def __init__(self):
        self._users = 0
        self._worker_startup = None

    def _add_user(self):
        self._users += 1

    def _rm_user(self):
        self._users -= 1
        if self._users:
            return
        self._worker_startup_task.cancel()
        self._worker_startup_task = None
        if self._worker_task is not None:
            self._worker_task.cancel()
            self._worker_task = None

    async def stop(self):
        self._rm_user()

    async def _worker(self):
        # actual worker...
        while True:
            await asyncio.sleep(1)
user_16008980

With my previous tests and integrating suggestions from @user4815162342 I came up with a re-usable solution:

st = SharedTask(test())
task1 = asyncio.ensure_future(st.wait())
task2 = asyncio.ensure_future(st.wait(timeout=15))
task3 = asyncio.ensure_future(st.wait())

This does the right thing: task2 cancels itself after 15s. Cancelling tasks has no effect on test() unless they all get cancelled. In that case, the last task to get cancelled will manually cancel test() and wait for cancellation handling to complete.

If passed a coroutine, it is only scheduled when first task starts waiting.

Lastly, awaiting the shared task after it has completed simply yields its result immediately (seems obvious, but initial version did not).

import asyncio
from contextlib import suppress


class SharedTask:
    __slots__ = ('_clients', '_task')

    def __init__(self, task):
        if not (asyncio.isfuture(task) or asyncio.iscoroutine(task)):
            raise TypeError('task must be either a Future or a coroutine object')
        self._clients = 0
        self._task = task

    @property
    def started(self):
        return asyncio.isfuture(self._task)

    async def wait(self, *, timeout=None):
        self._task = asyncio.ensure_future(self._task)

        self._clients += 1
        try:
            return await asyncio.wait_for(asyncio.shield(self._task), timeout=timeout)
        except:
            self._clients -= 1
            if self._clients == 0 and not self._task.done():
                self._task.cancel()
                with suppress(asyncio.CancelledError):
                    await self._task
            raise

    def cancel(self):
        if asyncio.iscoroutine(self._task):
            self._task.close()
        elif asyncio.isfuture(self._task):
            self._task.cancel()

The re-raising of task exception cancellation (mentioned in comments) is intentional. It allows this pattern:

async def my_task():
    try:
        await do_stuff()
    except asyncio.CancelledError as exc:
        await flush_some_stuff()     # might raise an exception
        raise exc

The clients can cancel the shared task and handle an exception that might arise as a result, it will work the same whether my_task is wrapped in a SharedTask or not.

user_32128650
Related