Appendix_EventLoop

Event Loop

import asyncio


loop = asyncio.get_event_loop()


def normal_fn():
    print(f"Here is normal function")


async def async_fn():
    print(f"Here is async function")



# loop.run_until_complete(normal_fn) # This will be throw exception
loop.run_until_complete(async_fn()) # This will be throw exception


print(loop)

Best Practice

asyncio.run(my_async_fn())

Diagram

┌───────────────────────────────┐
│        Python Program         │
└───────────────────────────────┘
                 │
                 ▼
      ┌───────────────────────┐
      │   Main OS Thread      │
      └───────────────────────┘
                 │
   ┌─────────────┼───────────────────────────┐
   │             │                           │
   ▼             ▼                           ▼
Loop A       Loop B                      Loop C
(new_event_loop)  (new_event_loop)       (new_event_loop)
 ┌───────┐      ┌───────┐                ┌───────┐
 │ Idle  │      │ Idle  │                │ Idle  │
 └───────┘      └───────┘                └───────┘
   ▲
   │
   │    Only ONE loop can be RUNNING
   │    in this thread at a time:
   │
   ▼
┌─────────────────────────────────────────┐
│      Running Event Loop (e.g., A)       │
│  - Executes async tasks                 │
│  - Handles timers / sockets             │
│  - Manages await points                 │
└─────────────────────────────────────────┘

Threads and Even loop

┌────────────────────┐
│     Main Thread    │
└────────────────────┘
          │
          ▼
   Event Loop A
   (running or idle)


┌────────────────────┐
│   Worker Thread 1  │
└────────────────────┘
          │
          ▼
   Event Loop B
   (running or idle)

Important asyncio rules

Rule 1 — One running event loop per thread Each thread may have one running event loop.

Rule 2 — Event loop is NOT shared between threads If you try to use a loop created in Thread A while in Thread B → RuntimeError.

Rule 3 — Threads communicate via thread-safe asyncio functions

loop.call_soon_threadsafe(callback, arg)

Examples:

import threading
import asyncio

def thread_worker():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

    async def run():
        print("Hello from worker loop in thread:", threading.current_thread().name)

    loop.run_until_complete(run())
    loop.close()


from datetime import datetime
def thread_run_forever():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

    async def _run_forever():
        while True:
            await asyncio.sleep(1)

            print(f"Tick: {datetime.now()}")

    loop.create_task(_run_forever())
    loop.run_forever()



# Main thread
main_loop = asyncio.new_event_loop()
asyncio.set_event_loop(main_loop)

# Spawn worker thread
t = threading.Thread(target=thread_worker)
t.start()

async def main():
    print("Hello from main loop in thread:", threading.current_thread().name)

main_loop.run_until_complete(main())
main_loop.close()
t.join()


# Run thread forever ?
t_run_forever = threading.Thread(target=thread_run_forever)
t_run_forever.start()
t_run_forever.join()

Implement signal stop


import threading
import asyncio
import signal
import time

stop_event = threading.Event()  # Event to signal thread stop

def thread_worker():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

    async def periodic():
        while not stop_event.is_set():  # keep running until stop_event is set
            print("Tick from worker thread")
            await asyncio.sleep(1)



    # Stop loop when stop_event is set
    def check_stop():
        if stop_event.is_set():
            loop.stop()
        else:
            loop.call_later(0.1, check_stop)  # check again after 0.1s
            # pass


    loop.call_soon(check_stop)
    loop.create_task(periodic())



    loop.run_forever()
    # loop.run_until_complete(periodic())



    loop.close()
    print("Worker thread stopped.")

# Start worker thread
t = threading.Thread(target=thread_worker, name="worker")
t.start()

# Signal handler (Ctrl+C sends SIGINT; terminals do not send Ctrl+X as a signal)
def signal_handler(sig, frame):
    if sig == signal.SIGINT:
        print("\nCtrl+C pressed, stopping...")
    elif sig == signal.SIGTERM:
        print("\nSIGTERM received, stopping...")
    stop_event.set()

signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)

# Keep main thread alive without busy-waiting
try:
    # Wait until stop_event is set by a signal
    while not stop_event.wait(0.1):
        pass
except KeyboardInterrupt:
    # Fallback: in case default KeyboardInterrupt is raised
    stop_event.set()

t.join()
print("Main thread exiting.")

I use call_soon, call_later to call check_stop which monitor the stop_signal event in global scope
In check_stop function, we can continue register a function can be executed in the loop. By logical, it register its self to keep continue checking signal for every 0.1 seconds
We can have other way to implement without using call_soon, call_later by register other task which have a loop to monitor stop_signal

# ... In the `thread_worker`
async def check_stop_task():
        while not stop_event.is_set():
            await asyncio.sleep(0.1)

        loop.stop()

# Then register this task
loop.create_task(check_stop_task())

Event loop in `async`

while True:

    # Phase 1 — RUN immediate callbacks
    while loop._ready:
        handle = loop._ready.pop(0)
        handle._run()

    # Phase 2 — WAIT for I/O until next timer
    timeout = compute_timeout_from_scheduled()
    events = selector.select(timeout)

    # Convert I/O events into immediate callbacks
    for event in events:
        loop._ready.append(event.handle)

    # Phase 3 — MOVE expired timers into ready queue
    now = loop.time()
    while loop._scheduled and loop._scheduled[0].when <= now:
        timer = heapq.heappop(loop._scheduled)
        loop._ready.append(timer)

On this page