Using BlockingPortal leads to RuntimeError: Attempted to exit a cancel scope that isn't the current tasks's current cancel scope

[leon1995]

My goal is to make async code available to sync code, where the async code is running in a different thread, because I will have async tasks witch run in the background. Therefore, I have two threads. One where the async event loop runs (event_loop_thread) and the other one for the blocking world (MainThread).
I chose to use the BlockingPortal for that. Here is the code how I do the setup

import queue
import threading
import time
from collections.abc import Awaitable, Callable
from typing import TypeVar, TypeVarTuple, Unpack

import anyio
import anyio.from_thread


class _PortalQueue(queue.Queue[anyio.from_thread.BlockingPortal]):
    """Queue used to pass the blocking portal from one thread to the main thread."""

    def get_no_remove(self) -> anyio.from_thread.BlockingPortal:
        with self.mutex:
            return self.queue[0]


__PORTAL_QUEUE__ = _PortalQueue()


async def enter_portal_in_event_loop_thread(portal_queue: _PortalQueue):
    async with anyio.from_thread.BlockingPortal() as portal:
        portal_queue.put_nowait(portal)
        await portal.sleep_until_stopped()


def create_event_loop_thread_and_start_blocking_portal():
    t = threading.Thread(
        target=anyio.run,
        args=(enter_portal_in_event_loop_thread, __PORTAL_QUEUE__),
        daemon=True,
        name='event_loop_thread',
    )
    t.start()
    with __PORTAL_QUEUE__.not_empty:
        __PORTAL_QUEUE__.not_empty.wait()


T_Retval = TypeVar('T_Retval')
PosArgsT = TypeVarTuple('PosArgsT')


def run_async(func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval]], *args: *PosArgsT):
    if __PORTAL_QUEUE__.empty():
        create_event_loop_thread_and_start_blocking_portal()
    portal = __PORTAL_QUEUE__.get_no_remove()
    portal.call(func, *args)

Now I can call run_async with an async function and everything works. Except, when I have a TaskGroup with background tasks. Imagine the following example. I have a Subscription object that starts a renew task after subscribing. In order to test this out, I created the following script

from anyio.abc import TaskStatus

subscriptions = []


class Subscription:
    def __init__(self):
        self.tg = anyio.create_task_group()

    async def __aenter__(self):
        await self.tg.__aenter__()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print('exit')
        await self.tg.__aexit__(exc_type, exc_val, exc_tb)

    async def subscribe(self):
        print(f'I am subscribing in "{threading.get_ident()}"!')
        await self.tg.start(self.renew)

    async def renew(self, *, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORED):
        task_status.started()
        while True:
            print(f'I am renewing in "{threading.get_ident()}"!')
            await anyio.sleep(0.5)


async def make_a_subscription():
    subscription = Subscription()
    subscriptions.append(subscription)
    await subscription.__aenter__()
    await subscription.subscribe()


run_async(make_a_subscription)
run_async(make_a_subscription)
time.sleep(5)

But after the first run_async call, I get a RuntimeError: Attempted to exit a cancel scope that isn't the current tasks's current cancel scope.
Is there a solution to my problem?

[agronholm]

Why are you not using anyio.start_blocking_portal(), or even BlockingPortalProvider? Your approach seems unnecessarily complicated.

[agronholm]

Also notice the asymmetry: you're only calling __aenter__ but not __aexit__. Do not call these methods directly unless you really know what you're doing.

[leon1995]

useing anyio.from_thread.start_blocking_portal leads to the same error. I adapted the run_async method as follows

def run_async(func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval]], *args: *PosArgsT):
    with anyio.from_thread.start_blocking_portal() as portal:
        portal.call(func, *args)

using BlockingPortalProvider I might be able to reduce complexity in the program

__PORTAL_PROVIDER__ = anyio.from_thread.BlockingPortalProvider()


def run_async(func: Callable[[Unpack[PosArgsT]], Awaitable[T_Retval]], *args: *PosArgsT):
    with __PORTAL_PROVIDER__ as portal:
        portal.call(func, *args)

this also leads to the same error.
I know that I am not calling __aexit__ because I want the renew coroutine to keep running in the background, even when leaving the context of the BlockingPortal. Thats why I created the portal in a different thread. To keep the context running.

Maybe my whole approach is wrong as @smurfix pointed out, because it violates Structured Concurrency principles. However, how can I create a blocking interface to my async code, where some async code is scheduled as a background task until cancelled?

[smurfix]

I would start with some refactoring. Run the async runtime in your main task. That way you can use anyio's native thread support. I'd recommend not to directly use threading.

Also, you should your subscription using an async with block. make_a_subscription violates Structured Concurrency principles. The best pattern to use is

async with Subscription() as subscr:
    async for msg in subscr:
        await process(msg)

(thus your Subscription class needs to implement __aiter__ and __anext__)

[agronholm]

From the looks of it, you need some task to run a task group which can then support multiple subscriptions.

[leon1995]

can you please give a small example what do you mean?

[agronholm]

An example for running a task group you mean? Or...?

[leon1995]

I meant a task to run a task group. Or is a task just a coroutine running in a task group? so you have a task group inside a task group? and why would this solve my issue?

[agronholm]

A task encapsulates a coroutine, and tasks are typically created by task groups.

At any rate, your code is so far removed from the recommended best practices that I'm not even sure where to begin. I think I would have to first understand how you came to run that sync code in the first place, and then see what it wants to do on the event loop thread.

[plusls]

same problem

import contextlib

import anyio
from anyio.from_thread import start_blocking_portal


async def func1() -> None:
    print("1")


@contextlib.asynccontextmanager
async def func0() -> None:
    print("0")
    async with anyio.create_task_group() as tg:
        tg.start_soon(func1)
        print("2")
        yield
        print("3")


def main() -> None:
    with start_blocking_portal() as portal:
        portal.call(lambda: func0().__aenter__())
main()

[agronholm]

This is not really a valid use case. What are you really using it for?

[plusls]

I am sorry, the right way is

import contextlib
from collections.abc import AsyncGenerator

import anyio
from anyio.from_thread import start_blocking_portal


async def func1() -> None:
    print("1")


@contextlib.asynccontextmanager
async def func0() -> AsyncGenerator[None]:
    print("0")
    async with anyio.create_task_group() as tg:
        tg.start_soon(func1)
        yield
        print("2")



def main() -> None:
    with start_blocking_portal() as portal:
        with portal.wrap_async_context_manager(func0()) as a:
            assert a is None

main()

[agronholm]

I take it that you're not doing this just to spawn tasks, as you could do that directly with the portal.