Writing a Custom Event Loop¶

asyncio.AbstractEventLoop declares very many methods. Implementing all themfrom scratch is a tedious job.

A loop can get many common methods implementation for free by inheriting fromasyncio.BaseEventLoop.

In turn, the successor should implement a bunch ofprivate methods declared but notimplemented inasyncio.BaseEventLoop.

For example,loop.create_connection() checks arguments, resolves DNS addresses, andcallsloop._make_socket_transport() that should be implemented by inherited class.The_make_socket_transport() method is not documented and is considered as aninternal API.

Future and Task private constructors¶

asyncio.Future andasyncio.Task should be never created directly,please use correspondingloop.create_future() andloop.create_task(),orasyncio.create_task() factories instead.

However, third-partyevent loops mayreuse built-in future and task implementationsfor the sake of getting a complex and highly optimized code for free.

For this purpose the following,private constructors are listed:

Future.__init__(*,loop=None)¶

Create a built-in future instance.

loop is an optional event loop instance.

Task.__init__(coro,*,loop=None,name=None,context=None)¶

Create a built-in task instance.

loop is an optional event loop instance. The rest of arguments are described inloop.create_task() description.

在 3.11 版的變更:context argument is added.

Task lifetime support¶

A third party task implementation should call the following functions to keep a taskvisible byasyncio.all_tasks() andasyncio.current_task():

asyncio._register_task(task)¶

Register a newtask as managed byasyncio.

Call the function from a task constructor.

asyncio._unregister_task(task)¶

Unregister atask fromasyncio internal structures.

The function should be called when a task is about to finish.

asyncio._enter_task(loop,task)¶

Switch the current task to thetask argument.

Call the function just before executing a portion of embeddedcoroutine(coroutine.send() orcoroutine.throw()).

asyncio._leave_task(loop,task)¶

Switch the current task back fromtask toNone.

Call the function just aftercoroutine.send() orcoroutine.throw()execution.