Gain insights into data processing and concurrency in Elixir. Explore GenServer, GenStage, Flow, and Broadway to enhance performance, set up pipelines, and handle increased workloads efficiently.

Concurrent Data Processing in Elxir by The Pragmatic Programmer-01.png

myDokcer.tar.gz

ch1_1

ch_3

ch_3.1

ch1_1-task supervisor

ch_4.1

ch_5.1 RabbitMQ

Ch_5.2 RabbitMQ

Most engineers don’t think that concurrency is distinct from programming. Data processing includes transforming information in some way, like reporting, data aggregation, or analytics. On the other hand, concurrency involves significantly improving performance.

This course is designed to help you navigate the world of concurrency tools available in the Elixir ecosystem. You’ll explore popular modules and libraries like GenServer, GenStage, Flow, and Broadway. You’ll be able to ingest data, set up pipelines, and run concurrent processes with Elixir.

By the end of the course, you’ll have a wide range of new techniques to simplify your product, improve code performance, and make your application more resilient to increased workloads.

Concurrent Data Processing in Elixir

To retrieve the task result, we can use the `Task.await/1` or the `Task.yield/1` function. Both accept a `Task` struct as an argument. There is an essential difference in how `await/1` and `yield/1` work, so we have to choose wisely. They both stop the program and try to retrieve the task’s result. The difference comes from the way they handle process timeouts.

# The `await` function

Process timeouts ensure that processes don’t get stuck waiting forever. Let’s increase the time of the `send_email/1` function to 30 seconds:

Here is the executable command:

Task.async(fn -> Sender.send_email("hi@world.com") end) |> Task.await()

It will now run the code asynchronously and pipe the task result into `await/1`.

This is the output we get: 

```shell
iex(1)> Task.async(fn -> Sender.send_email("hi@world.com") end) |> Task.await()
** (exit) exited in: Task.await(%Task{owner: #PID<0.163.0>, pid: #PID<0.165.0>, ref: #Reference<0.3155108343.272891905.32384>}, 5000)
    ** (EXIT) time out
    (elixir) lib/task.ex:607: Task.await/2
```

After five seconds, we receive an exception similar to the one above.

When using `await/1`, we expect a task to finish within a certain amount of time.

* By default, this time is set to 5000ms, which is five seconds.

* We can change that by passing an integer with the number of milliseconds as a second argument—for example, `Task.await(task, 10000)`.

* We can also disable the timeout by passing the `:infinity` atom.

# The `yield` function

* It simply returns `nil` if the task hasn’t been completed.

* The timeout of `yield/1` is also 5000ms but it does not cause an exception and crash.

* We can also use `Task.yield(task)` repeatedly to check for a result, which is not allowed by `await/1`.

* A completed task will return either `{:ok, result}` or `{:exit, reason}`.

Now, we execute the following commands by calling `Task.yield` after different intervals after the first command.

Here is the executable command: 


Learn to retrieve results using the task await and task yield functions from task results.

Getting Started

Easy Concurrency with the Task Module

Long-Running Processes Using GenServer

Data Processing Pipelines with GenStage

Process Collections with Flow

Data Ingestion Pipelines with Broadway

Concluding the Course

Use the await and yield Functions to Retrieve Results