Is there any way to specify the type of event loop for the asyncio REPL in Python?

107 Views Asked by At

I'm working on some networking code in Python that uses the asyncio module and I like to use the Python REPL from -m asyncio a lot for testing examples. What I've noticed though is the default event loop type for the asyncio REPL changes depending on the OS. For example, on Linux it uses the Selector Event Loop while on Windows it uses Proactor.

My question is this:

  1. Is there any way to specify what event loop asyncio REPL uses (without having to patch the module yourself which works obviously.)
  2. Follow up question -- if there isn't -- is there perhaps a way to replace a running event loop in a thread with another type of event loop?

Let me know what your thoughts are.

2

There are 2 best solutions below

0
haikent On

Yes, you're correct. The default event loop used by the asyncio module can depend on the operating system. The default is typically chosen to be the most appropriate for the platform.

On Linux, the default event loop is usually the SelectorEventLoop, and on Windows, it is the ProactorEventLoop. This is because the underlying I/O models are different on these platforms. SelectorEventLoop is based on the select system call, which is well-suited for Linux, while ProactorEventLoop is based on the Windows Proactor I/O model.

If you want to explicitly set the event loop type, you can do so in your code. For example, if you always want to use the SelectorEventLoop, you can set it explicitly:

import asyncio

asyncio.set_event_loop(asyncio.SelectorEventLoop())

Or, if you want to check which event loop is currently being used, you can do:

import asyncio

loop = asyncio.get_event_loop()
print(f"Current event loop: {type(loop)}")

On Linux by default this will print

<class 'asyncio.unix_events._UnixSelectorEventLoop'>

todo note: Anyone on a Windows platform please edit and add the Windows output

0
Matthew Roberts On

If anyone comes here from Google in the future: I ended up just using the existing async REPL code in the asyncio.__main__ module and modifying it for my own project so that my project can be called like python -m project and it spawns a shell (with my preferences.)

It's not such a hard thing to do. But I'll show you how to modify the REPLs environment so it will also import your project:

class SelectorEventPolicy(asyncio.DefaultEventLoopPolicy):
    @staticmethod
    def exception_handler(self, context):
        print("exception handler")
        print(context)

    @staticmethod
    def loop_setup(loop):
        loop.set_debug(False)
        loop.set_exception_handler(SelectorEventPolicy.exception_handler)
        loop.default_exception_handler = SelectorEventPolicy.exception_handler

    def new_event_loop(self):
        selector = selectors.SelectSelector()
        loop = asyncio.SelectorEventLoop(selector)
        SelectorEventPolicy.loop_setup(loop)
        return loop

asyncio.set_event_loop_policy(SelectorEventPolicy())
class REPLThread(threading.Thread):
    def run(self):
        try:
            # ... previous code to push some text to the user
            # This code executes the statement in the new prompt
            # Modify it to suite your project
            console.push("from p2pd.do_imports import *")
            # ...

The original module for Python's async REPL is here: https://github.com/python/cpython/blob/main/Lib/asyncio/__main__.py

While my fork (with minimal changes) for my own module is here: https://github.com/robertsdotpm/p2pd/blob/test_sprint1/p2pd/__main__.py

Virtually the same but I wanted to use the selector event loop for all platforms. There might be a better way to do this but this seems to work well for my project for now. Maybe useful if others come here from Google.