actor system: low-level signal layer about concurrency-utils HOT 2 CLOSED

Turns out, just using a coroutine to interact with the mailbox while having a separate queue for signals is not enough. Both queue must be waited on together, and blocking on one prevents the other from being processed. A few approaches

• All blocking calls on queues(signal or mailbox) must use poll looping. A receive call on the mailbox would always be implemented by doing a poll loop(e.g. doing a blocking get with a small timeout value) on the queue, and yielding on each loop to also check on the signal queue. The problem is that this would make implementing timeout on the receive call difficult(it already is broken though, since a receive with a timeout can block multiple times on the queue, as long as each individual block is less than the timeout).
• Implement a "selectable queue" that would enable usage of select.select to implement simultaneous waiting on the two queues. This would involve associating os file descriptors to queues, and overriding write methods(e.g. put, put_nowait) to manipulate the file descriptors.
• Use a single queue: signals would go through the mailbox as well, but the implementation would make sure signals are processed first independently of user expectations.
• Use an async framework(e.g. Trio) as underlying implementation to manage all events(signals, messages). The user code doesn't need to know: it can still be a coroutine generator and use sync code. The coroutine would only need to yield often enough to allow the framework to handle events. This is always true anyway.

from concurrency-utils.

The third approach was implemented as of ff50961 using asyncio. A Trio-based reimplementation could be attempted.

• Each ThreadActor runs its own asyncio event loop
• signal queue and mailbox queue become asyncio.Queue.
• Handling Receive calls involve simultaneous waiting on the signal queue and the mailbox queue, using concurrency.utils.select(itself implemented using concurrency.utils.race).
• User-specified Receive timeouts are implemented using asyncio.wait_for on the queue iteration, which means it has more correct timeout semantics.
• System methods that interact with actors(e.g. sending a mailbox message using send, sending signals using send_signal) now use asyncio.run_coroutine_threadsafe to interact with asyncio.Queue objects in the actor's thread.

from concurrency-utils.