Channels and Pipelines
Channel[T] connects concurrent producers and consumers with explicit sealing behavior.
Channel[T]
channel(capacity) creates a bounded channel for values of type T.
func main(args: List[String]) -> Void:
ch: Channel[Int] = channel(10)
Send/Recv
send blocks when the channel is full; recv returns Option[T] and returns None when the channel is sealed and drained.
_ = send(ch, 42)
value: Option[Int] = recv(ch)
Seal
seal tells receivers that no more values will arrive. close remains as a compatibility alias.
seal(ch)
match recv(ch):
None: print("sealed")
Some(v): print(v)
For-In Over Channels
A for loop over a channel receives values until the channel is sealed and drained.
for value in ch:
total += value
Typed Attempts
wait_send and wait_recv expose sealing through Result. Attempts distinguish values, would-block, sealed, and timed-out states.
import:
channel:
RecvAttempt(RecvSealed, RecvTimedOut, RecvValue, RecvWouldBlock),
recv_timeout_attempt_for,
try_recv_attempt,
units: milliseconds
func describe_receive(ch: Channel[Int]) -> String:
match try_recv_attempt(ch):
RecvValue(value):
"value: ${value}"
RecvWouldBlock:
"empty"
RecvSealed:
"sealed"
RecvTimedOut:
"timed out"
func describe_timeout() -> String:
ch: Channel[Int] = channel(1)
match recv_timeout_attempt_for(ch, milliseconds(1)):
RecvValue(value):
"value: ${value}"
RecvWouldBlock:
"empty"
RecvSealed:
"sealed"
RecvTimedOut:
"timed out"
Select
select runs one ready arm in the current task: a value, a sealed-and-drained channel, or a deadline. It is a statement, not a value expression.
import:
units: milliseconds
func wait_for_message(ch: Channel[String]) -> String:
var result: String = "timeout"
select:
message from ch:
result = "message: ${message}"
sealed ch:
result = "done"
_ after milliseconds(100):
result = "timeout"
result
Example
Producers and the consumer run together through a capacity-one channel. One coordinator seals it after both producers finish; the consumer drains every accepted value. Five messages sum to 10 regardless of arrival order. The outer deadline cancels and joins blocked work rather than reporting a partial sum as success.
import:
channel: ChannelSealed, ConcurrencyError, TaskResult, wait_send
units: seconds
union PipelineError:
ProducerSealed(ChannelSealed)
ProducerTask(ConcurrencyError)
func produce(ch: Channel[Int], start: Int, end: Int) -> Result[Void, ChannelSealed]:
var i: Int = start
var failure: Option[ChannelSealed] = None
while i < end and failure.is_none():
match wait_send(ch, i):
Ok(_): void
Err(error): failure = Some(error)
i += 1
match failure:
None: Ok(void)
Some(error): Err(error)
func consume(ch: Channel[Int]) -> Int:
var total: Int = 0
for n in ch:
total += n
total
func producer_outcome(result: TaskResult[Result[Void, ChannelSealed]]) -> Result[Void, PipelineError]:
match result:
Ok(Ok(_)): Ok(void)
Ok(Err(error)): Err(ProducerSealed(error))
Err(error): Err(ProducerTask(error))
func coordinate(ch: Channel[Int]) -> Result[Void, PipelineError]:
concurrent:
first = produce(ch, 0, 3)
second = produce(ch, 3, 5)
-- Only the coordinator seals, after both producers have joined.
seal(ch)
match producer_outcome(first):
Ok(_): producer_outcome(second)
Err(error): Err(error)
func main(args: List[String]) -> Int:
ch: Channel[Int] = channel(1)
-- The deadline bounds all work, including nested joins and blocked receives.
concurrent(timeout: seconds(5)):
coordinator = coordinate(ch)
consumer = consume(ch)
match (coordinator, consumer):
(Ok(Ok(_)), Ok(total)):
print(total.to_string()) -- prints: 10
0
(Ok(Err(ProducerSealed(_))), _):
print_error("producer channel was sealed")
1
(Ok(Err(ProducerTask(_))), _):
print_error("producer task failed or was cancelled")
1
(Err(_), _):
print_error("coordinator failed or was cancelled")
1
(Ok(Ok(_)), Err(_)):
print_error("consumer failed or was cancelled")
1
Try it
blorp run pipeline.brp