The Primitives of Elixir Concurrency: a Full Example

March 19, 2019

X Follow Button

Previous articles about Concurrency in Elixir:

Let’s put in practice what we’ve seen in the last few articles about concurrency. In this article we see how to fully make our initial cryptocurrency example, using just HTTPoison module and spawn, send and receive to handle concurrency. At the end we will refactor it using Task, which makes everything far easier!

Our goal is to download concurrently different prices from Coinbase and return a map with the latest prices.

`get_price` function

Let’s see step by step how to get the current price of a single product.

iex> product_id = "BTC-USD"
iex> url = "https://api.pro.coinbase.com/products/#{product_id}/ticker"

iex> %HTTPoison.Response{body: body} = HTTPoison.get!(url)
%HTTPoison.Response{
  body: "{\"trade_id\":60538353, ..."
  headers: ...
  ...
}

iex> ticker = Jason.decode!(body)
%{
  "price" => "3891.76000000",
  ...
}

iex> price = String.to_float(ticker["price"])
3891.76

Elixir

Copy

Ok, it works and we get the price as a float, but I wouldn’t put this code in a function straightaway: it’s coupled and needs a bit of refactoring.

Decoupling

Let’s start with something simple, defining a function ticker_url(product_id) to return the URL string for the given product.

defmodule Coinbase.Client do
  @coinbase_base_url "https://api.pro.coinbase.com"

def ticker_url(product_id),
    do: "#{@coinbase_base_url}/products/#{product_id}/ticker"
end

Elixir

Copy

Easy.

Now, by using libraries like HTTPoison and Jason directly in our get_price function, we are coupling the get_price implementation with its external dependencies. If in the future we want to change our HTTP client to something like Tesla or Mint, we’ll need to change the implementation of all depending functions all over our code.

Instead of doing so, we can create a new module called Coinbase.HTTPClient, which we use to wrap HTTPoison and Jason.

defmodule Coinbase.HTTPClient do
  def get_json!(url) do
    HTTPoison.get!(url)
    |> Map.get(:body)
    |> Jason.decode!()
  end
end

Elixir

Copy

For brevity we are not handling any HTTPoison or Jason error. get_json!(url) makes a HTTP GET request and deserialises the JSON response, returning it in the form of a Map.

We now define our get_price function using only our HTTPClient module instead of HTTPoison and Jason.

defmodule Coinbase.Client do
  alias Coinbase.HTTPClient

def get_price(product_id) do
    product_id
    |> ticker_url()
    |> HTTPClient.get_json!()
    |> Map.get("price")
    |> String.to_float()
  end

end

Elixir

Copy

We see how this code is easier to read, thanks to the Elixir pipes, and also easier to change, since if we want to Jason to Poison or HTTPoison to Tesla, we just need to change our HTTPClient.get_json!/1 implementation.

This kind of refactoring is also shown in the José Valim’s article, Mocks and explicit contracts, in which we see how decoupling makes easier to test our code. If you haven’t read the article, I think it’s worth it.

In sequence

We define a new function to get multiple prices.

defmodule Coinbase.Client do
  def get_prices(products) do
    products
    |> Enum.map(&get_price/1)
  end
end

Elixir

Copy

get_prices/1 enumerates the products list and sequentially requests the price for each one, returning a list of prices.

As we saw in a previous article where we started talking about concurrency, getting prices one at a time, is not very efficient. Our computer is idle most of the time waiting for the response from the server.

Let’s see how much time it takes to sequentially get prices for seven different products, so we can compare it later with the concurrent one.

Sequential requests

To benchmark the time we define Coinbase.measure_time(func), which runs the passed function, calculating the elapsed time.

defmodule Coinbase do

def cyan_text(text) do
    IO.ANSI.cyan() <> text <> IO.ANSI.reset()
  end

def measure_time(func) do
    time_start = System.monotonic_time(:millisecond)
    result = func.()
    time_end = System.monotonic_time(:millisecond)
    seconds = (time_end - time_start)/1000
    cyan_text("time #{seconds}s") |> IO.puts()
    result
  end

end

Elixir

Copy

After reading Cool CLIs in Elixir (Part 2) with IO.ANSI, I decided to add some color to the benchmarking output. The time will then be printed in cyan, using the cyan_text function.

Great, let’s get the prices of these seven products, sequentially, and see how much time we need.

iex> products = ["BTC-USD","ETH-USD","LTC-USD",\
...>  "BCH-USD","XRP-USD","XLM-USD",\
...>  "ZRX-USD"]

iex> import Coinbase, only: [measure_time: 1]
iex> measure_time fn ->
...>   Coinbase.Client.get_prices(products)
...> end

time 2.132s
[3895.98, 136.01, 58.04, 141.6, 0.3131, 0.105945, 0.271112]

Elixir

Copy

The function returns a list of prices: the first price correspond to the first product in the list “BTC-USD”, the second to “ETH-USD” etc. To make it easier to understand it’s better to return a Map like this

%{
    "BTC-USD" => 3895.98,
    "ETH-USD" => 136.01,
    ...
}

Elixir

Copy

where the product is the key of the map, and the price is the value.

Let’s change in get_prices/1 the function passed to Enum.map, returning a tuple with both product_id and price. We can then use Enum.into/2 to convert the list of 2-element tuples to a Map.

def get_prices(products) do
  products
  |> Enum.map(fn product_id ->
    {product_id, get_price(product_id)}
  end)
  |> Enum.into(%{})
end

Elixir

Copy

And this time it returns a human readable result

iex> Coinbase.Client.get_prices(products)
%{
  "BCH-USD" => 141.66,
  "BTC-USD" => 3895.99,
  "ETH-USD" => 136.05,
  ...
}

Elixir

Copy

Concurrent

We saw thatspawn runs the given function in a new process, returning immediately. We were able to make our requests concurrent. But, without messages we were only able to print the result.

We are now making the get_prices/1 concurrent without changing the function result. The output will be still a map with products and prices.

Let’s start focusing on making one single concurrent request, then scaling to multiple requests will be easy.

def spawn_and_send_price(product_id, dst_pid) do
  spawn fn ->
    price = get_price(product_id)
    send dst_pid, {self(), {product_id, price}}
  end
end

Elixir

Copy

spawn_and_send_price/2 creates a new process and returns immediately its pid. As we saw in the previous article, the way we receive the result back is using messages. For this reason we pass dst_pid as second parameter, which is the pid where we want to receive the result back in form of a tuple.

{from_pid, {product_id, price}}

Elixir

Copy

When we send the message back to dst_pid, we pass self() as first element of the tuple.self() returns the pid of the process where is called. In that case the pid is the one of the spawned process.

It’s always good to send some reference along with the message, so it’s easier for the receiver to understand what the message is about and from who is coming. Passing the pid is also useful to filter messages, in thereceive block, coming just from the chosen process.

Let’s try this function on iex.

iex> self
#PID<0.205.0>
iex> pid = Coinbase.Client.spawn_and_send_price("BTC-USD",self())
#PID<0.208.0>

iex> receive do
...>   {^pid, result} -> result
...> end
{"BTC-USD", 3899.98}

Elixir

Copy

iex spawns a process which requests the price and sends it back to iex mailbox

Nice, we can define await/1 with exactly the receive block we’ve just used.

def await(pid) do
  receive do
    {^pid, result} -> result
  end
end

Elixir

Copy

await(pid) waits to receive, and returns, the result sent from the given pid.

We now have everything to write a compact concurrent get_prices(products) function, using Elixir pipes.

def get_prices(products) do
  products
  |> Enum.map(fn product ->
    spawn_and_send_price(product, self())
  end)
  |> Enum.map(&await/1)
  |> Enum.into(%{})
end

Elixir

Copy

The first Enum.map runs spawn_and_send_price(product_id, dst_pid)for each product, returning a list of pids.
The second Enum.map takes the list of pids as input, running await(pid) for each pid. It then returns the list of results.
Enum.into transforms the list of tuples [{"BTC-USD",3902.0}, {"ETH-USD", 135.98}, ..], in a Map.

multiple concurrent requests and receiving results

Let’s try it while measuring the time on iex.

iex> measure_time fn ->
...>   Coinbase.Client.get_prices(products)
...> end

time 0.421s
%{
  "BTC-USD" => 3902.0,
  "ETH-USD" => 135.98,
  "LTC-USD" => 58.08,
  "BCH-USD" => 142.52,
  "XRP-USD" => 0.3133,
  "XLM-USD" => 0.106001,
  "ZRX-USD" => 0.269627
}

Elixir

Copy

Fantastic, only 0.421 seconds. We see how this version is faster than the other one which took 2.132 seconds.

Task

We’ve built our concurrent function using directly spawn, send and receive. This was useful to understand how concurrency is handled in Elixir, but usually it’s much better to use modules like Task, which makes concurrency much easier.

Conveniences for spawning and awaiting tasks.

Tasks are processes meant to execute one particular action throughout their lifetime, often with little or no communication with other processes.

Elixir Task

We refactor get_prices(products), getting rid of the two functions we wrote, spawn_and_send_price(product_id,dst_pid) and await(pid). We now use Task.async and Task.await

def get_prices(products, :task) do
  products
  |> Enum.map(&Task.async(fn -> {&1, get_price(&1)} end))
  |> Enum.map(&Task.await/1)
  |> Enum.into(%{})
end

Elixir

Copy

The dynamic is really similar of the one we saw in our custom version. If we run just Task.async, we see that the messages are sent to our current process.

iex> products \
...> |> Enum.map(&Task.async(\
...> fn ->
...>  {&1, Coinbase.Client.get_price(&1)}
...> end))
[%Task{}, %Task{}, ...]

iex> iex(3)> :erlang.process_info self(), :messages
{:messages,
 [\
   {#Reference<...>, {"ETH-USD", 139.63}},\
   {#Reference<...>, {"LTC-USD", 60.52}},\
   {#Reference<...>, {"BCH-USD", 152.76}},\
   ...\
]}

Elixir

Copy

Task.async returns a Task struct instead of a pid, which will be used by Task.await to get the result from the iex process’ mailbox.

For the most passionate, a small challenge for you!

After playing around with the code above, try to remove the last step Enum.into into get_prices. The function will then return a list of tuples {product, price}.

Try to get the prices multiple times! Do you see anything particular? The requests we make are concurrent, each one takes a random and different amount of time, but the results are always in the same order. Do you know why?

Any ideas on how to put the list of results in order from quickest request to the slowest?

Feel free to answer below, in the comments section! I look forward for your suggestions!

Credits

A special thanks to Greg Vaughn, who helped me give a perfect title to this article!

tempest.services.disqus.com is blocked

This page has been blocked by an extension

Try disabling your extensions.

ERR_BLOCKED_BY_CLIENT

Reload

This page has been blocked by an extension

Disqus Comments

We were unable to load Disqus. If you are a moderator please see our troubleshooting guide.

Join the discussion…

Comment

Log in with

or sign up with Disqus or pick a name

Disqus is a discussion network

Don't be a jerk or do anything illegal. Everything is easier that way.

Read full terms and conditions

This comment platform is hosted by Disqus, Inc. I authorize Disqus and its affiliates to:

Use, sell, and share my information to enable me to use its comment services and for marketing purposes, including cross-context behavioral advertising, as described in our Terms of Service and Privacy Policy, including supplementing that information with other data about me, such as my browsing and location data.
Contact me or enable others to contact me by email with offers for goods or services
Process any sensitive personal information that I submit in a comment. See our Privacy Policy for more information

Acknowledge I am 18 or older

1
Discussion Favorited!

Favoriting means this is a discussion worth sharing. It gets shared to your followers' Disqus feeds, and gives the creator kudos!

Find More Discussions

Tweet this discussion
- Share this discussion on Facebook
- Share this discussion via email
- Copy link to discussion
Best
- Newest
- Oldest

I enjoy a lot reading your Articles! #3

About why the order is always the same, it is because firstly you wait for the first msg, then second and so on. I think this could be solved passing a list of the awaited PIDs to the await function, so any of the requests can match at first.

Hi Juan Carlos!

Yes, you got it! :D

Regarding the await is not possible to add a guard clause in the `receive` block, like

{pid, result} when pid in pids -> result

since the pids list should be fixed.

So, on this what I would do is to give a reference to the current download, and use the reference to pin the messages

We create a ref for each get_prices request. The ref is passed to the Enum.map, so the message we now receive is something like {ref, {pid, result}}.

def get_prices(products) do
  ref = make_ref()
  products
  |> Enum.map(fn product-> {ref, spawn_and_send_price(product)}} end)
  |> Enum.map(&await(&1,ref))
  |> Enum.into(%{})
end

And the await function accepts the the ref as second argument, we could just only pass the ref and then number of messages we suppose tu receive.

def await(pids,ref) when is_list(pids) do
  Enum.map pids, fn _->
    receive do
      {^ref, {_pid, result}} -> result
    end
  end
end

What do you think?

aahh yes, I forgot to take into account that guards were for fixed lists.

Your solution seems to work great.

One doubt, why do you pass two arguments at Enum.map(&await(&1, ref)) you should only need the second one?

Thanks for the post! Keep up this amazing site!

Good point. Yes you are right... my mistake, I did the map two times, one in get_prices and the other one in await. So as you said we could have something like

def get_prices(products) do
  ref = make_ref()
  products
  |> Enum.map(fn product-> {ref, spawn_and_send_price(product)}} end)
  |> Enum.map(fn _-> await(ref) end)
  |> Enum.into(%{})
end

def await(ref) do
  receive do
      {^ref, {_pid, result}} -> result
    end
end

So, in this way Enum.map(fn _-> await(ref) end) should call await just one time for each pid.

Thanks :D

Show more replies

Thank you for the post.

I enjoy reading your articles. I found a little bit not good parts. for example:

Enum.map(&spawn_and_send_price/1)

you actually define spawn_and_send_price/2 in your code so that we need add default parameter to it if we want to use &spawn_and_send_price/1

Hi lambeta, thanks a lot for telling me this issue. I've just updated the code example.

I enjoy a lot reading your Articles! #2

I enjoy a lot reading your Articles!

thanks :D

Load more comments

live.rezync.com

live.rezync.com is blocked

This page has been blocked by an extension

Try disabling your extensions.

ERR_BLOCKED_BY_CLIENT

Reload

This page has been blocked by an extension

pippio.com

pippio.com is blocked

This page has been blocked by an extension

Try disabling your extensions.

ERR_BLOCKED_BY_CLIENT

Reload

This page has been blocked by an extension

tempest.services.disqus.com

tempest.services.disqus.com is blocked

This page has been blocked by an extension

Try disabling your extensions.

ERR_BLOCKED_BY_CLIENT

Reload

This page has been blocked by an extension

Elixir

Nerves powered Vision – Deploy YOLOv8 on RPi5 with…

Alvise Susmel

Sep 5, 202514 sec read

Elixir

Building a YOLOX Plate Detector – Setup, Fine-Tuning, Metrics,…

Alvise Susmel

Aug 29, 20253 min read

Elixir

Fine-Tuning YOLO to Watch Soccer Matches

Alvise Susmel

Jul 17, 20255 min read

Twitter Widget Iframe

The Primitives of Elixir Concurrency: a Full Example

get_price function

Decoupling

In sequence

Concurrent

Task

For the most passionate, a small challenge for you!

Credits

Share this:

tempest.services.disqus.com is blocked

Log in with

or sign up with Disqus or pick a name

Disqus is a discussion network

Discussion Favorited!

live.rezync.com is blocked

pippio.com is blocked

tempest.services.disqus.com is blocked

Nerves powered Vision – Deploy YOLOv8 on RPi5 with…

Building a YOLOX Plate Detector – Setup, Fine-Tuning, Metrics,…

Fine-Tuning YOLO to Watch Soccer Matches

`get_price` function