# Repeat previous pipe while condition?

**URL:** <https://forum.elixirforum.com/t/repeat-previous-pipe-while-condition/15160>\
**Category:** Questions\
**Tags:** troubleshooting\
**Created:** [July 16, 2018, 8:51am UTC](https://forum.elixirforum.com/t/repeat-previous-pipe-while-condition/15160 "2018-07-16T08:51:51Z")\
**Posts on this page:** 1\
**Showing post:** 15

<div class="post-metadata">

**Author:** ![peerreynders](https://forum.elixirforum.com/user_avatar/forum.elixirforum.com/peerreynders/32/5826_2.png) [@peerreynders](https://forum.elixirforum.com/u/peerreynders)\
**Post date:** [July 18, 2018, 4:06pm UTC](https://forum.elixirforum.com/t/repeat-previous-pipe-while-condition/15160/15 "2018-07-18T16:06:08Z")

</div>

> [@sabri](#):
>
> I saw the example, its using GenServer worker, I thought you mean `Task` with a function also, in the example your sent, I didn’t see `Task.Supervisor` … ?

`Task.Supervisor` is started as part of the application in `MyApp.Application.start/2`. It is then later used by [`Task.Supervisor.async_stream_nolink/6`](https://hexdocs.pm/elixir/Task.Supervisor.html#async_stream_nolink/6) within `Worker.stream_tasks/2`

`Worker` is simply a `GenServer` that needs results from multiple tasks - while being “safe” from them should they misbehave. `async` links to the spawning process so a crashing task will take down everything else.

`Worker.stream_fun` is the function being run as a task. The input list `[1000, -500, 2000]` results in three separate tasks:

1. Sleeps 1000 ms then returns an integer value of 1000
2. Sleeps 500 ms and then crashes
3. Sleeps 2000 ms and is terminated by the 1500 ms timeout

The `Worker` `GenServer` receives the completed results from all the tasks in

```Elixir
  def handle_info({task_ref, result}, {app, name, task_ref}) do
    # normal task result arrives here - demonitor/flush :normal :DOWN
    Process.demonitor(task_ref, [:flush])
    IO.puts("Result: #{inspect result}")
    {:stop, :normal, {app, name, nil}}
  end

```

which happens to be:

```Elixir
[ok: 1000, exit: :crash, exit: :timeout]

```

i.e. only the first task completed successfully while the others exited: due to a crash or was terminated via a timeout.

> [@dimitarvp](#):
>
> improved the code speed by 30 - 80%. Theoretically, with 4 cores the improvement should be 400% but oh well, it’s still better.

Joe Armstrong’s stated goal for the BEAM is:

> applications run 0.75 x N times faster on an N-core computer

[How we program multicores - Joe Armstrong](https://erlangcentral.org/blog/how-we-program-multicores-joe-armstrong-youtube/#.W091kNhKg6Q)

Also it’s not too hard to imagine that the `async_stream` way of doing things is likely highly optimized while there are lots of ways to foul up a bunch of hand-coded `async`s.

> There are a number of kernel parameters (like FD limits) that can make a code that downloads 1000 URLs fast or very slow.

My intent wasn’t to advocate boundless process spawning. I can merely see the attraction of simply firing up “a few” tasks via `async` and then using `Enum.map/2` in combination with `Task.await/2` as a kind of `await_all` - especially if you aren’t particularly concerned about what happens when one of the tasks crashes.

[`Task.Supervisor.async_stream_nolink/6`](https://hexdocs.pm/elixir/Task.Supervisor.html#async_stream_nolink/6) seems like a much more adept solution just for the built-in fault isolation and capture - it’s easy enough to set `:max_concurrency` to a high enough value to adjust for a _reasonable_ amount of concurrency given the job at hand (e.g. waiting for a bunch of blocked requests).

---

_[View the full topic](https://forum.elixirforum.com/t/repeat-previous-pipe-while-condition/15160)._