Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Use threading.Event for _eventloop_set instead of anyio.Event #1293

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Use threading.Event for _eventloop_set instead of anyio.Event #1293

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions ipykernel/kernelapp.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@

import zmq
import zmq.asyncio
from anyio import create_task_group, run
from anyio import create_task_group, run, to_thread
from IPython.core.application import ( # type:ignore[attr-defined]
BaseIPythonApplication,
base_aliases,
Expand Down Expand Up @@ -738,7 +738,7 @@ def start(self) -> None:
return

async def _wait_to_enter_eventloop(self):
await self.kernel._eventloop_set.wait()
await to_thread.run_sync(self.kernel._eventloop_set.wait)
await self.kernel.enter_eventloop()

async def main(self):
Expand Down
9 changes: 4 additions & 5 deletions ipykernel/kernelbase.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@

import psutil
import zmq
from anyio import TASK_STATUS_IGNORED, Event, create_task_group, sleep, to_thread
from anyio import TASK_STATUS_IGNORED, create_task_group, sleep, to_thread
from anyio.abc import TaskStatus
from IPython.core.error import StdinNotImplementedError
from jupyter_client.session import Session
Expand Down Expand Up @@ -226,7 +226,7 @@ def _parent_header(self):
"list_subshell_request",
]

_eventloop_set: Event = Event()
_eventloop_set: threading.Event = threading.Event()

def __init__(self, **kwargs):
"""Initialize the kernel."""
Expand Down Expand Up @@ -553,9 +553,8 @@ async def start(self, *, task_status: TaskStatus = TASK_STATUS_IGNORED) -> None:
tg.start_soon(self.shell_main, None)

def stop(self):
if not self._eventloop_set.is_set():
# Stop the async task that is waiting for the eventloop to be set.
self._eventloop_set.set()
# Stop the async task that is waiting for the eventloop to be set.
self._eventloop_set.set()
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

you possibly want to use https://anyio.readthedocs.io/en/stable/threads.html#calling-asynchronous-code-from-an-external-thread

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

or if that's overkill you could use:

def current_token() -> tuple[Literal["asyncio"], AbstractEventLoop] | tuple[Literal["trio"], trio.lowlevel.TrioToken]:
    lib = sniffio.current_async_library()
    if lib == "asyncio":
        return lib, asyncio.get_running_loop()
    if lib == "trio:
        return lib, trio.lowlevel.current_trio_token()

def from_thread_run_sync[**P](token: tuple[Literal["asyncio"], AbstractEventLoop] | tuple[Literal["trio"], trio.lowlevel.TrioToken], fn: Callable[P, object], /, *args: P.args, **kwargs: P.kwargs) -> None:
    lib, tok = token
    if lib == "asyncio":
        tok.call_soon_threadsafe(functools.partial(fn, *args, **kwargs)
        return
    if lib == "trio":
        trio.from_thread.run_sync(functools.partial(fn, *args, **kwargs), trio_token=tok)
    async def _wait_to_enter_eventloop(self):
        self.kernel._token = current_token()
        await self.kernel._eventloop_set.wait()
        await self.kernel.enter_eventloop()
def stop(self):
    from_thread_run_sync(self._token, self._eventloop_set.set)


self.shell_stop.set()
self.control_stop.set()
Expand Down
Loading