Async + Event Loop

[timkpaine]

Putting this up for some discussion, we've been talking/integrating csp with lots of other frameworks, and many of them are based on asyncio. This PR adds some bridging utilities. For now they are mostly one direction, but we can also add the other direction which should be easy (e.g. running csp in a background thread and making edges awaitable / async generators).

import csp
from csp import ts
from datetime import timedelta, datetime
from typing import AsyncIterator
import asyncio


async def async_in() -> int:
    await asyncio.sleep(0.1)
    return 42

async def async_out(n: int) -> None:
    await asyncio.sleep(0.1)
    print(f"Output: {n}")

async def async_node(n: int) -> int:
    await asyncio.sleep(0.1)
    return n * 2

async def async_counter_node(n: int) -> AsyncIterator[int]:
    for i in range(n):
        await asyncio.sleep(0.1)
        yield i

@csp.node
def csp_counter_node() -> ts[int]:
    with csp.alarms():
        tick_alarm = csp.alarm(bool)

    with csp.state():
        s_counter = 0

    with csp.start():
        csp.schedule_alarm(tick_alarm, timedelta(), True)

    if csp.ticked(tick_alarm):
        s_counter += 1
        csp.schedule_alarm(tick_alarm, timedelta(seconds=0.1), True)
        return s_counter


@csp.node
def csp_async_node(trigger: ts[int]) -> ts[int]:
    """
    Example node that uses async operations with the alarm pattern.
    The completed coroutine wakes the node itself, so no polling alarm is needed.
    """
    with csp.alarms():
        async_alarm = csp.async_alarm(int)

    if csp.ticked(trigger):
        # Schedule async operation to double the counter
        csp.schedule_async_alarm(async_alarm, async_node(trigger))

    if csp.ticked(async_alarm):
        # Async operation completed
        return async_alarm


@csp.graph
def graph():
    csp_counter = csp_counter_node()
    async_counter = csp.async_for(async_counter_node(15))

    csp.print("counter", csp_counter)
    csp.print("async_counter", async_counter)

    # async_in: coroutine that ticks once when ready
    async_in_result = csp.async_in(async_in())
    csp.print("async_in", async_in_result)

    # async_out: invoke async function when input ticks
    csp.async_out(csp_counter, async_out)

    # async_node: takes input, runs async, outputs result
    async_node_result = csp.async_node(csp_counter, async_node)
    csp.print("async_node", async_node_result)

    # async for within CSP node
    csp.print("csp_async_node", csp_async_node(csp_counter))

if __name__ == "__main__":
    csp.run(graph, realtime=True, endtime=timedelta(seconds=2))

[timkpaine]

5 critical issues fixed:

• Ctrl-C is broken for realtime runs; process_one_cycle violates the Python C-API contract
• PyEventLoopAdapter::callSoonThreadsafe touches the Python C-API without the GIL
• PyEventLoop.{h,cpp} is dead code
• FdWaiter lost wakeup: the de-dup flag is incompatible with clear-after-process
• FdWaiter is destroyed while producer threads may be inside notify()

[timkpaine]

More key fixes:

• call_soon_threadsafe never wakes the loop (self-pipe never created)
• CspEventLoop restarts an already-finished engine
• Adapter threads can push_tick() into a stopped engine
• await_(..., loop=current_loop) deadlocks the loop thread
• _run_on_async_loop deadlocks on cancellation
• Positive-delay asyncio timers never fire in simulation mode
• m_cycleCount now counts cycles that do no work
• Windows socket handles are truncated to int
• macOS pipe fds are not close-on-exec
• nextScheduledTime() returns a sentinel that never compares equal to None
• Pervasive silent exception swallowing
• Async-alarm codegen materialises a type by name lookup
• visit_Name rewrites async-alarm names in Store/Del context
• get_shared_loop() can create two loops and orphan a thread
• Unbounded queues and task lists, no backpressure
• Socket-operation cancellation leaks selector registrations
• _scheduled timers: cancelled handles never reaped, pop(0) is O(n)
• CspEventLoop.time() switches clock domains mid-run
• Timing-based performance assertions will flake in CI
• run_on_thread silently opts into the new async path
• _run_asyncio_engine's error path masks the original exception