eaa5d3498c
This PR is a follow up to #8055 and implements a Selector for `Std.Channel` in order to allow multiplexing using channels. There is one subtlety to the implementation: Suppose we are in a situation where we run `select` in a loop on two channels. One of the channels is always quiet while the other has data available occasionally (however not always as this would trigger the `tryFn` fast path and hide the issue). In this situation the select receivers that are enqueued on the silent channel would usually just remain there indefinitely as nothing ever happens, causing a memleak. To avoid this we want to make a channel select clean up after itself, even if it fails. In an imperative programming language we could implement the receive queue as a doubly linked list and simply make each receive select maintain a pointer to its element in the queue and then remove itself in `O(1)` upon failure. As that is not possible in Lean trivially we decided to go for another approach for now: simply filter the queue for selects that have failed in `unregisterFn`. While this approach is `O(n)` we expect the amount of receivers enqueued on a channel to not be terribly large and thus this to be a reasonably fast operation compared to the remaining overhead. If it ever ends up becoming an issue, we could switch to an approach that uses a `TreeMap` with numbered receivers instead at a certain wait queue size and go to `O(log(n))`.
99 lines
2.3 KiB
Text
99 lines
2.3 KiB
Text
import Std.Sync.Channel
|
|
|
|
open Std Internal IO Async
|
|
|
|
namespace A
|
|
|
|
def testReceiver (ch1 ch2 : Std.Channel Nat) (count : Nat) : IO (AsyncTask Nat) := do
|
|
go ch1 ch2 count 0
|
|
where
|
|
go (ch1 ch2 : Std.Channel Nat) (count : Nat) (acc : Nat) : IO (AsyncTask Nat) := do
|
|
match count with
|
|
| 0 => return AsyncTask.pure acc
|
|
| count + 1 =>
|
|
Selectable.one #[
|
|
.case ch1.recvSelector fun data => go ch1 ch2 count (acc + data),
|
|
.case ch2.recvSelector fun data => go ch1 ch2 count (acc + data),
|
|
]
|
|
|
|
def testIt (capacity : Option Nat) : IO Bool := do
|
|
let amount := 1000
|
|
let messages := Array.range amount
|
|
let ch1 ← Std.Channel.new capacity
|
|
let ch2 ← Std.Channel.new capacity
|
|
let recvTask ← testReceiver ch1 ch2 amount
|
|
|
|
for msg in messages do
|
|
if (← IO.rand 0 1) = 0 then
|
|
ch1.sync.send msg
|
|
else
|
|
ch2.sync.send msg
|
|
|
|
let acc ← recvTask.block
|
|
return acc == messages.sum
|
|
|
|
/-- info: true -/
|
|
#guard_msgs in
|
|
#eval testIt none
|
|
|
|
/-- info: true -/
|
|
#guard_msgs in
|
|
#eval testIt (some 0)
|
|
|
|
/-- info: true -/
|
|
#guard_msgs in
|
|
#eval testIt (some 1)
|
|
|
|
/-- info: true -/
|
|
#guard_msgs in
|
|
#eval testIt (some 128)
|
|
|
|
end A
|
|
|
|
namespace B
|
|
|
|
def testReceiver (ch1 ch2 : Std.CloseableChannel Nat) (count : Nat) : IO (AsyncTask Nat) := do
|
|
go ch1 ch2 count 0
|
|
where
|
|
go (ch1 ch2 : Std.CloseableChannel Nat) (count : Nat) (acc : Nat) : IO (AsyncTask Nat) := do
|
|
match count with
|
|
| 0 => return AsyncTask.pure acc
|
|
| count + 1 =>
|
|
Selectable.one #[
|
|
.case ch1.recvSelector fun data => go ch1 ch2 count (acc + data.getD 0),
|
|
.case ch2.recvSelector fun data => go ch1 ch2 count (acc + data.getD 0),
|
|
]
|
|
|
|
def testIt (capacity : Option Nat) : IO Bool := do
|
|
let amount := 1000
|
|
let messages := Array.range amount
|
|
let ch1 ← Std.CloseableChannel.new capacity
|
|
let ch2 ← Std.CloseableChannel.new capacity
|
|
let recvTask ← testReceiver ch1 ch2 amount
|
|
|
|
for msg in messages do
|
|
if (← IO.rand 0 1) = 0 then
|
|
ch1.sync.send msg
|
|
else
|
|
ch2.sync.send msg
|
|
|
|
let acc ← recvTask.block
|
|
return acc == messages.sum
|
|
|
|
/-- info: true -/
|
|
#guard_msgs in
|
|
#eval testIt none
|
|
|
|
/-- info: true -/
|
|
#guard_msgs in
|
|
#eval testIt (some 0)
|
|
|
|
/-- info: true -/
|
|
#guard_msgs in
|
|
#eval testIt (some 1)
|
|
|
|
/-- info: true -/
|
|
#guard_msgs in
|
|
#eval testIt (some 128)
|
|
|
|
end B
|