# Eio, streams, clean termination

**URL:** <https://discuss.ocaml.org/t/eio-streams-clean-termination/14811>\
**Category:** Learning\
**Tags:** concurrency, eio\
**Created:** [June 18, 2024, 6:21am UTC](https://discuss.ocaml.org/t/eio-streams-clean-termination/14811 "2024-06-18T06:21:15Z")\
**Posts on this page:** 5\
**Page:** 1

<div class="post-metadata">

**Author:** ![R\_Huxton](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.ocaml.org/r_huxton/32/4216_2.png) [@R\_Huxton](https://discuss.ocaml.org/u/R_Huxton)\
**Post date:** [June 18, 2024, 6:21am UTC](https://discuss.ocaml.org/t/eio-streams-clean-termination/14811/1 "2024-06-18T06:21:15Z")

</div>

Imagine a system with a producer reading an indeterminate batch of jobs, which  
are processed by workers before being consumed. Each job might generate 0…n  
results and we don’t particularly care about order. At the end of a batch it  
would be useful for the consumer to log some statistics of results processed.

I currently have code something like this:

```auto
  let work_stream = Eio.Stream.create n in
  let results_stream = Eio.Stream.create n in
  Eio.Fiber.all [
    (fun () -> work_producer id_source work_stream);
    (fun () -> workers work_stream results_stream);
    (fun () -> results_consumer results_stream result_sender);
  ]

```

Now, reading from a  
[stream](https://ocaml-multicore.github.io/eio/eio/Eio/Stream/index.html) is blocking and there is no “EOF” equivalent - you can’t close streams. That means we need to know there are no more results going to be available or our `results_consumer` will block forever.

It’s not enough to know that the producer has finished reading from its source,  
you also need to know there are no active workers or queued jobs or results.

I have ended up with a set of mutex-protected counters (one per stream and fiber) and when all (except the consumer) reach zero then we are finished.

But… that feels overly complicated. What am I missing? Were streams the wrong tool for the job? Is the pipeline structured incorrectly?

TIA

---

<div class="post-metadata">

**Author:** ![silene](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.ocaml.org/silene/32/2707_2.png) [@silene](https://discuss.ocaml.org/u/silene)\
**Post date:** [June 18, 2024, 7:50am UTC](https://discuss.ocaml.org/t/eio-streams-clean-termination/14811/2 "2024-06-18T07:50:42Z")

</div>

Have the producer write in the result stream whenever it submits a new job and also when it is done submitting new jobs. Have the workers write whenever they complete a job. Then the consumer can just count the number of events in the result stream.

---

<div class="post-metadata">

**Author:** ![talex5](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.ocaml.org/talex5/32/353_2.png) [@talex5](https://discuss.ocaml.org/u/talex5)\
**Post date:** [June 18, 2024, 10:37am UTC](https://discuss.ocaml.org/t/eio-streams-clean-termination/14811/3 "2024-06-18T10:37:58Z")

</div>

Another option would be to write one “stop” message to `work_stream` for each worker. Each worker exits after reading a stop message, so each will only take one.

(note: `Eio.Executor_pool` cheats; it uses an internal stream API that does provide `close`. I’d like to make this public in the future.)

---

<div class="post-metadata">

**Author:** ![R\_Huxton](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.ocaml.org/r_huxton/32/4216_2.png) [@R\_Huxton](https://discuss.ocaml.org/u/R_Huxton)\
**Post date:** [June 18, 2024, 4:14pm UTC](https://discuss.ocaml.org/t/eio-streams-clean-termination/14811/4 "2024-06-18T16:14:52Z")

</div>

Thanks for the suggestions both - I’ll try them out at some point in the next few days.

Was there a particular reason for not exposing `close` in the standard `Switch` api @talex5 ? Or is it just that it is always easier to expand an API’s interface than to reduce an overly large one?

---

<div class="post-metadata">

**Author:** ![talex5](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.ocaml.org/talex5/32/353_2.png) [@talex5](https://discuss.ocaml.org/u/talex5)\
**Post date:** [June 19, 2024, 10:14am UTC](https://discuss.ocaml.org/t/eio-streams-clean-termination/14811/5 "2024-06-19T10:14:37Z")

</div>

> [@R\_Huxton](#):
>
> Was there a particular reason for not exposing `close` in the standard `Switch` api @talex5 ?

Internally, `Stream` has two implementations, depending on whether you ask for a 0-capacity stream or not. So far, I only added `close` to the 0-capacity one (but doing the other would be easy). We might also want separate types for closable and non-closable streams.

And then if we’re changing the API then we might want other changes too (see [Consider renaming Stream · Issue #267 · ocaml-multicore/eio · GitHub](https://github.com/ocaml-multicore/eio/issues/267))… just needs a bit of thinking about.