Skip to content

event_loop

Referenced by:

Classes:

  • Host –

    A class that provides the necessary callbacks to run a gui event loop with a backend started using start_guest_run.

Functions:

  • run –

    Run func to completion asynchronously in the current thread using a backend with an optional host (gui event loop).

Host

Bases: Generic[T]

A class that provides the necessary callbacks to run a gui event loop with a backend started using start_guest_run.

Referenced by: Returned by: Subclassed by:

Methods:

  • current –

    The host running in the corresponding thread or current thread.

  • run –

    Run the loop in the current thread with a backend guest.

  • mainloop –

    Start the main event loop of the host.

Attributes:

Source code in src/async_kernel/event_loop/run.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
class Host(Generic[T]):
    """A class that provides the necessary callbacks to run a gui event loop with a `backend` started using `start_guest_run`."""

    HOST: Hosts
    MATPLOTLIB_GUIS = ()
    _subclasses: ClassVar[dict[Hosts, type[Self]]] = {}
    _instances: ClassVar[dict[threading.Thread, Host]] = {}

    _outcome: Outcome[T] | None = None
    start_guest: Callable[[], Any] = staticmethod(lambda: None)
    "A callback to start the guest. This must be called by a subclass."

    def __init_subclass__(cls) -> None:
        if cls.HOST is not Hosts.custom:
            cls._subclasses[cls.HOST] = cls

    @classmethod
    def current(cls, thread: threading.Thread | None = None) -> Host | None:
        """The host running in the corresponding thread or current thread."""
        thread = thread or threading.current_thread()
        return cls._instances.get(thread)

    @classmethod
    def run(cls, func: Callable[..., CoroutineType[Any, Any, T]], args: tuple, settings: RunSettings, /) -> T:
        """Run the loop in the current thread with a backend guest."""
        if (thread := threading.current_thread()) in cls._instances:
            msg = "A host is already running in this thread"
            raise RuntimeError(msg)

        host = Hosts(settings.get("host"))
        backend = Backend(settings.get("backend", "asyncio"))
        backend_options = settings.get("backend_options") or {}
        host_options = settings.get("host_options") or {}

        if "host_class" in host_options:
            host_options = host_options.copy()
            cls_ = host_options.pop("host_class")
            if isinstance(cls_, str):
                cls_ = import_item(cls_)
            if not issubclass(cls_, cls):
                msg = f"{cls_} is not a subclass of {cls}!"
                raise TypeError(msg)
        else:
            assert host != backend
            if host not in cls._subclasses:
                import_module(f"async_kernel.event_loop.{host}_host")
                assert host in cls._subclasses, f"Host for {host=} is not implemented correctly!"
            cls_ = cls._subclasses[host]
        assert cls_.HOST is host

        host = cls_(**host_options)
        # set the `start_guest` function (runs once).
        backend_options.setdefault("host_uses_signal_set_wakeup_fd", host.host_uses_signal_set_wakeup_fd)
        start_guest_run = get_start_guest_run(backend)
        host.start_guest = lambda: [
            start_guest_run(
                func,
                *args,
                run_sync_soon_threadsafe=host.run_sync_soon_threadsafe,
                run_sync_soon_not_threadsafe=host.run_sync_soon_not_threadsafe,
                done_callback=host.done_callback,
                **backend_options,
            ),
            setattr(host, "start_guest", lambda: None),
        ][1]
        host._instances[thread] = host
        try:
            return host.mainloop()
        finally:
            host._instances.pop(threading.current_thread())

    # Override the methods/attributes below as required.
    host_uses_signal_set_wakeup_fd = False

    def run_sync_soon_threadsafe(self, fn: Callable[[], Any]) -> None: ...
    def run_sync_soon_not_threadsafe(self, fn: Callable[[], Any]) -> None: ...

    def done_callback(self, outcome: Outcome) -> None:
        self._outcome = outcome

    def mainloop(self) -> T:
        """Start the main event loop of the host."""
        self.start_guest()  # Call at an appropriate time in the overriding subclass.
        if not self._outcome:
            msg = "The mainloop should only exit once done_callback has been called!"
            raise RuntimeError(msg)
        return self._outcome.unwrap()  # pragma: no cover

start_guest class-attribute instance-attribute 

start_guest: Callable[[], Any] = staticmethod(lambda: None)

A callback to start the guest. This must be called by a subclass.

current classmethod 

current(thread: Thread | None = None) -> Host | None

The host running in the corresponding thread or current thread.

Source code in src/async_kernel/event_loop/run.py 
95
96
97
98
99
@classmethod
def current(cls, thread: threading.Thread | None = None) -> Host | None:
    """The host running in the corresponding thread or current thread."""
    thread = thread or threading.current_thread()
    return cls._instances.get(thread)

run classmethod 

run(
    func: Callable[..., CoroutineType[Any, Any, T]], args: tuple, settings: RunSettings
) -> T

Run the loop in the current thread with a backend guest.

Source code in src/async_kernel/event_loop/run.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
@classmethod
def run(cls, func: Callable[..., CoroutineType[Any, Any, T]], args: tuple, settings: RunSettings, /) -> T:
    """Run the loop in the current thread with a backend guest."""
    if (thread := threading.current_thread()) in cls._instances:
        msg = "A host is already running in this thread"
        raise RuntimeError(msg)

    host = Hosts(settings.get("host"))
    backend = Backend(settings.get("backend", "asyncio"))
    backend_options = settings.get("backend_options") or {}
    host_options = settings.get("host_options") or {}

    if "host_class" in host_options:
        host_options = host_options.copy()
        cls_ = host_options.pop("host_class")
        if isinstance(cls_, str):
            cls_ = import_item(cls_)
        if not issubclass(cls_, cls):
            msg = f"{cls_} is not a subclass of {cls}!"
            raise TypeError(msg)
    else:
        assert host != backend
        if host not in cls._subclasses:
            import_module(f"async_kernel.event_loop.{host}_host")
            assert host in cls._subclasses, f"Host for {host=} is not implemented correctly!"
        cls_ = cls._subclasses[host]
    assert cls_.HOST is host

    host = cls_(**host_options)
    # set the `start_guest` function (runs once).
    backend_options.setdefault("host_uses_signal_set_wakeup_fd", host.host_uses_signal_set_wakeup_fd)
    start_guest_run = get_start_guest_run(backend)
    host.start_guest = lambda: [
        start_guest_run(
            func,
            *args,
            run_sync_soon_threadsafe=host.run_sync_soon_threadsafe,
            run_sync_soon_not_threadsafe=host.run_sync_soon_not_threadsafe,
            done_callback=host.done_callback,
            **backend_options,
        ),
        setattr(host, "start_guest", lambda: None),
    ][1]
    host._instances[thread] = host
    try:
        return host.mainloop()
    finally:
        host._instances.pop(threading.current_thread())

mainloop 

mainloop() -> T

Start the main event loop of the host.

Source code in src/async_kernel/event_loop/run.py 
159
160
161
162
163
164
165
def mainloop(self) -> T:
    """Start the main event loop of the host."""
    self.start_guest()  # Call at an appropriate time in the overriding subclass.
    if not self._outcome:
        msg = "The mainloop should only exit once done_callback has been called!"
        raise RuntimeError(msg)
    return self._outcome.unwrap()  # pragma: no cover

run 

run(
    func: Callable[..., CoroutineType[Any, Any, T]], args: tuple, settings: RunSettings
) -> T

Run func to completion asynchronously in the current thread using a backend with an optional host (gui event loop).

The default backend is 'asyncio'.

If host is specified in settings. A host (gui) mainloop will be started with the backend running as a guest (in the same thread). The backend will execute func asynchronously to completion. Once completed the backend and host are stopped and finally the result is returned.

Parameters:

Custom host

A custom host can be started by subclassing Host and passed as the 'host_class' as the class or a dotted path if it is importable.

loop_factory

When there is no host and the backend is 'asyncio', the loop factory can be specified in backend_options as a function or an importatable dotted path such as 'asyncio.new_event_loop'.

Referenced by:
Source code in src/async_kernel/event_loop/run.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def run(func: Callable[..., CoroutineType[Any, Any, T]], args: tuple, settings: RunSettings, /) -> T:
    """Run `func` to completion asynchronously in the current thread using a [backend][async_kernel.typing.Backend] with an optional host (gui event loop).

    The default backend is ['asyncio'][async_kernel.typing.Backend.asyncio].

    If [host][async_kernel.typing.Hosts] is specified in `settings`. A _host_ (gui) mainloop
    will be started with the `backend` running as a guest (in the same thread). The `backend`
    will execute `func` asynchronously to completion. Once completed the `backend` and `host`
    are stopped and finally the result is returned.

    Args:
        func: A coroutine function.
        args: Args to use when calling func.
        settings: Settings to use when running func.

    Custom host:
        A custom host can be started by subclassing [Host][] and passed as the 'host_class' as the
        class or a dotted path if it is importable.

    loop_factory:
        When there is no host and the backend is 'asyncio', the loop factory can be specified
        in backend_options as a function or an importatable dotted path such as `'asyncio.new_event_loop'`.
    """
    if settings.get("host"):
        # A gui with the backend running as a guest.
        return Host.run(func, args, settings)
    # backend only.
    if (backend_options := settings.get("backend_options")) and isinstance(
        (loop_factory := backend_options.get("loop_factory")), str
    ):
        # Import the loop factory
        backend_options["loop_factory"] = import_item(loop_factory)
    return anyio.run(func, *args, backend=Backend(settings.get("backend", "asyncio")), backend_options=backend_options)

Bases: Host[T]

Methods:

Source code in src/async_kernel/event_loop/tk_host.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
class TkHost(Host[T]):
    HOST = Hosts.tk
    MATPLOTLIB_GUIS = ("tk",)

    def __init__(self) -> None:
        root = tk.Tk()
        root.withdraw()
        self.root = root
        self._tk_func_name = root.register(self._tk_func)
        self._q = collections.deque()

    def _tk_func(self) -> None:
        self._q.popleft()()

    @override
    def run_sync_soon_threadsafe(self, fn: Callable[[], Any]) -> None:
        """Use Tcl "after" command to schedule a function call.

        Based on [tkinter source comments](https://github.com/python/cpython/blob/a5d6aba318ead9cc756ba750a70da41f5def3f8f/Modules/_tkinter.c#L1472-L1555)
        the issuance of the tcl call to after itself is thread-safe since it is sent
        to the [appropriate thread](https://github.com/python/cpython/blob/a5d6aba318ead9cc756ba750a70da41f5def3f8f/Modules/_tkinter.c#L814-L824>) on line 1522.
        `Tkapp_ThreadSend` effectively uses "after 0" while putting the command in the
        event queue so the ["after idle after 0"](https://wiki.tcl-lang.org/page/after#096aeab6629eae8b244ae2eb2000869fbe377fa988d192db5cf63defd3d8c061)` incantation
        is unnecessary here.

        Compare to [tkthread](https://github.com/serwy/tkthread/blob/1f612e1dd46e770bd0d0bb64d7ecb6a0f04875a3/tkthread/__init__.py#L163)
        where definitely thread unsafe [eval](https://github.com/python/cpython/blob/a5d6aba318ead9cc756ba750a70da41f5def3f8f/Modules/_tkinter.c#L1567-L1585)
        is used to send thread safe signals between tcl interpreters.
        """
        # self.root.after_idle(lambda:self.root.after(0, func)) # does a fairly intensive wrapping to each func
        self._q.append(fn)
        self.root.call("after", "idle", self._tk_func_name)

    @override
    def run_sync_soon_not_threadsafe(self, fn) -> None:
        """Use Tcl "after" command to schedule a function call from the main thread.

        If .call is called from the Tcl thread, the locking and sending are optimized away
        so it should be fast enough.

        The incantation ["after idle after 0"](https://wiki.tcl-lang.org/page/after#096aeab6629eae8b244ae2eb2000869fbe377fa988d192db5cf63defd3d8c061)
        avoids blocking the normal event queue when faced with an unending stream of tasks, for
        example `while True: await trio.sleep(0)`.
        """
        self._q.append(fn)
        self.root.call("after", "idle", "after", 0, self._tk_func_name)
        # Not sure if this is actually an optimization because Tcl parses this eval string fresh each time.
        # However it's definitely thread unsafe because the string is fed directly into the Tcl interpreter
        # from the current Python thread
        # self.root.eval(f'after idle after 0 {self._tk_func_name}')

    @override
    def done_callback(self, outcome) -> None:
        """End the Tk app."""
        super().done_callback(outcome)
        self.root.destroy()

    @override
    def mainloop(self) -> T:
        self.start_guest()
        self.root.mainloop()
        return super().mainloop()

run_sync_soon_threadsafe 

run_sync_soon_threadsafe(fn: Callable[[], Any]) -> None

Use Tcl "after" command to schedule a function call.

Based on tkinter source comments the issuance of the tcl call to after itself is thread-safe since it is sent to the appropriate thread on line 1522. Tkapp_ThreadSend effectively uses "after 0" while putting the command in the event queue so the "after idle after 0"` incantation is unnecessary here.

Compare to tkthread where definitely thread unsafe eval is used to send thread safe signals between tcl interpreters.

Source code in src/async_kernel/event_loop/tk_host.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
@override
def run_sync_soon_threadsafe(self, fn: Callable[[], Any]) -> None:
    """Use Tcl "after" command to schedule a function call.

    Based on [tkinter source comments](https://github.com/python/cpython/blob/a5d6aba318ead9cc756ba750a70da41f5def3f8f/Modules/_tkinter.c#L1472-L1555)
    the issuance of the tcl call to after itself is thread-safe since it is sent
    to the [appropriate thread](https://github.com/python/cpython/blob/a5d6aba318ead9cc756ba750a70da41f5def3f8f/Modules/_tkinter.c#L814-L824>) on line 1522.
    `Tkapp_ThreadSend` effectively uses "after 0" while putting the command in the
    event queue so the ["after idle after 0"](https://wiki.tcl-lang.org/page/after#096aeab6629eae8b244ae2eb2000869fbe377fa988d192db5cf63defd3d8c061)` incantation
    is unnecessary here.

    Compare to [tkthread](https://github.com/serwy/tkthread/blob/1f612e1dd46e770bd0d0bb64d7ecb6a0f04875a3/tkthread/__init__.py#L163)
    where definitely thread unsafe [eval](https://github.com/python/cpython/blob/a5d6aba318ead9cc756ba750a70da41f5def3f8f/Modules/_tkinter.c#L1567-L1585)
    is used to send thread safe signals between tcl interpreters.
    """
    # self.root.after_idle(lambda:self.root.after(0, func)) # does a fairly intensive wrapping to each func
    self._q.append(fn)
    self.root.call("after", "idle", self._tk_func_name)

run_sync_soon_not_threadsafe 

run_sync_soon_not_threadsafe(fn) -> None

Use Tcl "after" command to schedule a function call from the main thread.

If .call is called from the Tcl thread, the locking and sending are optimized away so it should be fast enough.

The incantation "after idle after 0" avoids blocking the normal event queue when faced with an unending stream of tasks, for example while True: await trio.sleep(0).

Source code in src/async_kernel/event_loop/tk_host.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
@override
def run_sync_soon_not_threadsafe(self, fn) -> None:
    """Use Tcl "after" command to schedule a function call from the main thread.

    If .call is called from the Tcl thread, the locking and sending are optimized away
    so it should be fast enough.

    The incantation ["after idle after 0"](https://wiki.tcl-lang.org/page/after#096aeab6629eae8b244ae2eb2000869fbe377fa988d192db5cf63defd3d8c061)
    avoids blocking the normal event queue when faced with an unending stream of tasks, for
    example `while True: await trio.sleep(0)`.
    """
    self._q.append(fn)
    self.root.call("after", "idle", "after", 0, self._tk_func_name)

done_callback 

done_callback(outcome) -> None

End the Tk app.

Source code in src/async_kernel/event_loop/tk_host.py 
87
88
89
90
91
@override
def done_callback(self, outcome) -> None:
    """End the Tk app."""
    super().done_callback(outcome)
    self.root.destroy()

options: show_root_heading: true

Bases: Host[T]

Source code in src/async_kernel/event_loop/qt_host.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
class QtHost(Host[T]):
    HOST = Hosts.qt
    MATPLOTLIB_GUIS = ("qt",)

    def __init__(self, module: Literal["PySide6", "PySide2", "PyQt5", "PyQt6"] = "PySide6") -> None:
        if threading.current_thread() is not threading.main_thread():
            msg = "QT can only be run in main thread!"
            raise RuntimeError(msg)

        globals()["QtCore"] = import_from(module, "QtCore")
        globals()["QtWidgets"] = import_from(module, "QtWidgets")

        REENTER_EVENT_TYPE = QtCore.QEvent.Type(QtCore.QEvent.registerEventType())

        class ReenterEvent(QtCore.QEvent):
            fn: Callable[[], Any]

        class Reenter(QtCore.QObject):
            @override
            def event(self, event: ReenterEvent) -> Literal[False]:  # pyright: ignore[reportIncompatibleMethodOverride]
                event.fn()
                return False

        reenter = Reenter()

        if (app := QtWidgets.QApplication.instance()) is None:
            app = QtWidgets.QApplication([])
            app.setQuitOnLastWindowClosed(False)  # prevent app sudden death

        def run_soon_threadsafe(fn):
            event = ReenterEvent(REENTER_EVENT_TYPE)
            event.fn = fn
            app.postEvent(reenter, event)

        self.run_sync_soon_threadsafe = run_soon_threadsafe
        self.run_sync_soon_not_threadsafe = run_soon_threadsafe
        self.app = app

    @override
    def done_callback(self, outcome) -> None:
        super().done_callback(outcome)
        self.app.quit()

    @override
    def mainloop(self) -> T:
        self.start_guest()
        self.app.exec()
        return super().mainloop()

options: show_root_heading: true

Defines a background ZMQ poller thread.

Classes:

ZMQPollSocket

Bases: Socket[bytes]

A zmq socket which uses a ZMQPoll thread to perform sensitive operation.

For best reliability, sensitive operations are performed in the zmq_poll's thread. This socket will close automatically when Poll is stopped. It is still better to close the socket when it is no longer required. The socket must be bound/connected in the context of the ZMQPoll to which it is associated.

Referenced by: Returned by: Used by:

Attributes:

Source code in src/async_kernel/event_loop/zmq_poll.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
class ZMQPollSocket(zmq.sugar.Socket[bytes]):
    """A zmq socket which uses a [ZMQPoll][] thread to perform sensitive operation.

    For best reliability, sensitive operations are performed in the zmq_poll's thread.
    This socket will close automatically when Poll is stopped. It is still better to
    close the  socket when it is no longer required. The socket must be bound/connected
    in the context of the [ZMQPoll][] to which it is associated.
    """

    _zmq_poll_ref: weakref.ref[ZMQPoll]
    lock: threading.Lock

    if TYPE_CHECKING:
        # magic attributes cannot be be stored in `__annotations__`.

        curve_secretkey: bytes | None
        """The curve encryption secret key."""

        curve_publickey: bytes | None
        """The curve encryption public key."""

        curve_server: bool
        """If it is the server (must be bound)."""

        curve_serverkey: bytes | None
        """The key to use when the socket is not a serve (normally matches curve_publickey)."""

    def __init__(
        self,
        socket_type: zmq.SocketType,
        shadow: ZMQPollSocket | int = 0,
        *,
        zmq_poll: ZMQPoll,
        copy_threshold: int | None = None,
    ) -> None:
        self.lock = threading.Lock()
        self._zmq_poll_ref = weakref.ref(zmq_poll)

        if zmq_poll.stopped.done():
            msg = f"{zmq_poll} is stopped!"
            raise RuntimeError(msg)
        super().__init__(
            ctx_or_socket=zmq_poll._zmq_context,  # pyright: ignore[reportPrivateUsage],
            socket_type=int(socket_type),
            copy_threshold=copy_threshold,
        )
        zmq_poll.sockets.add(self)

    @property
    def zmq_poll(self) -> ZMQPoll:
        return self._zmq_poll_ref()  # pyright: ignore[reportReturnType]

    @override
    def set(self, option: int, optval: int | bytes) -> None:
        assert not self.closed
        self.zmq_poll.execute(super().set, option, optval)

    @override
    def send_multipart(
        self,
        msg_parts: Sequence,
        flags: int = 0,
        copy: bool = True,
        track: bool = False,
        **kwargs,
    ) -> MessageTracker | None:
        self.lock.acquire()
        try:
            if self.closed:
                return None
            return super().send_multipart(msg_parts, flags, copy, track)
        finally:
            self.lock.release()

    @override
    def close(self, linger=None) -> None:
        with self.lock:
            if not self.closed:
                # Remove from `zmq_poll._handlers` before closing to avoid heap corruption (Windows).
                for k in (k_ for k_ in list(self.zmq_poll._handlers.keys()) if k_[0] is self):  # pyright: ignore[reportPrivateUsage]
                    self.zmq_poll._handlers.pop(k, None)  # pyright: ignore[reportPrivateUsage]
                self.zmq_poll.sockets.discard(self)
                self.zmq_poll.execute(super().close, linger)

    @override
    def bind(self, addr: str) -> _SocketContext[Self]:
        assert not self.closed
        return self.zmq_poll.execute(super().bind, addr)

    @override
    def unbind(self, addr: str | bytes) -> None:
        if not self.closed:
            self.zmq_poll.execute(super().unbind, addr)

    @override
    def connect(self, addr: str) -> _SocketContext[Self]:
        assert not self.closed
        return self.zmq_poll.execute(super().connect, addr)

    @override
    def disconnect(self, addr: str | bytes) -> None:
        if not self.closed:
            self.zmq_poll.execute(super().disconnect, addr)

    @override
    def subscribe(self, topic: str | bytes) -> None:
        assert not self.closed
        topic = topic.encode("utf8") if isinstance(topic, str) else topic
        return self.zmq_poll.execute(super().subscribe, topic)

    @override
    def unsubscribe(self, topic: str | bytes) -> None:
        if not self.closed:
            topic = topic.encode("utf8") if isinstance(topic, str) else topic
            self.zmq_poll.execute(super().unsubscribe, topic)

curve_secretkey instance-attribute 

curve_secretkey: bytes | None

The curve encryption secret key.

curve_publickey instance-attribute 

curve_publickey: bytes | None

The curve encryption public key.

curve_server instance-attribute 

curve_server: bool

If it is the server (must be bound).

curve_serverkey instance-attribute 

curve_serverkey: bytes | None

The key to use when the socket is not a serve (normally matches curve_publickey).

ZMQPoll

A zmq_poll based event loop.

When a (socket, flags) -> handler mapping context manager is used ZMQPoll.event_handler handler is called with the socket and event when the event occurs on the socket. The callback occurs in the zmq poll thread. async_kernel.caller.Caller is recommended to to scheduling code execution in different threads.

Only ZMQPollSocket sockets created using the ZMQPoll.socket factory function are allowed.

The methods ZMQPoll.execute and the async version ZMQPoll.aexecute are provided to executed code in the zmq_poll thread, which is useful when creating and configuring sockets to reduce thread switching.

Referenced by:

Methods:

  • validate_socket –

    Check sock is correctly registered.

  • socket –

    Create a new ZMQPollSocket.

  • execute –

    Execute func in the 'zmq_poll' thread waiting for the result synchronously.

  • aexecute –

    Execute func in the 'zmq_poll' thread waiting for the result asynchronously.

  • event_handler –

    A context manager where handler is called in the 'zmq_poll' thread with the event number when it occurs for sock.

Attributes:

Source code in src/async_kernel/event_loop/zmq_poll.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
class ZMQPoll:
    """A [zmq_poll](https://libzmq.readthedocs.io/en/latest/zmq_poll.html) based event loop.

    When a `(socket, flags) -> handler` mapping context manager is used [ZMQPoll.event_handler][]
    `handler` is called with the socket and event when the event occurs on the socket. The
    callback occurs in the zmq poll thread. [async_kernel.caller.Caller][] is recommended to
    to scheduling code execution in different threads.

    Only [ZMQPollSocket][] sockets created using the [ZMQPoll.socket][] factory function are
    allowed.

    The methods [ZMQPoll.execute][] and the async version [ZMQPoll.aexecute][] are provided
    to executed code in the zmq_poll thread, which is useful when creating and configuring
    sockets to reduce thread switching.
    """

    stopped: Fixed[Self, ProtectedPending] = Fixed(ProtectedPending)
    """Set when the poll thread event has stopped."""

    sockets: Fixed[Self, set[ZMQPollSocket]] = Fixed(set)
    """The sockets currently registered with this instance."""

    def __init__(self) -> None:

        def socket_factory(
            ctx_or_socket: zmq.Context | None = None,
            socket_type: int = 0,
            *,
            copy_threshold: int | None = None,
        ) -> ZMQPollSocket:
            return ZMQPollSocket(
                socket_type=zmq.SocketType(socket_type),
                copy_threshold=copy_threshold,
                zmq_poll=ref(),  # pyright: ignore[reportArgumentType]
            )

        self._zmq_context = zmq.Context()
        ref = weakref.ref(self)
        self._zmq_context._socket_class = socket_factory  # pyright: ignore[reportPrivateUsage, reportAttributeAccessIssue]
        self._handlers: dict[T_key, Callable[[ZMQPollSocket, int], Any]] = {}
        self._count: dict[T_key, tuple[int, Callable[[], Any]] | None] = {}
        self._execute: deque[Pending] = deque[Pending[Any]]()
        self.log = logging.LoggerAdapter(logging.getLogger())
        self._cancellers = deque()
        self._ctx_count = 0
        self._lock = BinarySemaphore()

    def __enter__(self) -> Self:
        with self._lock:
            if self.stopped.done():
                msg = "This zmq_poll event loop is stopped!"
                raise RuntimeError(msg)
            if self._ctx_count == 0:
                self.__start()
            self._ctx_count = self._ctx_count + 1
        return self

    def __exit__(self, type, value, traceback) -> Literal[False]:
        with self._lock:
            self._ctx_count = self._ctx_count - 1
            if self._ctx_count == 0:
                self.stopped.set_result(None)
                self.thread.join()
        return False

    def _wake(self) -> None:
        """Unblock the thread."""

    def _on_stopped(self) -> None:
        self._wake()
        del self._wake

    def __start(self) -> None:

        def zmq_poll_thread(
            *,
            context=self._zmq_context,
            handlers: dict[T_key, Callable[[ZMQPollSocket, int], Any]] = self._handlers,
            stopped: Pending[None] = self.stopped,
            count: dict[T_key, tuple[int, Callable[[], Any]] | None] = self._count,
            execute=self._execute,
            zmq_poll_sockets: set[ZMQPollSocket] = self.sockets,
            cancellers=self._cancellers,
            log=self.log,
        ) -> None:
            """Runs the 'event' loop."""
            # Thread: zmq_poll_thread
            if not utils.LAUNCHED_BY_DEBUGPY:
                utils.mark_thread_pydev_do_not_trace()

            def on_wake(sock: ZMQPollSocket, flags: int) -> None:
                """On receipt of a wake event clear the sockets."""
                nonlocal sockets
                # Called on receipt of a message (b'') on the 'wake' socket.
                sockets = None
                sock.recv()

            def do_execute() -> None:
                """Execute pending items added by the `execute` and `aexecute` methods."""
                while execute:
                    md = (pen := execute.popleft()).metadata
                    try:
                        pen.set_result(md["func"](*md["args"], **md["kwargs"]))
                    except BaseException as e:
                        pen.set_exception(e)
                    del pen

            send: ZMQPollSocket = context.socket(zmq.SocketType.PAIR)  # pyright: ignore[reportAssignmentType]
            wake: ZMQPollSocket = context.socket(zmq.SocketType.PAIR)  # pyright: ignore[reportAssignmentType]
            addr = "inproc://async_kernel_zmq_poller_wake"
            sockets = None
            handlers[(wake, zmq.POLLIN)] = on_wake

            with context, wake, send, wake.bind(addr), send.connect(addr):
                k: T_key
                c: tuple[int, Callable] | None
                started.set_result(send)
                # The main loop polls the handler keys for events in a loop.
                # It will block until an event occurs.
                try:
                    while not stopped.done():
                        if not sockets:
                            sockets = list(handlers)
                        if execute:
                            do_execute()
                            continue
                        try:
                            for k in _zmq_poll(sockets, timeout=-1):  # pyright: ignore[reportAssignmentType]
                                try:
                                    handlers[k](*k)
                                except KeyError:
                                    sockets = None
                                except SystemExit:
                                    stopped.set_result(None)
                                except BaseException as e:
                                    self.log.exception("Ignoring exception in handler.", exc_info=e)
                                if count and (c := count.get(k)) is not None:
                                    c = count[k] = (int(c[0]) - 1, c[1])
                                    # Auto eject after 'n' events
                                    if c[0] == 0:
                                        handlers.pop(k, None)
                                        count[k] = sockets = None
                                        c[1]()
                        except zmq.ZMQError:
                            for k, v in handlers.copy().items():
                                if k[0].closed:
                                    handlers.pop(k, None)
                                    log.debug("Closed sockets detected %s -> %s", k[0], v)
                                sockets = None
                        except Exception as e:
                            self.log.exception("Ignoring exception in zmq_poll_thread.", exc_info=e)
                finally:
                    do_execute()
                    while cancellers:
                        try:
                            cancellers.popleft()()
                        except Exception as e:
                            self.log.exception("A canceller failed", exc_info=e)
                    handlers.clear()
                    while zmq_poll_sockets:
                        try:
                            zmq_poll_sockets.pop().close()
                        except Exception as e:
                            self.log.exception("Socket close call failed", exc_info=e)
                    log.debug("Stopped zmq_poll event loop")

        self.log.debug("Starting ZMQPoll event loop")
        started = Pending[ZMQPollSocket]()
        ref = weakref.ref(self)
        self.thread = threading.Thread(target=zmq_poll_thread)
        self.thread.start()
        send = started.wait_sync()

        def _wake(sock=send, lock=send.lock) -> None:
            lock.acquire()
            sock.send(b"")
            lock.release()

        self._wake = _wake
        self.stopped.add_done_callback(lambda _: (self := ref()) and self._on_stopped())
        self.log.debug("ZMQPoll event loop started")

    def validate_socket(self, sock: ZMQPollSocket | Any) -> ZMQPollSocket:
        """Check `sock` is correctly registered."""
        if sock not in self.sockets:
            msg = f"Invalid socket detected! {sock=}"
            raise ValueError(msg)
        return sock

    def socket(self, socket_type: zmq.SocketType) -> ZMQPollSocket:
        """Create a new [ZMQPollSocket][].

        Args:
            socket_type: The type of socket.
        """
        if self.stopped.done():
            msg = f"{self} is stopped!"
            raise RuntimeError(msg)
        return self.validate_socket(self.execute(self._zmq_context.socket, socket_type))

    def execute(self, func: Callable[P, T], /, *args: P.args, **kwargs: P.kwargs) -> T:
        """Execute `func` in the 'zmq_poll' thread waiting for the result synchronously."""
        if hasattr(self, "thread"):
            if threading.current_thread() is self.thread:
                return func(*args, **kwargs)
            self._execute.append(pen := Pending[T](func=func, args=args, kwargs=kwargs))
            if not self.stopped.done():
                self._wake()
                try:
                    return pen.wait_sync()
                finally:
                    pen.metadata.clear()
                    del pen
        msg = f"Unable to execute {func=} in {self}. Execution is only supported while in context."
        raise RuntimeError(msg)

    async def aexecute(self, func: Callable[P, T], /, *args: P.args, **kwargs: P.kwargs) -> T:
        """Execute `func` in the 'zmq_poll' thread waiting for the result asynchronously."""
        if hasattr(self, "thread"):
            self._execute.append(pen := Pending[T](func=func, args=args, kwargs=kwargs))
            if not self.stopped.done():
                self._wake()
                return await pen
        msg = f"Unable to execute {func=} in {self}. Execution is only supported while in context."
        raise RuntimeError(msg)

    @contextmanager
    def event_handler(
        self,
        sock: ZMQPollSocket,
        handler: Callable[[ZMQPollSocket, int], Any],
        /,
        *,
        flags: Literal[zmq.PollEvent.POLLIN, zmq.PollEvent.POLLOUT] = zmq.PollEvent.POLLIN,
        count: tuple[int, Callable[[], Any]] | None = None,
        canceller: Callable[[], Any] | NoValue | None = NoValue,
    ) -> Generator[None, Any, None]:
        """A context manager where `handler` is called in the 'zmq_poll' thread with the event number when it occurs for `sock`.

        Only one `handler` is allowed per `(socket, flags)` combination.

        Args:
            sock: A registered [ZMQPollSocket][].
            handler: A handler to handle the event. The handler is called inside the
                zmq_poll thread. Thread-safe primitives must be used by the handler such
                as [async_kernel.caller.Caller.call_soon][],[async_kernel.caller.Caller.queue_call][], etc.
            flags: The type of event to listen for.
                [zmq.PollEvent.POLLIN][]: `sock` is readable.
                [zmq.PollEvent.POLLOUT][]: `sock` was read from.
            count: A tuple ('n', callback) where the handler is run exactly 'n' times.
                The callback could be an `event.set` to release the context.
            canceller: A callback to use on the event the poll is stopped. The default cancellation
                behavior is to cancel the pending returned by [async_kernel.caller.Caller.current_pending][].
                Set to `None` to disable cancellation support. This is safe when this context manager is
                inside the context of this instance.

        Raises:
            BusyResourceError: If the `(sock, flags)` combination is already in use.
            RuntimeError: If the default canceller can not be created.

        Tip:
            The zmq_poll thread normally disables debugging in the zmq_poll thread so inserting breakpoints
            in the event handler may interfere with debugging.
        """
        assert not self.stopped.done()
        if canceller is NoValue:
            if not (pen := Caller.current_pending()):
                msg = "This context is not cancellable!."
                raise RuntimeError(msg)

            def default_handler(pen=pen) -> None:
                pen.cancel("The ZMQPoll event loop has stopped!")

            canceller = default_handler

        self.validate_socket(sock)
        if count:
            assert count[0] > 0
            assert callable(count[1])
        if canceller:
            self._cancellers.append(canceller)
        if handler is not self._handlers.setdefault(k := (sock, int(flags)), handler):
            raise BusyResourceError
        self._count[k] = count
        self._wake()
        try:
            yield None
        finally:
            if canceller:
                with contextlib.suppress(ValueError):
                    self._cancellers.remove(canceller)
            self._handlers.pop(k, None)
            self._count.pop(k, None)
            self._wake()

stopped class-attribute instance-attribute 

stopped: Fixed[Self, ProtectedPending] = Fixed(ProtectedPending)

Set when the poll thread event has stopped.

sockets class-attribute instance-attribute 

sockets: Fixed[Self, set[ZMQPollSocket]] = Fixed(set)

The sockets currently registered with this instance.

validate_socket 

validate_socket(sock: ZMQPollSocket | Any) -> ZMQPollSocket

Check sock is correctly registered.

Source code in src/async_kernel/event_loop/zmq_poll.py 
336
337
338
339
340
341
def validate_socket(self, sock: ZMQPollSocket | Any) -> ZMQPollSocket:
    """Check `sock` is correctly registered."""
    if sock not in self.sockets:
        msg = f"Invalid socket detected! {sock=}"
        raise ValueError(msg)
    return sock

socket 

socket(socket_type: SocketType) -> ZMQPollSocket

Create a new ZMQPollSocket.

Parameters:

Referenced by:
Source code in src/async_kernel/event_loop/zmq_poll.py 
343
344
345
346
347
348
349
350
351
352
def socket(self, socket_type: zmq.SocketType) -> ZMQPollSocket:
    """Create a new [ZMQPollSocket][].

    Args:
        socket_type: The type of socket.
    """
    if self.stopped.done():
        msg = f"{self} is stopped!"
        raise RuntimeError(msg)
    return self.validate_socket(self.execute(self._zmq_context.socket, socket_type))

execute 

execute(func: Callable[P, T], /, *args: args, **kwargs: kwargs) -> T

Execute func in the 'zmq_poll' thread waiting for the result synchronously.

Referenced by:
Source code in src/async_kernel/event_loop/zmq_poll.py 
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
def execute(self, func: Callable[P, T], /, *args: P.args, **kwargs: P.kwargs) -> T:
    """Execute `func` in the 'zmq_poll' thread waiting for the result synchronously."""
    if hasattr(self, "thread"):
        if threading.current_thread() is self.thread:
            return func(*args, **kwargs)
        self._execute.append(pen := Pending[T](func=func, args=args, kwargs=kwargs))
        if not self.stopped.done():
            self._wake()
            try:
                return pen.wait_sync()
            finally:
                pen.metadata.clear()
                del pen
    msg = f"Unable to execute {func=} in {self}. Execution is only supported while in context."
    raise RuntimeError(msg)

aexecute async 

aexecute(func: Callable[P, T], /, *args: args, **kwargs: kwargs) -> T

Execute func in the 'zmq_poll' thread waiting for the result asynchronously.

Referenced by:
Source code in src/async_kernel/event_loop/zmq_poll.py 
370
371
372
373
374
375
376
377
378
async def aexecute(self, func: Callable[P, T], /, *args: P.args, **kwargs: P.kwargs) -> T:
    """Execute `func` in the 'zmq_poll' thread waiting for the result asynchronously."""
    if hasattr(self, "thread"):
        self._execute.append(pen := Pending[T](func=func, args=args, kwargs=kwargs))
        if not self.stopped.done():
            self._wake()
            return await pen
    msg = f"Unable to execute {func=} in {self}. Execution is only supported while in context."
    raise RuntimeError(msg)

event_handler 

event_handler(
    sock: ZMQPollSocket,
    handler: Callable[[ZMQPollSocket, int], Any],
    /,
    *,
    flags: Literal[POLLIN, POLLOUT] = POLLIN,
    count: tuple[int, Callable[[], Any]] | None = None,
    canceller: Callable[[], Any] | NoValue | None = NoValue,
) -> Generator[None, Any, None]

A context manager where handler is called in the 'zmq_poll' thread with the event number when it occurs for sock.

Only one handler is allowed per (socket, flags) combination.

Parameters:

Raises:

Tip

The zmq_poll thread normally disables debugging in the zmq_poll thread so inserting breakpoints in the event handler may interfere with debugging.

Referenced by:
Source code in src/async_kernel/event_loop/zmq_poll.py 
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
@contextmanager
def event_handler(
    self,
    sock: ZMQPollSocket,
    handler: Callable[[ZMQPollSocket, int], Any],
    /,
    *,
    flags: Literal[zmq.PollEvent.POLLIN, zmq.PollEvent.POLLOUT] = zmq.PollEvent.POLLIN,
    count: tuple[int, Callable[[], Any]] | None = None,
    canceller: Callable[[], Any] | NoValue | None = NoValue,
) -> Generator[None, Any, None]:
    """A context manager where `handler` is called in the 'zmq_poll' thread with the event number when it occurs for `sock`.

    Only one `handler` is allowed per `(socket, flags)` combination.

    Args:
        sock: A registered [ZMQPollSocket][].
        handler: A handler to handle the event. The handler is called inside the
            zmq_poll thread. Thread-safe primitives must be used by the handler such
            as [async_kernel.caller.Caller.call_soon][],[async_kernel.caller.Caller.queue_call][], etc.
        flags: The type of event to listen for.
            [zmq.PollEvent.POLLIN][]: `sock` is readable.
            [zmq.PollEvent.POLLOUT][]: `sock` was read from.
        count: A tuple ('n', callback) where the handler is run exactly 'n' times.
            The callback could be an `event.set` to release the context.
        canceller: A callback to use on the event the poll is stopped. The default cancellation
            behavior is to cancel the pending returned by [async_kernel.caller.Caller.current_pending][].
            Set to `None` to disable cancellation support. This is safe when this context manager is
            inside the context of this instance.

    Raises:
        BusyResourceError: If the `(sock, flags)` combination is already in use.
        RuntimeError: If the default canceller can not be created.

    Tip:
        The zmq_poll thread normally disables debugging in the zmq_poll thread so inserting breakpoints
        in the event handler may interfere with debugging.
    """
    assert not self.stopped.done()
    if canceller is NoValue:
        if not (pen := Caller.current_pending()):
            msg = "This context is not cancellable!."
            raise RuntimeError(msg)

        def default_handler(pen=pen) -> None:
            pen.cancel("The ZMQPoll event loop has stopped!")

        canceller = default_handler

    self.validate_socket(sock)
    if count:
        assert count[0] > 0
        assert callable(count[1])
    if canceller:
        self._cancellers.append(canceller)
    if handler is not self._handlers.setdefault(k := (sock, int(flags)), handler):
        raise BusyResourceError
    self._count[k] = count
    self._wake()
    try:
        yield None
    finally:
        if canceller:
            with contextlib.suppress(ValueError):
                self._cancellers.remove(canceller)
        self._handlers.pop(k, None)
        self._count.pop(k, None)
        self._wake()