58c7e5da94
This PR adds an implementation of an async IO multiplexing framework as well as an implementation of it for the `Timer` API in order to demonstrate it. The main motivation is to have fair and data loss free multiplexing of event sources. To illustrate two situations where just naively racing two tasks that read from an event source might be the wrong thing: 1. Suppose we are waiting on two channel reads that are continuously being filled up. As the first channel will always be ready when we start its receive function it will instantly resolve the race before the second one can even try. Thus the path where we receive data from the second channel gets starved. For this reason we want to try in random order (for fairness) if the event sources already have data available for us. 2. Suppose we are waiting on two socket reads and both happen to finish at the same time. As we are now only going to select one of them to execute further, we are going to loose data on the second one (unless there is a user written buffering mechanism involved) as we are going to disregard the buffer it received and do a new receive next time. For this reason it is important to wait for an event source to be available without committing to actually fetching some data until we know that this particular event source is going to win the select race. The implementation is inspired by the Oslo framework written by @haesbaert as well as Go's [`select`](https://go.dev/src/runtime/select.go) implementation. Given a list of event sources to select one from it is going to: 1. Randomly shuffle them 2. Attempt to fetch data from them (in their new random order) without blocking (for fairness). If any of them succeeds return right away. 3. If none has data available right away set all of them up to resolve a promise. They will then race to win the right to resolve that promise. Only the data source that wins the race is allowed to then actually fetch data, ensuring that no other event source actually fetches data and then fails to deliver it to the consumer. Follow up PRs are going to add implementations of `Selector` for `Std.Channel` as well as TCP and UDP sockets. --------- Co-authored-by: Markus Himmel <markus@lean-fro.org>
29 lines
700 B
Text
29 lines
700 B
Text
import Std.Internal.Async.Timer
|
|
|
|
open Std Internal IO Async
|
|
|
|
def test1 : IO (AsyncTask Nat) := do
|
|
let s1 ← Sleep.mk 1000
|
|
let s2 ← Sleep.mk 1500
|
|
Selectable.one #[
|
|
.case (← s2.selector) fun _ => return AsyncTask.pure 2,
|
|
.case (← s1.selector) fun _ => return AsyncTask.pure 1,
|
|
]
|
|
|
|
/-- info: 1 -/
|
|
#guard_msgs in
|
|
#eval show IO _ from do
|
|
let task ← test1
|
|
IO.ofExcept task.get
|
|
|
|
def test2 : IO (AsyncTask Nat) := do
|
|
Selectable.one #[
|
|
.case (← Selector.sleep 1500) fun _ => return AsyncTask.pure 2,
|
|
.case (← Selector.sleep 1000) fun _ => return AsyncTask.pure 1,
|
|
]
|
|
|
|
/-- info: 1 -/
|
|
#guard_msgs in
|
|
#eval show IO _ from do
|
|
let task ← test2
|
|
IO.ofExcept task.get
|