DEV Community: Tracey Onim

Leverage Concurrency Efficiently When Managing Multiple Tasks in Elixir

Tracey Onim — Tue, 09 Jul 2024 10:53:15 +0000

When working on multiple tasks, it's important to consider performing them concurrently. However, when using concurrency, we need to be careful not to overload our system resources.

In this article, we will cover the following:

What is concurrency?
How can we use concurrency in Elixir?
Common issues when working with a series of tasks concurrently.
How to efficiently manage a series of tasks using the Task module.

Let's get going!

What is Concurrency?

Concurrency is the ability to execute multiple tasks at the same time.

Concurrency and parallelism are crucial features in modern programming. The operating systems installed on our computers are designed to incorporate these features. Both concurrency and parallelism involve executing multiple tasks simultaneously. However, concurrency focuses on managing multiple tasks within one resource. This can be achieved through techniques like multitasking and asynchronous programming. In an operating system, concurrency can be achieved by opening multiple applications to perform different tasks at the same time, also referred to as multitasking.

On the other hand, parallelism involves executing multiple tasks simultaneously across multiple processing units (such as multiple CPU cores or distributed computing resources), thus improving system speed.

How Can We Use Concurrency in Elixir?

To achieve concurrency in Elixir, you need to initiate a process and execute the function or code responsible for performing a specific task within that process.

Here's an example of some basic Elixir code that we can take a look at.

def send_message(phone_number) do
  # Simulate some processing time
  Process.sleep(3000)
  IO.puts("Sending message to #{phone_number}")
  {:ok, "sent"}
end

In this example, we've added a 3-second sleep to simulate the fact that sending a message to an external phone number can sometimes cause some latency.

When send_message/1 is invoked, this function will delay processing for 3 seconds before printing a message. It will also return an {:ok, "sent"} tuple to indicate that the message was successfully sent to the given phone number.

When executing synchronous Elixir code like the code above, we must wait for it to complete before receiving the output. If we need to send messages to multiple phone numbers, each with a delay of 3 seconds, it could take a significant amount of time to receive a notification that all messages have been successfully sent.

Taking It One Step Further

Now, let's take it a step further and write some code that sends a message to multiple phone numbers.

def send_to_all(phone_numbers) do
  Enum.each(phone_numbers, &send_message/1)
end

The send_to_all/1 function takes a list of phone numbers as an argument and uses the Enum.each/2 function to iterate over the list, sending a message to each phone number using the send_message function.

If you run this function in iex, you will notice two things:

It takes longer to receive the final output, which is an atom :ok. Enum.each/2 returns :ok when invoked.
Printing happens one at a time, with each print taking 3 seconds. Depending on the number of phone numbers provided, send_message/1 is called 5 times in our case, taking about 15 seconds to complete.

This means that when sending a message to a given phone number, each phone number has to wait for the preceding phone number to execute before moving on to the next one. The execution is synchronous, also known as blocking code.

Elixir provides the ability to run code concurrently or asynchronously. Asynchronous code allows you to execute multiple tasks simultaneously. This type of code is also referred to as non-blocking code because it does not hinder the primary execution of a program.

Let's improve the send_to_all/1 function by running the task of sending messages to each phone number asynchronously. We need to start a process for each task.

A process is a separate entity where code execution occurs. These processes are lightweight, run concurrently, and are isolated from each other. The code we write runs inside processes. The IEx (interactive shell) is an example of an Elixir process.

def send_to_all(phone_numbers) do
  Enum.each(phone_numbers, fn phone_number ->
    Task.start(fn -> send_message(phone_number) end)
  end)
end

Here, we've used Task.start/1 (which we'll go over in more detail later) to start a process for each task.
After we recompile and run this function again in our IEx, you will notice two things:

It runs faster than before because it's now running asynchronously. The final output — :ok — is immediately returned even before we see the printed success messages.
It instantly prints the success message at once for all tasks, which means that all the messages are being sent at the same time.

Running Tasks Concurrently using the Task Module

Elixir has a standard library Task module commonly used for running code concurrently. This allows you to perform tasks asynchronously, which can lead to improved performance and responsiveness in your applications.

Let's go back to our send_to_all/1 function. At first, we used this function to send messages to multiple phone numbers sequentially.

# Sequential version of send_to_all/1
def send_to_all(phone_numbers) do
  Enum.each(phone_numbers, fn phone_number ->
    Task.start(fn -> send_message(phone_number) end)
  end)
end

In the sequential version, send_message/1 was called for each message in the list, one after the other.

Later on, we converted the sequential code into concurrent code using Task.start/1.

# Asynchronous version using Task.start/1
def send_to_all(phone_numbers) do
  Enum.each(phone_numbers, fn phone_number ->
    Task.start(fn -> send_message(phone_number) end)
  end)
end

In the asynchronous version, Task.start/1 creates a separate task for each message, allowing the tasks to execute concurrently.

The Task module has several useful functions to run code concurrently. Let's now look at a few of these functions.

`Task.start/1`

Task.start/1 is used to start a new process/task. It takes in an anonymous function with zero arity as an argument, where the intended work is performed. Usually, it doesn't return the result of the executed function. Instead, it returns an {:ok, PID} tuple, where :ok means the process was started successfully, and PID stands for process identifier: a number that uniquely identifies an Elixir process.

Let's try it out in IEx:

iex > {:ok, pid} = Task.start(fn ->  IO.puts("converting to asyncronous code") end)
         converting to asyncronous code
          {:ok, #PID<0.153.0>}

The message is printed instantly and the result is {:ok, #PID<0.153.0>}. This means that the process was started successfully. pid represents the process started by Task.start/1.

However, it's hard to know if all the messages were sent successfully using Task.start/1, since it will still return {:ok, PID}, which stands for a process that's started successfully.

It is useful in cases where you don't have interest in the returned result or whether it completes successfully.

A good example is working on a background job processing system, such as sending messages to registered users every midnight. You can use Task.start/1 to asynchronously handle each task that sends messages to users. In this context, you're not particularly interested in the return value of Task.start/1, because your main concern is to handle the tasks asynchronously, and you have a mechanism to track the status of each task. This mechanism can involve saving the status in the database, ensuring that any pending tasks are retried in subsequent runs.

In case something goes wrong when sending messages to users, the status of the message will still be pending in the database. You can guarantee that the worker will run again at midnight and retry for pending messages.

`Task.async/1`

Like Task.start/1, Task.async/1 is also used to start a process. The difference between Task.async/ and Task.start/ is that with Task.async/1, you can retrieve the function result executed within the process.
It takes in an anonymous function with zero arity which starts a process and then returns a %Task{} struct.

Let's try it out in IEx:

iex > Task.async(fn -> IO.puts("converting to asyncronous code using async") end)
    converting to asyncronous code using async
      %Task{
        mfa: {:erlang, :apply, 2},
        owner: #PID<0.109.0>,
        pid: #PID<0.110.0>,
        ref: #Reference<0.0.13955.2933390707.3162308609.257437>
      }

The message is printed immediately and returns a %Task{} struct as the output. The returned Task struct includes useful information about the process that's started:

owner represents the Process Identifier (PID) of the process that started the Task. In our example, the owner is IEx since we are running code within the shell. If we run self(), the function that returns the PID of the current process, then the PID returned by self() will indeed match the PID of the owner of the task started.
pid: The process identifier of the process started by the task. You can use this PID to monitor the process further if needed.
ref: This is a process monitor reference that can be used in cases where you want to be notified about how and where a process exits.

To retrieve a result from the returned task, use either Task.await/2 or Task.yield/2. They both accept a %Task{} struct as an argument and handle process timeout. However, they handle the timeout differently.

Task.await/2 and Task.yield/2 have a default timeout of 5000ms or 5 seconds to ensure that processes don't get stuck waiting for a result forever. Task.await/2 will cause an exception and crash the process if there is a slower task taking more than 5000ms to complete, while Task.yield/2 returns nil.

Let's modify send_to_all/1 to use Task.async/1, then retrieve its result using Task.await/2. Later, use Task.yield/2:

def send_to_all(phone_numbers) do
  phone_numbers
  |> Enum.map(fn phone_number ->
    Task.async(fn -> send_message(phone_number) end)
  end)
  |> IO.inspect(label: "started tasks++++")
  |> Enum.map(fn task -> Task.await(task) end)
end

send_to_all/1 iterates over a given list of phone numbers using Enum.map/2 instead of Enum.each/2 to retrieve the result of each task, and spawns a task for each using Task.async/1. Then we use Task.await/2 to retrieve the result of each task. It waits for the task to complete and then returns a result.

Run the send_to_all/1 function in IEx:

iex > send_to_all(1..10)
started tasks++++[
  %Task{
    mfa: {:erlang, :apply, 2},
    owner: #PID<0.150.0>,
    pid: #PID<0.151.0>,
    ref: #Reference<0.0.19203.3913757788.1002242054.113465>
  },
  %Task{
    mfa: {:erlang, :apply, 2},
    owner: #PID<0.150.0>,
    pid: #PID<0.152.0>,
    ref: #Reference<0.0.19203.3913757788.1002242054.113469>
  },..
]

sending message  to 1
sending message  to 2
sending message  to 3
sending message  to 4
sending message  to 5
sending message  to 6
sending message  to 7
sending message  to 8
sending message  to 9
sending message  to 10

[
  {:ok, "sent"},
  {:ok, "sent"},
...
]

Looking at the inspected result, each task represents a spawned process. They all have a different pid with a similar owner pid meaning that the caller process (IEx) is one. Once each task is completed, the messages are printed at the same time and the result returned. The final result returned is a list of output expected from send_message/1.

Now, let's retrieve the result from each spawned task using Task.yield/2:

def send_to_all(phone_numbers) do
  phone_numbers
  |> Enum.map(fn phone_number ->
    Task.async(fn -> send_message(phone_number) end)
  end)
  |> IO.inspect(label: "started tasks++++")
  |> Enum.map(fn task -> Task.yield(task) end)
end

iex > send_to_all(1..10)
started tasks++++[
  %Task{
    mfa: {:erlang, :apply, 2},
    owner: #PID<0.150.0>,
    pid: #PID<0.151.0>,
    ref: #Reference<0.0.19203.3913757788.1002242054.113465>
  },
  %Task{
    mfa: {:erlang, :apply, 2},
    owner: #PID<0.150.0>,
    pid: #PID<0.152.0>,
    ref: #Reference<0.0.19203.3913757788.1002242054.113469>
  },..
]

sending message  to 1
sending message  to 2
sending message  to 3
sending message  to 4
sending message  to 5
sending message  to 6
sending message  to 7
sending message  to 8
sending message  to 9
sending message  to 10

[
  :ok, {:ok, "sent"},
  :ok, {:ok, "sent"},
...
]

The result is similar. However, Task.yield/2 returns an {:ok, term()} tuple, where term represents the result returned by the function executed within the process. That's why our output is a list of the {:ok, {:ok, "sent"}} tuple where {:ok, "sent"} is the expected result from send_message/1.

In our send_to_all/1 function, we used Task.start/2 and Task.async/2 to start each task of sending a message to a given phone number in the background, allowing them to run concurrently. Starting a separate process for each task when sending a message ensures its isolation. Therefore, if one task encounters an error or fails for any reason, it won't affect the execution of other tasks.

An Example with an Error

For example, let's assume we have introduced an error in one of the tasks.

def send_message(3) do
  # Simulate some processing time
  Process.sleep(3000)
  IO.puts("Sending message to #{3 + "a"}")
  {:ok, "sent"}
end

def send_message(phone_number) do
  # Simulate some processing time
  Process.sleep(3000)
  IO.puts("Sending message to #{phone_number}")
  {:ok, "sent"}
end

# Asynchronous version using Task.start/1
def send_to_all(phone_numbers) do
  Enum.each(phone_numbers, fn phone_number ->
    Task.start(fn -> send_message(phone_number) end)
  end)
end

Here, the send_message/1 function has two function heads. The first function head pattern matches on 3 and deliberately raises an arithmetic error by adding the number 3 to a string "a". The second function head handles other phone numbers values and simulates sending messages as before.

iex > send_to_all(1..5)
06:19:17.562 [error] Task #PID<0.294.0> started from #PID<0.291.0> terminating
** (ArithmeticError) bad argument in arithmetic expression
    :erlang.+(3, "a")

[
  ok: #PID<0.292.0>,
  ok: #PID<0.293.0>,
  ok: #PID<0.294.0>,
  ok: #PID<0.295.0>,
  ok: #PID<0.296.0>
]
sending message  to 1
sending message  to 2
sending message  to 4
sending message  to 5

If you pass a list of phone numbers to send_to_all/1 and one of them is 3, the corresponding task will raise an arithmetic error, but other tasks will continue to run. From the results, we can see that we are notified of an arithmetic error in one of the started processes (#PID<0.294.0>). We also receive printed messages indicating that the messages were sent successfully to numbers 1, 2, 4, and 5. That means that the task started for number 3 encountered an error, but didn't prevent other processes from performing their tasks.

This approach enhances concurrency by allowing tasks to run concurrently, while also ensuring that they are isolated processes. This helps to create a more robust system.

A Problem: Working with a Series of Tasks

We've seen how we have been able to send messages to the given phone numbers concurrently using Task.async/1 and Task.start/1. In the examples provided, send_to_all/1 spawned a process for each task when sending a message. Even though our aim is to leverage concurrency and build a fault-tolerant and robust application, we should also take into consideration that processes don't share memory.

The more we increase phone numbers, the more we start a high number of concurrent tasks, each one occupying their own memory. Therefore, increasing the number of concurrent tasks can lead to an application consuming a significant amount of memory. That, in turn, can cause slow application performance and a spike in system usage, degrading system performance and maybe even making other services unresponsive.

Let's make send_to_all/1 send messages to 1 million phone numbers. But first, we'll modify it by renaming send_to_all/1 to differentiate the send_to_all/1 using Task.async/1 from the one using Task.start/1.

defmodule Sender do
  def send_message(phone_number) do
    # Simulate some processing time
    Process.sleep(3000)
    IO.puts("Sending message to #{phone_number}")
    {:ok, "sent"}
  end

  # Asynchronous version using Task.start/1
  def send_to_all_start(phone_numbers) do
    Enum.each(phone_numbers, fn phone_number ->
      Task.start(fn -> send_message(phone_number) end)
    end)
  end

  # Asynchronous version using Task.async/1
  def send_to_all_async(phone_numbers) do
    phone_numbers
    |> Enum.map(fn phone_number ->
      Task.async(fn -> send_message(phone_number) end)
    end)
  end
end

In this modification, send_to_all_async/1 and send_to_all_start/1 functions have been defined to differentiate the behavior of Task.async/1 and Task.start/1 when attempting to run a million processes.

Now, open IEx (or recompile it using the recompile() command if it was already running) and call send_to_all_start/1 with a million phone numbers. We're not using a real phone number, therefore we simply pass a 1..1_000_000 range of numbers.

iex > Sender.send_to_all_start(1..1_000_000)

06:12:14.067 [error] Too many processes
** (SystemLimitError) a system limit has been reached
    :erlang.spawn(Task.Supervised, :noreply, [{:nonode@nohost, #PID<0.150.0>, #PID<0.150.0>}, [#PID<0.150.0>, #PID<0.142.0>], [#PID<0.150.0>], {:erlang, :apply, [#Function<7.61290973/0 in Sender.send_to_all_start/1>, []]}])
    (elixir 1.15.5) lib/task/supervised.ex:6: Task.Supervised.start/3
    (elixir 1.15.5) lib/enum.ex:4356: Enum.map_range/4
    (elixir 1.15.5) lib/enum.ex:4356: Enum.map_range/4
    (elixir 1.15.5) lib/enum.ex:4356: Enum.map/2
    iex:2: (file)

The first thing we can see is an error message notifying us that too many processes have been started, exceeding the system limit. This is happening as soon as Task.start/1 is invoked for each phone number in the range from 1 to 1,000,000. Success messages have been printed, but they are not showing for all of the phone numbers. The highest number of messages that I can see in the output is 262073. Since the numbers are not sorted, it is easy to miss the highest number. However, the total number of messages sent is around 262000.

What's Going On Here and How Do We Fix It?

We are having issues with starting a million processes because the number of tasks spawned by Task.start/1 exceeds the default limit of processes (262144, for performance and memory-saving reasons) that can be started in BEAM. However, we can override this limit using +P NUM. This means we can increase the limit by running the command iex --erl '+P 1000000'.

To apply this change, close IEx and restart it by running the command iex --erl '+P 1000000' -S mix. Then, you can call send_to_all_start/1 again, with a range of phone numbers from 1 to 1000000.

sender tracey$ iex --erl '+P 1000000' -S mix
iex > Sender.send_to_all_start(1..1_000_000)
 :ok

This time, the function Enum.each/2 has returned the final output :ok. Additionally, success messages have been printed for all of the one million phone numbers. It is good that we can override the default system limit, but we must be cautious when increasing the limit, as it can cause the VM to use more memory, potentially leading to performance issues.

To monitor memory usage before and after a system limit increase, use the :observer.start command in IEx to open the Observer tool. The Observer shows a spike in memory usage when the system limit is increased.

Before the system limit increase:

After the system limit increase:

Sending Messages To the Phone Numbers with `Task.async/1`

Next, we will use the send_to_all_async/1 function to send messages to a million phone numbers. This function uses Task.async/1 to start a process for each task.

iex > Sender.send_to_all_async(1..1_000_000)

With Task.async/1, you can send a message to all the phone numbers. However, this capability may not be available on all machines, so keep that in mind.

The Observer shows a spike in memory usage when using Task.async/1.

Running these tasks synchronously works out better in terms of memory usage compared to running them asynchronously when using a combination of Enum with either Task.async/1 or Task.start/1.

When using Task.start/ or Task.async/1 to start a process for each task, you're essentially creating a separate process for each message you want to send. While this can be an efficient way to handle concurrency, it can also lead to increased memory usage, especially when dealing with a large number of tasks simultaneously.

The spike in memory usage we're seeing in the Observer is likely due to the fact that each process created by Task.async/1 and Task.start/1 consumes memory. With a million phone numbers, you're creating a million processes, which can quickly exhaust available memory, especially if each process is doing significant work or holding onto a large amount of data.

The Solution: Use `Task.async_stream/3`

When utilizing Task processes, the goal is to achieve concurrency to efficiently send messages to designated phone numbers. However, it is important to be mindful of system resources and avoid sudden increases in pressure.

Fortunately, the Task module provides a useful feature known as Task.async_stream/3. It works similarly to Enum.map/2 and Task.async/2 combined, as it creates task processes from a given list of items.

With Task.async_stream/3, you can perform the task for each item in the list concurrently, by starting a process. The only difference is that you can set a limit on the number of processes running at the same time with this function.

For instance, suppose the message application needs to send messages to 100 phone numbers using Task.async_stream/3. In this case, we can set the concurrency limit to 5, which means that, at most, only 5 processes will start to send messages to the given phone numbers at the same time.

Task.async_stream/3 takes in three arguments:

Enumerable: This argument represents the collection of items that you want to process concurrently. It can be an Enum or a Stream.
Anonymous function: This must take a single argument that represents an element of the enumerable. The function is applied to each enumerable element. It defines the task to be executed concurrently for each item.
Options, used to control the level of concurrency, the time tasks are allowed to run, ordering of the results and the action to take when a task times out.

Task.async_stream/3 returns a stream, which is a lazy enumerable. Therefore, transformations or computations on the stream are not performed as soon as the stream is created. Instead, they are performed until the stream is explicitly consumed or operated upon.

An Example Using `Task.async_stream/3`

Here's an example of using Task.async_stream/3 in IEx:

iex > Task.async_stream(1..5, fn phone_number -> send_message(phone_number) end)
#Function<3.112246596/2 in Task.build_stream/3>

The result that's returned is a stream. To run this stream, we can use Stream.run/1, which returns :ok and isn't useful when we're not interested in the final result. Another alternative is to use the Enum functions. This is useful when you intend to perform other tasks with the expected result.

iex > 1..5 |> Task.async_stream(fn phone_number -> send_message(phone_number) end) |> Stream.run()
sending message  to 1
sending message  to 2
sending message  to 3
sending message  to 4
sending message  to 5
:ok

Stream.run/1 is used to run the stream without collecting the result. The messages are sent to the given phone numbers and once all tasks are completed, :ok is returned.

iex > 1..5 |> Task.async_stream(fn phone_number -> send_message(phone_number) end) |> Enum.to_list
sending message  to 1
sending message  to 2
sending message  to 3
sending message  to 4
sending message  to 5

[
:ok, {:ok, "sent"},
:ok, {:ok, "sent"},
:ok, {:ok, "sent"},
:ok, {:ok, "sent"},
:ok, {:ok, "sent"}
]

Enum.to_list/1 runs the stream and collects the results into a list.

Task.async_stream/3 accepts a list of options. When not explicitly passed, the options default to their default values.
Let's dive further into the options passed to Task.async_stream/3.

`:max_concurrency`

:max_concurrency is responsible for setting a limit on the number of processes running at the same time. Its value defaults to the number of logical cores available in a system.

When running Task.async_stream/3 in IEx with a 1..20 range of phone numbers:

iex >  Task.async_stream(1..20, fn phone_number -> send_message(phone_number) end) |> Stream.run()

I noticed that the number of items processed at the same time was in a batch of 10 until completion. This number can differ depending on the machine used, so don't be surprised if you're seeing a batch of 20 or even 5 items processed simultaneously in your machine. This is because of the logical cores available in different machines. To confirm the number of logical cores available, use System.schedulers_online/0. For machines whose CPU has less than 4 logical cores, Task.async_stream/3 will appear to be slower.

We can emulate the slow performance by setting :max_concurrency to a value less than 4.

iex >  Task.async_stream(1..20, fn phone_number -> send_message(phone_number) end, max_concurrency: 2) |> Stream.run()

When the concurrency limit is set to 2, it takes more time to complete compared to using the default value set, 10. This is due to the fact that with a lower concurrency limit, fewer tasks are processed concurrently.

When setting max_concurrency, consider your system resources. Verify that the system has enough resources to handle the specified concurrency limit effectively and also bear in mind the need for performance. Depending on what works for your use case, you can decrease or increase the concurrency limit.

Our application aims to send messages to a million phone numbers. At present, the max_concurrency setting on my machine is set to 10 and is working satisfactorily. However, it is taking a long time to process all the messages. After 17 minutes, only about 3,000 messages have been sent. To improve the processing speed, we need to increase the concurrency limit. I shall set it to 100, allowing more items to be processed concurrently, while also ensuring we do not overload the system.

iex > 1..20
    > |> Task.async_stream( fn phone_number -> send_message(phone_number) end, max_concurrency: 100)
    > |> Stream.run()

After increasing the max_concurrency to 100:

In 4 minutes, about 6,000 messages were sent.
There is no spike in memory allocation when we open the observer.

Here's the observer memory allocation when max_concurrency is set to 10:

And here it is when max_concurrency is set to 100:

`:ordered`

When we execute this code in IEx, you will notice that the results are returned in the same order as the input data:

iex > 1..5
    > |> Task.async_stream( fn phone_number -> send_message(phone_number) end, max_concurrency: 100)
    > |> Enum.to_list

sending message  to 1
sending message  to 2
sending message  to 3
sending message  to 4
sending message  to 5

[
:ok, {:ok, "sent"},
:ok, {:ok, "sent"},
:ok, {:ok, "sent"},
:ok, {:ok, "sent"},
:ok, {:ok, "sent"}
]

This is because we have the :ordered option set to true by default in the Task.async_stream/3 function.

However, setting :ordered to true can result in slower processing if a task takes longer to complete. This is because Task.async_stream/3 will wait for a slow task to finish before moving on to the next one. Therefore, it is important to consider the trade-off between ordered results and processing speed when using this function.

To improve processing speed, you can disable ordering by setting the :ordered option to false. This way, Task.async_stream/3 won't wait for a slow task to complete processing before moving on to the next task.

Let's introduce a slow process in our Sender application.

defmodule Sender do
  def send_message(3) do
    # Simulate slow processing time
    Process.sleep(4000)
    IO.puts("Sending message to 3")

    {:ok, "sent", 3}
  end

  def send_message(phone_number) do
    # Simulate some processing time
    Process.sleep(3000)
    IO.puts("Sending message to #{phone_number}")
    {:ok, "sent", phone_number}
  end

  # Asynchronous version using Task.async_stream/3
  def send_to_all(phone_numbers) do
    phone_numbers
    |> Task.async_stream(fn phone_number -> send_message(phone_number) end)
    |> Enum.to_list()
  end
end

The send_message/1 function accepts one argument, a phone number. If the phone number is 3, the function simulates a slow process by sleeping for 4 seconds. Otherwise, it simulates a regular process by sleeping for 3 seconds. The output result includes the phone number that's being processed, which allows us to check the order of the returned result.

iex > Sender.send_to_all(1..5)

sending message  to 1
sending message  to 2
sending message  to 4
sending message  to 5
sending message  to 3

[
{:ok, {:ok, "sent"}, 1},
{:ok, {:ok, "sent"}, 2},
{:ok, {:ok, "sent"}, 3},
{:ok, {:ok, "sent"}, 4},
{:ok, {:ok, "sent"}, 5}
]

The print message of number 3 is displayed last, indicating that it took longer to process before sending. However, the collected list is returned in the same order as the given input. This means that, while number 3 took longer to complete, numbers 4 and 5 had to wait before being operated on.

In our case, we want to check whether the status of the messages sent to the given phone numbers was successful or not. To speed up things, let's disable the :ordered option by setting it to false, since we don't care about the order.

defmodule Sender do
  def send_message(3) do
    # Simulate slow processing time
    Process.sleep(4000)
    IO.puts("Sending message to 3")

    {:ok, "sent", 3}
  end

  def send_message(phone_number) do
    # Simulate some processing time
    Process.sleep(3000)
    IO.puts("Sending message to #{phone_number}")
    {:ok, "sent", phone_number}
  end

  # Asynchronous version using Task.async_stream/3
  def send_to_all(phone_numbers) do
    phone_numbers
    |> Task.async_stream(fn phone_number -> send_message(phone_number) end, ordered: false)
    |> Enum.to_list()
  end
end

iex > Sender.send_to_all(1..5)

sending message  to 1
sending message  to 2
sending message  to 4
sending message  to 5
sending message  to 3

[
{:ok, {:ok, "sent"}, 1},
{:ok, {:ok, "sent"}, 2},
{:ok, {:ok, "sent"}, 4},
{:ok, {:ok, "sent"}, 5},
{:ok, {:ok, "sent"}, 3},
]

The number 3 is collected last in the returned result. This way, we can be sure that Task.async_stream/3 won't be idle waiting for the processing of number 3 to complete, before moving to the next.

`:timeout`

The Task.async_stream/3 function has a :timeout option which sets a limit on how long each task can run. By default, the timeout is set to 5000 milliseconds or 5 seconds. You can specify the time limit in milliseconds or set it to :infinity if you want to allow the task to run indefinitely. If a task takes longer than the specified timeout, it will raise an exception and terminate the current process.

defmodule Sender do
  def send_message(3) do
    # Simulate slow processing time
    Process.sleep(7000)
    IO.puts("Sending message to 3")

    {:ok, "sent", 3}
  end

  def send_message(phone_number) do
    # Simulate some processing time
    Process.sleep(3000)
    IO.puts("Sending message to #{phone_number}")
    {:ok, "sent", phone_number}
  end

  # Asynchronous version using Task.async_stream/3
  def send_to_all(phone_numbers) do
    phone_numbers
    |> Task.async_stream(fn phone_number -> send_message(phone_number) end, ordered: false)
    |> Enum.to_list()
  end
end

Here, the processing time of the slow process/task is increased to 7000ms, which exceeds the default timeout in Task.async_stream/3.

iex > Sender.send_to_all(1..5)

sending message  to 1
sending message  to 2
sending message  to 4
sending message  to 5

** (exit) exited in: Task.Supervised.stream(5000)
    ** (EXIT) time out
    (elixir 1.15.5) lib/task/supervised.ex:314: Task.Supervised.stream_reduce/7
    (elixir 1.15.5) lib/enum.ex:4387: Enum.reverse/1
    (elixir 1.15.5) lib/enum.ex:3704: Enum.to_list/1
    iex:3: (file)

When we execute the Sender.send_to_all(1..5) function in IEx, an exception is raised, which stops the stream and crashes the current process. This is because the slow task takes more than 5000ms to complete, while Task.async_stream/3 has a timeout limit of 5000ms.

`:on_timeout`

It's difficult to predict the duration of a task, as several factors can contribute to its slow processing. For instance, processing a large amount of data can be time-consuming, as can waiting for a slow third-party API to respond.

Instead of allowing Task.async_stream/3 to raise an exception that stops the stream, we can use the :on_timeout option by setting it to :kill_task. This option determines the action to take when the task times out. Setting it to :kill_task will cause Task.async_stream/3 to ignore the process that exits with a timeout and continue with other tasks.

By default, it's set to :exit. This results in the task exceeding the timeout to stop the stream and crash the current process, as seen with the :timeout option.

defmodule Sender do
  def send_message(3) do
    # Simulate slow processing time
    Process.sleep(7000)
    IO.puts("Sending message to #{3}")
    {:ok, "sent", 3}
  end

  def send_message(phone_number) do
    # Simulate some processing time
    Process.sleep(3000)
    IO.puts("Sending message to #{phone_number}")
    {:ok, "sent", phone_number}
  end

  # Asynchronous version using Task.async_stream/3
  def send_to_all(phone_numbers) do
    phone_numbers
    |> Task.async_stream(fn phone_number -> send_message(phone_number) end,
      ordered: false,
      on_timeout: :kill_task
    )
    |> Enum.to_list()
  end
end

Add the :on_timeout option and set it to :kill_task in Task.async_stream/3. Then, execute Sender.send_to_all(1..5) in IEx. We can observe that the other tasks are processed completely, while the slow process is ignored, and returns {:exit, :timeout} instead of crashing.

iex > Sender.send_to_all(1..5)

sending message  to 1
sending message  to 2
sending message  to 4
sending message  to 5

[
{:ok, {:ok, "sent"}, 1},
{:ok, {:ok, "sent"}, 2},
{:ok, {:ok, "sent"}, 4},
{:ok, {:ok, "sent"}, 5},
exit: :timeout
]

Before we wrap up, let's take a very quick look at monitoring memory using Observer.

Monitor Memory with Erlang Observer

Let's use Erlang Observer to monitor memory allocation. We can compare memory usage when defining the send_to_all/1 function using Task.async_stream/3 and using a combination of Enum.map/2 and Task.async/1.

defmodule Sender do
  def send_message(3) do
    # Simulate slow processing time
    Process.sleep(7000)
    IO.puts("Sending message to #{3}")
    {:ok, "sent", 3}
  end

  def send_message(phone_number) do
    # Simulate some processing time
    Process.sleep(3000)
    IO.puts("Sending message to #{phone_number}")
    {:ok, "sent", phone_number}
  end

  # Asynchronous version using Task.async_stream/3
  def send_to_all(phone_numbers) do
    phone_numbers
    |> Task.async_stream(fn phone_number -> send_message(phone_number) end,
      max_concurrency: 100,
      ordered: false,
      on_timeout: :kill_task
    )
    |> Stream.run()
  end
end

defmodule Sender do
  def send_message(3) do
    # Simulate slow processing time
    Process.sleep(7000)
    IO.puts("Sending message to #{3}")
    {:ok, "sent", 3}
  end

  def send_message(phone_number) do
    # Simulate some processing time
    Process.sleep(3000)
    IO.puts("Sending message to #{phone_number}")
    {:ok, "sent"}
  end

  # Asynchronous version using Task.async/1
  def send_to_all(phone_numbers) do
    phone_numbers
    |> Enum.map(fn phone_number ->
      Task.async(fn -> send_message(phone_number) end)
    end)
  end
end

The combination of Task.async/1 and Enum.map/2 can significantly impact memory consumption compared to using Task.async_stream/3.

Task.async_stream/3 is a powerful tool for effectively utilizing concurrency without overloading system resources when working with a series of tasks.

Here are a couple of benefits of using Task.async_stream/3:

It prevents sudden spikes in system usage: By setting the concurrency limit option, you can control the number of concurrent tasks being executed at any given time. This helps prevent the system from becoming overwhelmed with too many tasks running simultaneously, which could lead to spikes in resource usage, such as memory or CPU.
It handles back pressure: Back pressure occurs when there's resistance to the progress of turning input to output. By setting the concurrency limit, Task.async_stream/3 ensures that tasks are processed at a manageable rate, preventing resource exhaustion.

And that's it!

Wrapping Up

Elixir has several tools that are useful when it comes to leveraging concurrency, and Task processes is one of them.

In this post, we explored the concept of concurrency and some of the issues we might face when dealing with a series of tasks while using concurrency.

We then explored the use of Task.async_stream/3, a Task module function that is effective in handling a large number of tasks, as it provides both concurrency and performance.

Happy coding!

P.S. If you'd like to read Elixir Alchemy posts as soon as they get off the press, subscribe to our Elixir Alchemy newsletter and never miss a single post!

How to Build a Memory-efficient Elixir App with Streams

Tracey Onim — Tue, 20 Feb 2024 15:51:27 +0000

We have all encountered collections of data at some point when working on Elixir applications. These collections are very handy for storing, retrieving, and manipulating data using different data structures, making them very efficient in managing clean code.

In this article, we'll go through the following:

What are collections in Elixir?
The problems likely to be faced when working with collections
The greedy approach when working with large datasets
The enumerable in Elixir, used to work around collections

Finally, we'll explore how to build a memory-efficient Elixir application using the lazy processing approach with streams.

What are Collections in Elixir?

You can view a collection as a container that groups multiple things/objects/elements into a single unit. This collection allows you to store, retrieve, and manipulate data.

A couple of examples of collections encountered in the real world include:

A mail folder that contains received and sent letters.
A phone directory mapping phone numbers to corresponding names.

In Elixir, List, Maps, Ranges, files, and tuples are the most common types of collections. Some of these collections can be iterated through. Each element is passed through once, in order, so we can term them enumerables. Things that can be iterated in Elixir implement the Enumerable protocol. List, Maps, and Ranges are the most common datatypes used as enumerables.

Elixir has two modules with iteration functions that use the Enumerable protocol: Enum and Stream.

The Challenges of Collections

Working with collections can be very efficient, especially when your collections aren't that large. However, manipulating these collections can be very difficult when dealing with large datasets.

For example, let's say we want to retrieve organization data from a CSV file and display it in our application. Ideally, we have to load all the organizations from the CSV file into an application's memory and transform the retrieved data into the data we want.

Our application can be safe if the memory contains a predictable number of organizations that can be easily transformed or manipulated. However, if we have extensive data on organizations retrieved from the CSV, managing this data can result in slow application performance, because more memory is needed to consume it.

So, how can we elegantly manage our large CSV file? (keeping in mind some of the problems we might face while processing it).

We can process our CSV file using one of two approaches:

The greedy approach that consumes more memory.
The memory-efficient lazy processing approach.

Earlier, we mentioned that Elixir has two modules, Enum and Stream, with iteration functions. We'll use these modules to process our CSV file.

The Greedy Approach For Our Elixir Application

In the greedy approach, we are going to use the Enum module because:

When given a collection, it will consume all the content of that collection. That means the whole collection should be loaded into memory before being processed.
The returned result is another collection that will be loaded into memory, which might be inefficient if data is not needed at that time.

Let's process our CSV file greedily.

We shall use an organization's CSV file with 2,000,000 records downloaded from here for demonstration purposes. The size of the CSV file (after unzipping) is around 283.4MB.

Our focus will be to get the records of all organizations and transform them into data we can easily display in the browser. The result should look like this:

[
%{
index: "1",
organization_id: "1234",
name: "PLC",
website: "https://park.com/",
country: "Kenya",
description: "Focused for the best",
founded: "2016",
industry: "sports",
num_of_employees: 300

},..
]

Before we begin, let's open the interactive shell to manipulate the CSV file. After that, start the Erlang Observer application by running:

iex > :observer.start

We shall use the Observer application to inspect memory allocators when processing the CSV file.

Step 1: Loading the CSV File Into Memory

Use File.read!("organization.csv") to load the whole CSV file into memory. The observer shows an extra 283.84 MB allocated to memory.

iex > org_csv = File.read!("organization.csv")


    "Index,Organization Id,Name,Website,Country,Description,Founded,Industry,Number of employees\r\n1,391dAA77fea9EC1,Daniel-Mcmahon,https://stuart-rios.biz/,Cambodia,Focused eco-centric help-desk,2013,Sports,1878\r\n2,9FcCA4A23e6BcfA, Mcdowell,http://jacobs.biz/,Guyana,Front-line real-time portal,2018,Legal Services,9743\r\n3,DB23330238B7B3D,\"Roberts, Carson and Trujillo\",http://www.park.com/,Jordan,Innovative hybrid data-warehouse,1992,Hospitality,7537\r\n4,bbf18835CFbEee7,\"Poole, Jefferson and Merritt\",http://hayden.com/,Cocos (Keeling) Islands,Extended regional Graphic Interface,1991,Food Production,9974\r\n5,74ECD725ceaDfd9,\"Ritter, Patel and Cisneros\",https://www.mason-blackwell.info/,Ecuador,Re-contextualized actuating website,2019,Computer Networking,5050\r\n6,ea42e2FECDAdF0c,Stafford Ltd,http://www.fuller.biz/,Qatar,Multi-channeled optimizing customer loyalty,1979,Aviation / Aerospace,7506\r\n7,b8EE5AE2Df8BA46,Roach Ltd,https://www.oconnell.com/,Guatemala,Switchable explicit complexity,2020,Hospitality,109\r\n8,21829Cf5a968024,Gill PLC,https://www.berry.com/,Lesotho,Future-proofed systemic pricing structure,1983,Printing,1822\r\n9,f9C605c034f47cD,Summers-Jordan,https://le.net/,Luxembourg,Cross-platform bottom-line system engine,1976,Primary / Secondary Education,1603\r\n10,0fBa55ecDA6cb4B,\"Richard, Lane and Weaver\",https://bauer-hardy.com/,United States Minor Outlying Islands,Optimized system-worthy complexity,2006,Building Materials,3505\r\n11,d085befF19Be6fB,\"Harrington, Sutton and Wilkins\",http://henson.com/,Guinea-Bissau,Diverse scalable instruction set,1985,Airlines / Aviation,1926\r\n12,13B77fAF602E949,Evans LLC,http://harrington-powers.com/,Cook Islands,Re-contextualized stable flexibility,2017,Design,3338\r\n13,90234C8d0D7eB75,\"Patterson, Deleon and Donovan\",https://thornton.net/,Holy See (Vatican City State),Versatile encompassing migration,1970,Automotive,9312\r\n14,155A7db5D8Cce47,\"Carlson, Snyder and Holland\",http://foley.com/,Korea,Devolved optimal secured line,1979,International Trade / Development,2649\r\n15,77B7F7fb1Ac2A44,Duffy-Stark,https://www.morales.com/,Monaco,Re-contextualized 24/7 Graphical User Interface,1990,Aviation / Aerospace,1714\r\n16,1f6bABBA98cd33E,\"Frederick, Fry and Poole\",http://www.madden.org/,Vanuatu,Synchronized empowering structure,2008,Fishery,1862\r\n17,05dbf87Ee09b6BC,\"Walton, Proctor and Peters\",http://www.casey-bell.com/,Grenada,Diverse local collaboration,1979,Alternative Medicine,4678\r\n18,20ef50A2fdB3993,Oneal and Sons,http://munoz.org/,Fiji,Operative uniform definition,1987,Religious Institutions,8332\r\n19,a9C703D1A2B074C,Maynard-Bush,http://leblanc.com/,Senegal,Customizable zero tolerance functionalities,2011,Apparel / Fashion,1668\r\n20,1ea33Ac2face255,\"Rosario, Martin and White\",https://www.mueller-rhodes.net/,Isle of Man,Synchronized 6thgeneration flexibility,1994,Music,5134\r\n21,CDCAe4a6DA0eC6d,Brady Ltd,http://montgomery.biz/,Guyana,Self-enabling modular archive,1989,Professional Training,5153\r\n22,f41b9Ce917bDD28,Odom Group,http://www.riley-dalton.net/,Saint Kitts and Nevis,Grass-roots bandwidth-monitored projection,2003,Computer Hardware,4793\r\n23,B1181c8C3d43216,Stanton-Arroyo,http://baird.org/,Montserrat,Integrated empowering core,1976,Mental Health Care,6919\r\n24,b3E9C4E4cbC4a8e,Stafford-Hickman,https://love.net/,United States Virgin Islands,Virtual background application,1996,Hospital / Health Care,3855\r\n25,f92AacaEEcB6eC6,Ryan and Sons,http://koch-raymond.org/,Chad,Public-key maximized definition,2017,Airlines / Aviation,1564\r\n26,8eb91E6eeDBC2A5,Wolf Group,https://aguilar-solomon.net/,Cayman Islands,Public-key value-added alliance,1984,Package / Freight Delivery,4509\r\n27,fDaeD26dB6fd79a,\"Horne, Bailey and Oconnor\",https://cruz.com/,Western Sahara,Customer-focused needs-based service-desk,1973,Wholesale,2670\r\n28,19Bbb8dB90dFaFF,Cobb Inc,http://miles.com/,Netherlands,Assimilated local ability,2008,Political Organization,2494\r\n29,00AF7EacF3fEb91,Parrish-Peterson,https://strong.org/,France,Multi-lateral encompassing structure,2004,Political Organization,9059\r\n30,8Ec27fFD6b945A9,\"Barr" <> ...

Step 2: Create a List for Each Row In the CSV

The data in Step 1 is loaded text, which means we are still far from our end goal. In this step, let's take the loaded text data and split it into lines. This will result in a list of string elements, with each element representing a row of the CSV.

iex > [_header | rows] = org_csv |> String.split("\n")

     ["Index,Organization Id,Name,Website,Country,Description,Founded,Industry,Number of employees\r",
 "1,391dAA77fea9EC1,Daniel-Mcmahon,https://stuart-rios.biz/,Cambodia,Focused eco-centric help-desk,2013,Sports,1878\r",
 "2,9FcCA4A23e6BcfA,Mcdowell, http://jacobs.biz/,Guyana,Front-line real-time portal,2018,Legal Services,9743\r",
...]

The observer shows an increase in allocated memory:

Step 3: Create a Column from Each Row

In the second step, we pattern-matched our result to [_header | rows]. We are not interested in the header of the CSV file, but in the body.

In this step, we will split each row into a list, with the elements in each list representing a column of the CSV file.

iex > org_records = rows |> Enum.map(&String.split(&1, ","))

  [
    ["1", "391dAA77fea9EC1", "Daniel-Mcmahon", "https://stuart-rios.biz/",
     "Cambodia", "Focused eco-centric help-desk", "2013", "Sports", "1878\r"],
    ["2", "9FcCA4A23e6BcfA", "Mcdowell",
     "http://jacobs.biz/", "Guyana", "Front-line real-time portal", "2018",
     "Legal Services", "9743\r"],
...]

This step alone takes almost 60 seconds to process. The observer also shows a spike increase in memory allocation:

Step 4: Create a Keyword List of Each Row

Here, each column in each row will be zipped to its corresponding header name.

iex > org_records = org_records |> Enum.map(fn record ->  Enum.zip([:org_id, :name, :website, :country, :description, :founded, :industry, :number_of_employees], record) end)
    [
      [
       index: "1",
       org_id: "391dAA77fea9EC1",
       name: "Daniel-Mcmahon",
       website: "https://stuart-rios.biz/",
       country: "Cambodia",
       description: "Focused eco-centric help-desk",
       founded: "2013",
       industry: "Sports",
       number_of_employees: "1878"
      ],
      [
        index: "2",
        org_id: "9FcCA4A23e6BcfA",
        name: "Mcdowell",
        website: "http://jacobs.biz/",
        country: "Guyana",
        description: "Front-line real-time portal",
        founded: "2018",
        industry: "Legal Services",
        number_of_employees: "9743"
      ],
...]

In this case, we use Enum.map/2 and Enum.zip/2 to process the organization record results from step 3 (a list of lists, where each list element represents a row of records). Remember that each call to the Enum module takes a collection and returns a collection. Both Enum.map/2 and Enum.zip/2 will greedily process our data, which explains the spike increase in memory allocation as seen from the observer.

Enum.map/2 takes in a list of lists, where each list element is a row of organization records. It iterates through each corresponding row and invokes Enum.zip/2 on each row to transform it into a keyword list.

The Enum.zip/2 function takes in two enumerables. The first enumerable is a list of header names, and the second is a list of each row. The zip function zips the corresponding elements from the list of headers and the list of each row into a list of tuples. The tuple element in the list has an atom as the first element (which is the header name and value of the corresponding header name).

Elixir represents this kind of output as a keyword list. In your interactive shell, you will notice each row is transformed into a keyword list, but not a list of tuples.

Enum.map/2 returns a new list of keyword lists.

Step 5: Convert to Map

Remember, our end goal was to work with data that's easy to manipulate when we want to display a list of organizations. In this final step, we will convert the keyword list above into maps.

iex > org_records |> Enum.map(fn data -> Map.new(data) end)

    [
      %{
        index: "1",
        org_id: "391dAA77fea9EC1",
        name: "Daniel-Mcmahon",
        website: "https://stuart-rios.biz/",
        country: "Cambodia",
        description: "Focused eco-centric help-desk",
        founded: "2013",
        industry: "Sports",
        number_of_employees: "1878"
      },
      %{
        index: "2",
        org_id: "9FcCA4A23e6BcfA",
        name: "Mcdowell",
        website: "http://jacobs.biz/",
        country: "Guyana",
        description: "Front-line real-time portal",
        founded: "2018",
        industry: "Legal Services",
        number_of_employees: "9743"
      },
...]

Compiling everything together:

iex > [_| rows ] =
  "organization.csv"
 |>  File.read!()
 |>  String.split("\n")
 |> Enum.map(&String.split(&1, ","))


iex >
   rows
|> Enum.map(fn record ->  Enum.zip([:org_id, :name, :website, :country, :description, :founded, :industry, :number_of_employees], record) end)
|> Enum.map(fn record -> Map.new(record) end)

Observation with `Enum`

In each Enum step, we generated an output before the final result. The intermediate result in each step is stored in memory as a full collection. That's because, with Enum implementation, the whole collection must be available for the next step in the pipeline to start processing.
With Enum, each step has to wait for its intermediate result to start processing, so it takes more time to process data in each step.
There is a spike increase in memory allocation in each step, meaning an intermediate output must be kept in memory.

The greedy approach, or working with Enum, works well with a predictable number of elements. We couldn't have easily noticed these problems in a small CSV file.

However, this approach is very inefficient when we have to keep an intermediate result in memory in each step. This is a massive waste of memory that can lead to slow application performance.

Solution: Build a Memory-efficient Elixir App with Streams

First, what are our goals?

Memory allocation that's only for the final result.
To process records only when we need them.

Using streams, we can do these things. Streams are known to be lazy enumerables.

This is because streams:

Lazily load data into memory: That is, one element at a time instead of loading everything at once.
Only load data when needed.
Instead of applying a transformation to the given enumerable, return a value with the intended specifications.
Process one piece of data at a time as it arrives instead of waiting for the whole collection to be available.
Are a composable enumerator that reads and pipes one single line at a time without needing to wait for all the passed data to be processed at every single step.

The Lazy Processing Approach with Streams in Elixir

We can leverage lazy processing to process our organization's CSV file.

The good thing is that more Elixir modules now support streams. Instead of opening the CSV file and loading the data on step 1, let's use File.stream!/3 to create a stream without necessarily opening the file.

iex > File.stream!("/org.csv")
    %File.Stream{
      line_or_bytes: :line,
      modes: [:raw, :read_ahead, :binary],
      path: "/org.csv",
      raw: true
    }

The File.stream!/3 function returns %File.Stream{} with intended specifications.

The observer doesn't reflect our CSV file's extra memory size, meaning the file still needs to be opened and loaded into memory.

Streams are composable enumerables. We can pipe one stream to another. We will replace the Enum functions from steps 2-4 of the greedy approach with stream functions. We'll avoid processing data and loading it into memory until we decide to run the stream (compute/process the data as intended) by passing it to a function in the Enum module.

Let's compose our first part of the pipeline with streams:

iex > header = [:index, :org_id, :name, :website, :country, :description, :founded, :industry, :number_of_employees]
iex >
"/org.csv"
|> File.Stream!()
|> Stream.drop(1)
|> Stream.map(&String.split(&1, ["\n", ","], trim: true))
|> Stream.map(&Stream.zip(header, &1))
|> Stream.map(&(Map.new(&1)))

  #Stream<[
  enum: %File.Stream{
    path:  "/org.csv",
    modes: [:raw, :read_ahead, :binary],
    line_or_bytes: :line,
    raw: true,
    node: :nonode@nohost
  },
  funs: [#Function<34.53678557/1 in Stream.drop/2>,
   #Function<48.53678557/1 in Stream.map/2>,
   #Function<48.53678557/1 in Stream.map/2>,
   #Function<48.53678557/1 in Stream.map/2>]
]>

The composed pipeline of streams above returns a stream that only stores the intended computation instead of applying the transformation.

Here is what is happening within our stream processing pipeline:

Instead of opening the file, we use File.Stream/3 to create a stream.
Stream.drop/2 lazily drops the first element, the header.
Stream.map/2:
- splits data into columns
- lazily zips each row into its corresponding header name with Stream.zip/2
- transforms each piece of data into a map

To process our desired result, we have to pass the stream value to a function in the Enum module.
First, let's take only the first 3 elements from our stream and observe what happens.

iex > csv
     |> File.stream!()
     |> Stream.drop(1)
     |> Stream.map(&String.split(&1, ["\n", ","], trim: true))
     |> Stream.map(&Stream.zip(header, &1))
     |> Stream.map(&(Map.new(&1)|> IO.inspect(label: "=============")))
     |> Enum.take(3)

Enum.take/2 receives the transformed map data from the stream and only gets the first 3 elements. Once it gets the 3 elements from the stream, there is no more processing.

Running this pipeline, you will notice that:

The result is returned instantaneously.
IO.inspect only prints the first 3 elements. This is proof that the elements are being processed one at a time by subsequent calls to Enum functions.

Lazy Approach Using Streams Vs. Greedy Approach

Let's use the greedy approach to take the first 3 elements from our code:

iex >
[_| rows ] =
  csv
 |>  File.read!()
 |>  String.split("\n")
 |> Enum.map(&String.split(&1, ","))


iex >
   rows
|> Enum.map(fn record ->  Enum.zip([:org_id, :name, :website, :country, :description, :founded, :industry, :number_of_employees], record) end)
|> Enum.map(fn record -> Map.new(record) |> IO.inspect(label: "+++++++++++++++++") end)
|> Enum.take(3)

IO.inspect prints the whole 2,000,000 records. This shows that with Enum, we have to wait for the whole collection to be available upfront before starting the next processing batch. Using this approach, an intermediate list of 2,000,000 records has to be kept in memory at each step (even if, in the end, we only get the first 3 records using Enum.take/2).

Unlike the greedy approach, using the lazy approach, there is no need to process the whole collection for the next process to take place. By passing Enum.take/2 to fetch the first 3 elements, we can see there isn't even a peak in memory.

With streams, even if we process 2,000,000 records, we can decide the amount of records to load in memory and ensure that the amount of utilized memory stores the final result. This is not the same case with Enum, since in each step there is a certain amount of records kept in memory.

We can also compare the greedy and lazy approaches by monitoring memory allocators using the Erlang observer.

Remember, in the greedy approach, we processed all records. We will do the same with our lazy approach by passing the stream pipeline to Enum.to_list/1.

iex > csv
     |> File.stream!()
     |> Stream.drop(1)
     |> Stream.map(&String.split(&1, ["\n", ","], trim: true))
     |> Stream.map(&Stream.zip(header, &1))
     |> Stream.map(&(Map.new(&1)|> IO.inspect(label: "=============")))
     |> Enum.to_list()

Here's the memory allocated to the lazy approach:

And to the greedy approach:

Wrapping Up

In this post, we first took a look at collections in Elixir and some of their associated challenges. We then explored two methods of processing large datasets: the greedy approach and the lazy approach using streams.

We've seen how working with streams is one of the best approaches to help create a memory-efficient Elixir application.

Happy coding!

P.S. If you'd like to read Elixir Alchemy posts as soon as they get off the press, subscribe to our Elixir Alchemy newsletter and never miss a single post!

Debugging Phoenix LiveView with open_browser/2

Tracey Onim — Tue, 04 Apr 2023 12:00:00 +0000

In this blog post, you'll see how useful open_browser/2 is when debugging LiveView tests. In addition, we'll give a brief introduction to testing LiveView.

Let's get started!

Introducing LiveView Testing

Testing during the software development process is one way to build confidence and ensure your application will work as expected.

The ability to easily write meaningful tests is an important factor for any framework, regardless of programming language.

Developers can easily test the LiveView framework's component, life-cycle, and behavior by writing LiveView tests with pure Elixir, since it uses ExUnit (a built-in testing framework) for all its testing. We can be confident in writing LiveView tests that are fast, concurrent, and stable.

You can test the functionality of your live views' behavior through the help of the Phoenix.LiveViewTest module, which offers convenient functions without the need to introduce JS testing frameworks. The test helpers assist us in writing meaningful tests for our LiveView modules with ease and speed.

Sample Feature in LiveView

Let's write a sample feature to use as our testing example.

We shall add a form to our application that allows users to enter their email address and password when attempting to register. This feature will only focus on the rendered HTML and not include the ability to add users into the system (the backend part of taking user params and adding them into the data store).

Let's begin!

First, start by creating a new LiveView application:

mix phx.new sample_live --live

NB: You can use an existing LiveView application if you have one.

The mix phx.new command with the --live flag will create a new application with LiveView installed and configured.

Add a live path in the router.

# lib/sample_live_web/router.ex

scope "/", SampleLiveWeb do

  live "/user_registration", RegistrationLive

end

Create registration_live.ex inside the sample_live_web folder.

# lib/sample_live_web/registration_live.ex

defmodule SampleLiveWeb.RegistrationLive do

  use SampleLiveWeb, :live_view


  def mount(_params, _session, socket) do
    {:ok, socket}
  end

  def render(assigns) do
    ~H"""

      <.form let={f} for={:changeset} id={"registration-form"} >

        <%= label f, :email %>
        <%= text_input f, :email, id: "email-input" %>
        <%= error_tag f, :email %>
        <%= label f, :password, id: "password-input" %>
        <%= password_input f, :password %>
        <%= error_tag f, :password %>
        <%= submit "Save" %>
      </.form>

    """
  end

end

In our LiveView application module, two callbacks have been defined: mount/3 and render/1.

The mount/3 callback expects three arguments — params, session, and liveview socket — and returns {:ok, socket}. When the LiveView page is rendered, the mount/3 callback will be invoked twice: once to perform the initial page load and again to establish the live connection.

The render/1 callback is responsible for displaying the HTML template/content — in this case, the added registration form. It receives the socket assigns and must return a template passed inside the ~H sigil. Whenever there is a change in our LiveView, the render/1 callback will be invoked.

In our template, we have defined a sample form where a user can enter their email and password.

Let's start the server and open the registration form in our browsers. You should see something similar to this:

Display the Feature on the Browser while Testing

When working on an HTTP-based web application, we usually write integration tests to validate passing expected attributes and properties to parts of our application.

We'll write an integration test validating user interactions with our application. The test should verify that when a user visits the /user_registration page, they can see the registration form properties/attributes rendered. In addition, we'll explore how we can use open_browser/2 to debug our LiveView test.

First, let's write a test for our form implemented above that will verify the HTML rendered. In here, we shall use open_browser/2 to verify the displayed form. The implemented registration form has an HTML id ("registration-form") that should help us select the element in our test.

# test/sample_live_web/registration_live_test.exs

defmodule SampleLiveWeb.RegistrationLiveTest do
  use SamplLiveWeb.ConnCase
  import Phoenix.LiveViewTest

  test "user can see registration form", %{conn: conn} do
    {:ok, view, html} = live(conn, "/user_registration")

    html =
      view
      |> element("#registration-form")
      |> open_browser()
      |> render()

    assert html =~ "Email</label>"
    assert html =~ "Password</label>"

  end

end

import Phoenix.LiveViewTest module gets access to the test helper functions.

In the sample feature, when a user is on the application and wants to visit the registration page, they will navigate to /user_registration. A similar thing happens here — we shall navigate the user to /user_registration using live/2, which performs a regular get(conn, path) and then upgrades the page to LiveView. It takes in the conn and the path then returns a three-element tuple with :ok, liveview process, and the rendered HTML.

open_browser/2 expects a view or element as the first argument. In the test, I have opted to pass an element, and this can be done by invoking element/3, passing the view and the query selector to it. element/3 returns an element which is then passed to open_browser/2. If open_browser/2 finds the form element with the query selector(#registration-form) within the LiveView page, we expect the following:

The default browser to be opened displaying the HTML of the form element.
The form element to be returned. The form element returned can be passed to render/1, which takes in a view_or_element and returns an HTML string of the view that we can finally assert in the test.

Run the test:

mix test test/sampl_live_web/registration_live_test.exs

We should expect the default browser to open and our test to pass.

Now, let's make the user enter their email and password by using form/3, which takes in the view, query selector, and form data, and then returns a form element. But this time we'll use a query selector that doesn't exist to demonstrate how open_browser/2 can be useful in debugging LiveView tests.

# test/sample_live_web/registration_live_test.exs

defmodule SampleLiveWeb.RegistrationLiveTest do
  use SamplLiveWeb.ConnCase
  import Phoenix.LiveViewTest

  test "user can see registration form", %{conn: conn} do
    {:ok, view, html} = live(conn, "/user_registration")

    view
    |> element("#registration-form")
    |> open_browser()

    view
    |> form("#wrong_registration-form", user: %{email: "hello@email.com", password: "hello123"})
    |> render()

  end
end

Let's run the test:

** (ArgumentError) expected selector "#wrong_registration-form" to return a single element, but got none within:

<main class="container"><p class="alert alert-info" role="alert" phx-click="lv:clear-flash" phx-value-key="info"></p><p class="alert alert-danger" role="alert" phx-click="lv:clear-flash" phx-value-key="error"></p><form action="#" method="post" id="registration-form" phx-submit="save"><input name="_csrf_token" type="hidden" value="CHwQRnBHAnF7JDQEXQk3AXx8QQ0hZxceYLY-91nA2PLS3bB13Q0HC8CQ"/><label for="registration-form_email">Email</label><input id="email-input" name="user[email]" type="text"/><label for="registration-form_password">Password</label><input id="password-input" name="user[password]" type="password"/><button type="submit">Save</button></form></main>

We notice that our test is failing with an ArgumentError. In this test, the form/3 function returns a form element with the query selector #wrong_registration-form. This form element is then passed to render/1, which expects the form element with the query selector #wrong_registration-form passed to it as the first argument to return a form element (but it gets none within the LiveView page). The test has made it clear that we are using the wrong query selector.

Our form template is very short, so it's easy to identify our mistake. Even from the HTML string displayed on the terminal after running the test, it's easy to point out the query selector we should have used.

Imagine working with larger HTML templates, though. It will be cumbersome to go through the HTML string to try and identify the mistake. Let's examine how open_browser/2 can come in handy in that case.

First, let's add open_browser/2 after form/3:

# test/sample_live_web/registration_live_test.exs

defmodule SampleLiveWeb.RegistrationLiveTest do
  use SamplLiveWeb.ConnCase
  import Phoenix.LiveViewTest

  test "user can see registration form", %{conn: conn} do
    {:ok, view, html} = live(conn, "/user_registration")

    view
    |> element("#registration-form")
    |> open_browser()

    view
    |> form("#wrong_registration-form", user: %{email: "hello@email.com", password: "hello123"})
    |> open_browser()
    |> render()

  end
end

This way, we expect the browser to open with the displayed registration form — but apparently, that won't be the case when we run our test. Just like the render/1 function, open_browser/2 expects the form element with the query selector #wrong_registration-form passed to it, but it doesn't exist.

The other option is to debug the LiveView page where the registration form is rendered. In the test, we shall call open_browser/2 and pass the view returned after invoking live/2 to it. Earlier on, I mentioned that open_browser/2 could either take in a view or an element. The option to use an element is failing due to the wrong query selector. So using view can be a better option in our case to help us understand why the test is failing.

# test/sample_live_web/registration_live_test.exs

defmodule SampleLiveWeb.RegistrationLiveTest do
  use SamplLiveWeb.ConnCase
  import Phoenix.LiveViewTest

  test "user can see registration form", %{conn: conn} do
    {:ok, view, html} = live(conn, "/user_registration")
    open_browser(view)
  end
end

The debug process will be as follows:

Run the test. This time, we should expect the default browser to be opened and our registration form displayed.
Use the element inspector in the browser to find the correct form selector.

From the image above, we can see that the form element under the inspect elements section highlighted in blue has a different query selector (the id attribute value) specified to the one passed in form/3 in our test. Our test expected the registration_form query selector. If we now refactor our test and pass the required query selector, it will definitely pass.

We have seen that we can use open_browser/2 as a debugging tool for our tests. With open_browser/2, we can easily verify that the correct HTML elements are rendered to users. If the correct view_or_element passes, the browser opens with the displayed HTML.

Monitor Your Phoenix LiveView App in Production with AppSignal

If you also want to keep track of your live view performance in production, you can set up LiveView instrumentation in AppSignal. Once you've integrated AppSignal with Phoenix LiveView, you'll be able to monitor incoming HTTP requests.

Check out our docs for Phoenix LiveView for more information.

We recommend using our automatic LiveView instrumentation in the majority of cases.

Wrapping Up

The introduction of open_browser/2 has made debugging LiveView tests easy and efficient. In this post, we introduced how to use open_browser/2 when testing to verify that the correct HTML is rendered within your live view.

Happy debugging!

P.S. If you'd like to read Elixir Alchemy posts as soon as they get off the press, subscribe to our Elixir Alchemy newsletter and never miss a single post!

How to Cache Locally in Elixir with Nebulex

Tracey Onim — Tue, 10 Jan 2023 11:43:24 +0000

In an Elixir application, you might need to access certain data frequently, which can be costly. The access time involved in retrieving data at every step can cause high latency, or even make the application crash (due to an increased workload on the database).

Caching is the best technique to store the most frequently accessed data and minimize database data retrieval, improving the overall performance of the application.

In this post, you'll learn how to use the Nebulex caching toolkit to:

cache locally in your Elixir applications
get familiar with different supported caching solutions/strategies
see how Nebulex can be a good fit for caching

Let's get going!

What Is Nebulex?

You've likely already come across some caching options for your existing Elixir application.

Here, I will introduce you to the Nebulex caching toolkit. Nebulex is an in-memory and distributed caching framework for Elixir that supports temporarily storing and accessing data in an Elixir application.

Why Use Nebulex in Your Elixir Application?

Caching can be implemented in many different ways depending on your application use case. You may need a local cache that stores data directly in your application or a distributed cache. Having a cache framework that supports a vast number of caching solutions is a plus. This is where we can leverage the Nebulex caching toolkit.

Nebulex has a flexible and pluggable architecture based on adapters. This makes it possible to integrate different caching options and also provides an easy workaround for various caching topologies, such as:

Partitioned caching: distribution of data between multiple nodes
Replicated caching: involving multiple servers, where each server has the same copy of data
Local caching: a single-server cache that resides on a single node

So Nebulex makes it easy for developers to scale their applications beyond a single node (with minimal impact on the code). You only need to choose the adapter that best suits your needs.

Check out this in-depth guide on supported caches and their adapters in Nebulex.

Now let's look at a real-world example where Nebulex proved useful.

The Problem: Fetching and Storing Data from an External API

A while ago, I was working on a payment integration system that involved fetching a transaction from an external API. The fetched transaction was then stored in the database.

We also had to confirm if the transaction could complete, so we scheduled a job using GenServer. GenServer sent a request to the external API after a certain period of time to confirm if the transaction was complete and then updated the stored transaction status.

This approach had the following shortcomings:

After every specified period, a request was sent to the external API to confirm if the transaction was completed. This occurred even when all fetched transactions stored in the database were completed.
After every specified period, a query ran to check if all transactions were complete. This approach resulted in unnecessary trips to the database and was deemed unfit when handling many transactions.

The Solution: Caching Locally with Nebulex

The application was being deployed to a single instance and would frequently access loaded transactions. So to solve the problem stated above, we used Nebulex to locally cache the fetched transactions.

This approach helped to overcome the following:

Making unnecessary trips to the external API after every specified period. It ensured requests were only made if the cache contained transactions. By default, all cached transactions were incomplete.
Making many trips to the database. Trips to the database were made only when inserting a fetched transaction and updating its status.

As mentioned earlier, Nebulex supports different caching solutions, one of which is local caching.

We'll now implement Nebulex in an Elixir application when caching locally.

Let's begin!

Add Nebulex to an Elixir Application

First, open the mix.exs file and add the nebulex dependency as shown:

defp deps do
  [
  {:nebulex, "~> 2.4"}

  ...

  ]
end

Install the Nebulex dependency:

mix deps.get

After installation, go ahead and generate your cache:

mix nbx.gen.cache -c Payment.Cache

This step involves:

Defining a cache module within the application
Setting up configurations

We'll use the name Payment.Cache to identify our cache (you can name yours as you wish). The command above will generate the Payment.Cache module defined in lib/payment/cache.ex.

#lib/payment/cache.ex
defmodule Payment.Cache do
  use Nebulex.Cache,
    otp_app: :my_app,
    adapter: Nebulex.Adapters.Local
end

This command generates the configuration Nebulex will use in config/config.exs,

#config/config.exs

config :my_app, Payment.Cache,
  # When using :shards as backend
  # backend: :shards,
  # GC interval for pushing new generation: 12 hrs
  gc_interval: :timer.hours(12),
  # Max 1 million entries in cache
  max_size: 1_000_000,
  # Max 2 GB of memory
  allocated_memory: 2_000_000_000,
  # GC min timeout: 10 sec
  gc_cleanup_min_timeout: :timer.seconds(10),
  # GC max timeout: 10 min
  gc_cleanup_max_timeout: :timer.minutes(10)

Finally, open lib/my_app/application.ex and add the Payment.Cache to the application supervision tree.

#lib/my_app/application.ex

use Application

  @impl true
  def start(_type, _args) do
    children = [

      Payment.Cache

      # ...

    ]

Setting Payment.Cache in the supervision tree starts the Nebulex process when the application starts up.

`Nebulex.Cache` Configuration in Elixir

The generated Payment.Cache module uses the Nebulex.Cache, a cache abstraction layer controlled by adapters. Nebulex.Cache expects :otp_app and :adapter as options.

:otp_app points to the Elixir application where nebulex can find the cache configuration. In our case, :my_app is specified.
:adapter - the desired adapter is configured at this point. In our case, Nebulex.Adapters.Local has been specified. It implements a local generation cache. This is a caching technique based on the age of the cached entries. It involves specifying the object's last modified date in the cache key.

The configurations specified in config/config.exs are specific to Nebulex.Adapters.Local passed in the :adapter option.

These are some of the options supported by Nebulex.Adapters.Local via the cache configuration:

:backend - the storage used for the adapter. The supported forms of storage are :ets and :shards, and the default is :ets.
:gc_interval - interval time in milliseconds for garbage collection to run, which involves deleting the oldest generation and creating a new one. Expects an integer > 0.
:max_size specifies the cache limit. Expects an integer > 0.
:allocated_memory - maximum size in bytes of the memory allocated for cache generation. Expects an integer > 0.
:gc_cleanup_min_timeout - minimum timeout in milliseconds for triggering the next clean-up and memory check. Defaults to 10_000 (10 seconds). Expects an integer > 0.
:gc_cleanup_max_timeout - maximum timeout in milliseconds for triggering the next clean-up and memory check. Defaults to 600_000 (10 minutes). Expects an integer > 0.

Query with `Nebulex.Cache` Callbacks

The Nebulex.Cache module has callbacks that can be leveraged to perform queries. Let's cache fetched transactions from an external API and manipulate cached entries using callbacks.

The implementation for manipulating cache entries is in lib/my_app/payment_cache.ex:

#lib/my_app/payment_cache.ex

defmodule MyApp.PaymentCache do
  @moduledoc """
  context for manipulating cache entries. It involves;
   - Inserting fetched transaction and incomplete transaction into the cache
   - Query for all cached transaction
   - Deleting cached transaction
  """
  alias Payment.Cache

  def insert_transaction(transaction) do
    Cache.put(transaction["id"], transaction)
  end

  def insert_all_transactions(transactions) do
    transactions
    |> Enum.map(fn transaction ->
      {transaction.id, %{id: transaction.id, status: transaction.status}}
    end)
    |> Cache.put_all()
  end

  def get_transaction(id) do
   Cache.get(id)
  end

  def all_cached_transactions do
    Cache.all()
  end

  def delete_transactions(transaction_ids) when is_list(transaction_ids) do
    unless Enum.empty?(transaction_ids) do
      Enum.each(transaction_ids, fn key -> delete_transaction(id) end)
    end
  end

  def delete_transaction(id) do
    Cache.delete(id)
  end
end

Here, we create a context for manipulating cache entries related to payments. We alias Payment.Cache (the module generated above in lib/payment/cache.ex). The module will give us access to the Nebulex.Cache callbacks.

The following functions are defined:

insert_transaction/1 - takes in a fetched transaction from the external API and inserts it into the cache. Payment.Cache.put/3 is responsible for inserting the transaction into the cache.
insert_all_transactions/1 - takes in a list of transactions and inserts them into the cache by invoking Payment.Cache.put_all/.
all_cached_transactions/0 - fetches all cached entries by invoking Payment.Cache.all/2.
delete_transaction/1 - deletes a cached entry by invoking Payment.Cache.delete/2.
get_transaction/1 - fetches a cached transaction from the given key by invoking Payment.Cache.get/2.

Now, open an interactive shell in your terminal, and put the functionality into use:

iex > alias MyApp.PaymentCache

iex > fetched_transaction = %{id: 1, status: "pending"}

        %{id: 1, status: "pending"}

iex > PaymentCache.insert_transaction(fetched_transaction)

      :ok

iex > PaymentCache.get_transaction(1)

      %{id: 1, status: "pending"}

iex > another_fetched_transaction =  %{id: 2, status: "pending"}

        %{id: 2, status: "pending"}

iex > PaymentCache.insert_transaction(another_fetched_transaction)

      :ok

iex > PaymentCache.all_cached_transactions()
      # returns the ids of the cached transactions
      [1, 2]

iex > PaymentCache.delete_transaction(1)

      :ok


iex > PaymentCache.all_cached_transactions()

      [2]

iex > PaymentCache.get_transaction(1)

      nil

Schedule a Job in Elixir Using Cached Entries

Caching fetched transactions comes in handy when scheduling jobs.

#lib/my_app/payment_scheduler.ex

defmodule MyApp.PaymentScheduler do
  use GenServer

  require Logger

  alias MyApp.PaymentCache

  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  @impl true
  def init(opts) do
    {:ok, opts, {:continue, :cache_incomplete_transactions}}
  end

  @impl true
  def handle_continue(:cache_incomplete_transactions, state) do

    insert_all_transaction()

    schedule_work()


    {:noreply, state}

  end

  @impl true
  def handle_info(:update, state) do
    update_pending_transactions()

    schedule_work()

    {:noreply, state}
  end

  defp schedule_work do
    Process.send_after(self(), :update, 10000)
  end

  defp update_pending_transactions do
    all_cached_transactions = PaymentCache.all_cached_transactions()

    case all_cached_transactions do
      [] -> Logger.info("All transactions are complete")

      all_cached_transactions ->

        complete_transactions = complete_transactions(all_cached_transactions)

        Payments.update_incomplete_transactions(complete_transactions)


        # delete cached transactions after update in the db
        PaymentCache.delete_transaction(complete_transactions)
    end
  end

  defp complete_transactions(cached_transactions) do

    #fetch all complete transactions from external API
    complete_transactions = Payment.confirmed_transactions()

    Enum.map(complete_transactions, fn transaction ->
      transaction.id in all_cached_transactions
      transaction.id
    end)

  end

  defp insert_all_transaction do
    pending_transactions = Transactions.all_pending_transactions()

    case Payment.all_pending_transactions() do
      [] ->  Logger.info("There are no incomplete transactions in the database")

      pending_transaction -> PaymentCache.insert_all_transaction(pending_transactions)
    end
  end
end

Let's run through what's going on here.

When an application starts, init/1 will be invoked. Our init/1 returns {:ok, opts, {:continue, :cache all incomplete transactions}}.

This will immediately trigger handle_continue(:cache_incomplete_transactions, state). Here, all incomplete transactions are fetched from the database and cached. By invoking schedule_work/0, an updated job is scheduled to take place every 10 seconds.

The handle_continue/2 callback allows us to perform a time-consuming job asynchronously, avoiding race condition cases. For more about handle_continue/2, I suggest Sophie DeBenedetto's excellent Elixir School article, 'TIL GenServer’s handle_continue/2'.

In schedule_work/0, an :update message notifies the parent process (the GenServer) that there is an update to perform. The :update message is then handled in the handle_info/2 callback. At this point, we confirm if there are incomplete transactions in the cache (keeping in mind that we only cache incomplete transactions).

When the cache is empty, that means there are no incomplete transactions. So we skip sending a request to an external API confirming whether the transactions have been completed, and we update their status in the database.

Wrapping Up

In this post, we explored how to use Nebulex for caching data locally in an Elixir application.

Implementing caching in an Elixir application depends purely on your business use case. When choosing a cache toolkit, make sure it meets your needs.

Nebulex's cache toolkit supports a vast number of caching solutions and allows you to implement different topologies with minimal impact on the code.

Visit Nebulex's guide to learn more.

Happy caching!

P.S. If you'd like to read Elixir Alchemy posts as soon as they get off the press, subscribe to our Elixir Alchemy newsletter and never miss a single post!

DEV Community: Tracey Onim

Leverage Concurrency Efficiently When Managing Multiple Tasks in Elixir

What is Concurrency?

How Can We Use Concurrency in Elixir?

Taking It One Step Further

Running Tasks Concurrently using the Task Module

Task.start/1

Task.async/1

An Example with an Error

A Problem: Working with a Series of Tasks

What's Going On Here and How Do We Fix It?

Sending Messages To the Phone Numbers with Task.async/1

The Solution: Use Task.async_stream/3

An Example Using Task.async_stream/3

:max_concurrency

:ordered

:timeout

:on_timeout

Monitor Memory with Erlang Observer

Wrapping Up

How to Build a Memory-efficient Elixir App with Streams

What are Collections in Elixir?

The Challenges of Collections

The Greedy Approach For Our Elixir Application

Step 1: Loading the CSV File Into Memory

Step 2: Create a List for Each Row In the CSV

Step 3: Create a Column from Each Row

Step 4: Create a Keyword List of Each Row

Step 5: Convert to Map

Observation with Enum

Solution: Build a Memory-efficient Elixir App with Streams

The Lazy Processing Approach with Streams in Elixir

Lazy Approach Using Streams Vs. Greedy Approach

Wrapping Up

Debugging Phoenix LiveView with open_browser/2

Introducing LiveView Testing

Sample Feature in LiveView

Display the Feature on the Browser while Testing

Monitor Your Phoenix LiveView App in Production with AppSignal

Wrapping Up

How to Cache Locally in Elixir with Nebulex

What Is Nebulex?

Why Use Nebulex in Your Elixir Application?

The Problem: Fetching and Storing Data from an External API

The Solution: Caching Locally with Nebulex

Add Nebulex to an Elixir Application

Nebulex.Cache Configuration in Elixir

Query with Nebulex.Cache Callbacks

Schedule a Job in Elixir Using Cached Entries

Wrapping Up

`Task.start/1`

`Task.async/1`

Sending Messages To the Phone Numbers with `Task.async/1`

The Solution: Use `Task.async_stream/3`

An Example Using `Task.async_stream/3`

`:max_concurrency`

`:ordered`

`:timeout`

`:on_timeout`

Observation with `Enum`

`Nebulex.Cache` Configuration in Elixir

Query with `Nebulex.Cache` Callbacks