Module: zmq.utils.win32#

Win32 compatibility utilities.


class zmq.utils.win32.allow_interrupt(action: Callable[[], Any] | None = None)#

Utility for fixing CTRL-C events on Windows.

On Windows, the Python interpreter intercepts CTRL-C events in order to translate them into KeyboardInterrupt exceptions. It (presumably) does this by setting a flag in its “console control handler” and checking it later at a convenient location in the interpreter.

However, when the Python interpreter is blocked waiting for the ZMQ poll operation to complete, it must wait for ZMQ’s select() operation to complete before translating the CTRL-C event into the KeyboardInterrupt exception.

The only way to fix this seems to be to add our own “console control handler” and perform some application-defined operation that will unblock the ZMQ polling operation in order to force ZMQ to pass control back to the Python interpreter.

This context manager performs all that Windows-y stuff, providing you with a hook that is called when a CTRL-C event is intercepted. This hook allows you to unblock your ZMQ poll operation immediately, which will then result in the expected KeyboardInterrupt exception.

Without this context manager, your ZMQ-based application will not respond normally to CTRL-C events on Windows. If a CTRL-C event occurs while blocked on ZMQ socket polling, the translation to a KeyboardInterrupt exception will be delayed until the I/O completes and control returns to the Python interpreter (this may never happen if you use an infinite timeout).

A no-op implementation is provided on non-Win32 systems to avoid the application from having to conditionally use it.

Example usage:

def stop_my_application():
    # ...

with allow_interrupt(stop_my_application):
    # main polling loop.

In a typical ZMQ application, you would use the “self pipe trick” to send message to a PAIR socket in order to interrupt your blocking socket polling operation.

In a Tornado event loop, you can use the IOLoop.stop method to unblock your I/O loop.