In Erlang, and therefor Elixir which leverages the Erlang VM (BEAM), it makes writing and reasoning about concurrent code feel effortless. While Ruby has some great libraries<\/a> for helping write concurrent code, with Elixir it\u2019s built-in and a first-class citizen.<\/p>\n That isn\u2019t to say that writing highly concurrent or distributed systems is easy. Far from it! But with Elixir, the language is on your side.<\/p>\n Before we look at how to go about writing concurrent code in Elixir, it\u2019s a good idea to understand the terms that we\u2019ll be using and the model of concurrency that Elixir employs.<\/p>\n Concurrency in Elixir (and Erlang) is based upon the Actor Model<\/a>. Actors are single threaded processes which can send and receive messages amongst themselves. The Erlang VM manages their creation, execution, and their communication. Their memory is completely isolated, which makes having to worry about \u201cshared state\u201d a non-issue.<\/p>\n So to bring it all together, a process in Elixir is the actor. It can communicate with another actor by sending a message to a specific PID. The recipient can receive a message by checking its mailbox for new messages.<\/p>\n In this section, we\u2019ll look at how the Actor Model for concurrency is actually used within Elixir.<\/p>\n Creating a new process is done with the Everything in Elixir runs within a process. You can find out the PID of your current process by calling the We can use this PID to ask Elixir for information about the process. This is done using the It\u2019s interesting what you can find here! For example, in Messages can be sent to a process using the Sending is only half the battle though. If the recipient isn\u2019t prepared to receive the message, it will fall on deaf ears. A process can receive messages by using the In the example below, we spawn a new process which waits to receive a message. Once it has received a message in its mailbox, we\u2019ll simply output it to the screen.<\/p>\n A process exits when it no longer has any code to execute. In the example above, the process will stay alive until it has received its first message, then exit. So the question then arises: How do we get a long running process?<\/p>\n We can do this by utilizing a Our current process doesn\u2019t track any state. It simply executes its code without maintaining any extra state or information.<\/p>\n What if we wanted our logger to keep track of some stats, like the number of messages it has logged? Notice the call Now, when the Another thing we\u2019ve done is added an additional action our logger can perform. It can now log messages and also print out the stats. To do this, we\u2019ll send our messages as a We can refactor our module a little bit to make it more user friendly. Instead of directly using the Our server module is quite simple. It consists of an If we are to use the code below, we don\u2019t really need to know how the server is implemented. We interact directly with the client, and it in turn sends messages to the server. I\u2019ve aliased the module just to avoid typing Notice that all of the messages being received by the server are being pattern matched in order to determine how to handle them? We can do better than having a single large function by creating a series of \u201chandler\u201d functions that pattern match on the data being received.<\/p>\n Not only does this clean up our code, it also makes it much easier to test. We can simply call the individual For a final example, let\u2019s take a look at performing a parallel map.<\/p>\n What we\u2019ll be doing is mapping a list of URLs to their returned HTTP status code. If we were to do this without any concurrency, our speed would be the sum of the speed of checking each URL. If we had five and each took oe second, it would take approximately five seconds to finish checking all the URLs. If we could check them in parallel though, the amount of time would be about one second, the time of the slowest URL since they are happening all at once.<\/p>\n Our test implementation looks like this:<\/p>\n Now for the implementation of the actual code. I\u2019ve added comments to make it clear what each step is doing.<\/p>\n When we ran the code, it took 2.2 seconds. This makes sense because one of the URLs is a faker URL service that we told to delay the response by two seconds\u2026so it took approximately the time of the slowest URL.<\/p>\n In this article, we covered the basics of spawning a new process, sending that process a message, maintaining state in the process via recursive looping, and receiving messages from other processes. This is a good start, but there is a lot more!<\/p>\n Elixir comes with some very cool modules to help us remove some of the boilerplate involved in what we did today. Lastly there is the whole topic of linking, monitoring, and responding to errors which may occur in a process. Elixir comes with a Supervisor<\/a> module for this and is all part of building a dependable fault-tolerant system.<\/p>\nProcesses, PIDs, and Mailboxes<\/h2>\n
Actor Model<\/h3>\n
![](\"http:\/\/www.javacodegeeks.com\/wp-content\/uploads\/2017\/02\/actors.png\")
<\/a><\/p>\n\n
Writing Concurrent Code<\/h2>\n
Creating processes<\/h3>\n
spawn<\/code> or spawn_link<\/code> functions. This function accepts an anonymous function which will be invoked in a separate process. In response, we are given a process identifier, often referred to as a PID. This is important if we want to communicate with this process going forward or ask the kernel for information about the process.<\/p>\npid = spawn(fn -> :timer.sleep 15000 end)\r\n#PID<0.89.0><\/pre>\n
self()<\/code> function. So even when you are in the iex<\/code> shell, by calling self()<\/code> you can see the PID for that iex session, something like #PID<0.80.0><\/code>.<\/p>\nProcess.info(pid)<\/code> function.<\/p>\nProcess.info(pid)\r\n\r\n[current_function: {:timer, :sleep, 1}, initial_call: {:erlang, :apply, 2},\r\n status: :waiting, message_queue_len: 0, messages: [], links: [],\r\n dictionary: [], trap_exit: false, error_handler: :error_handler,\r\n priority: :normal, group_leader: #PID<0.50.0>, total_heap_size: 233,\r\n heap_size: 233, stack_size: 2, reductions: 43,\r\n garbage_collection: [max_heap_size: %{error_logger: true, kill: true, size: 0},\r\n min_bin_vheap_size: 46422, min_heap_size: 233, fullsweep_after: 65535,\r\n minor_gcs: 0], suspending: []]<\/pre>\niex<\/code> if you ask for the info about itself Process.info(self())<\/code>, you\u2019ll see the history of the commands you\u2019ve typed:iex(1)> 5 + 5\r\niex(2)> IO.puts \"Hello!\"\r\niex(3)> pid = spawn(fn -> :timer.sleep 15000 end)\r\niex(4)> Process.info(self())[:dictionary][:iex_history]\r\n\r\n%IEx.History.State{queue: {[\r\n {3, 'pid = spawn(fn -> :timer.sleep 15000 end)\\n', #PID<0.84.0>},\r\n {2, 'IO.puts \"Hello!\"\\n', :ok}],\r\n [{1, '5 + 5\\n', 10}]},\r\n size: 3, start: 1}<\/pre>\nSending messages<\/h3>\n
send<\/code> function. You provide it with the PID of the process you wish to send a message to along with the data being sent. The message is sent to the receiving processes\u2019 mailbox.<\/p>\nreceive<\/code> construct, which pattern matches on the messages being received.<\/p>\npid = spawn(fn ->\r\n IO.puts \"Waiting for messages\"\r\n receive do\r\n msg -> IO.puts \"Received #{inspect msg}\"\r\n end\r\nend)\r\n\r\nsend(pid, \"Hello Process!\")<\/pre>\nKeeping our process alive<\/h3>\n
loop<\/code> function that calls itself recursively. This loop will simply receive a message and then call itself to wait for the next one.<\/p>\ndefmodule MyLogger do\r\n def start do\r\n IO.puts \"#{__MODULE__} at your service\"\r\n loop()\r\n end\r\n\r\n def loop do\r\n receive do\r\n msg -> IO.puts msg\r\n end\r\n loop()\r\n end\r\nend\r\n\r\n# This time we will spawn a new processes based on the MyLogger module's method `start`.\r\npid = spawn(MyLogger, :start, [])\r\n\r\nsend(pid, \"First message\")\r\nsend(pid, \"Another message\")<\/pre>\nMaintaining state<\/h3>\n
spawn(MyLogger, :start, [])<\/code>; the last parameter, which is an empty list, is actually a list of args that can be passed to the process. This acts as the \u201cinitial state\u201d or what is passed to the entry point function. Our state will simply be a number that tracks the number of messages we\u2019ve logged.<\/p>\ninit<\/code> function is called, it will be passed the number 0<\/code>. It\u2019s up to us to keep track of this number as we do our work, always passing the updated state to the next loop of our process.<\/p>\ntuple<\/code> where the first value is an atom<\/code> that represents the command we want our process to perform. Pattern matching in the receive<\/code> construct allows us to differ one message\u2019s intent from another.<\/p>\ndefmodule MyLogger do\r\n def start_link do\r\n # __MODULE__ refers to the current module\r\n spawn(__MODULE__, :init, [0])\r\n end\r\n\r\n def init(count) do\r\n # Here we could initialize other values if we wanted to\r\n loop(count)\r\n end\r\n\r\n def loop(count) do\r\n new_count = receive do\r\n {:log, msg} ->\r\n IO.puts msg\r\n count + 1\r\n {:stats} ->\r\n IO.puts \"I've logged #{count} messages\"\r\n count\r\n end\r\n loop(new_count)\r\n end\r\nend\r\n\r\npid = MyLogger.start_link\r\nsend(pid, {:log, \"First message\"})\r\nsend(pid, {:log, \"Another message\"})\r\nsend(pid, {:stats})<\/pre>\nRefactoring into a client and server<\/h3>\n
send<\/code> function, we can hide the details behind a client module. Its job will be to send messages to the process running the server module and optionally wait for a response for synchronous calls.<\/p>\ndefmodule MyLogger.Client do\r\n def start_link do\r\n spawn(MyLogger.Server, :init, [0])\r\n end\r\n\r\n def log(pid, msg) do\r\n send(pid, {:log, msg})\r\n end\r\n\r\n def print_stats(pid) do\r\n send(pid, {:print_stats})\r\n end\r\n\r\n def return_stats(pid) do\r\n send(pid, {:return_stats, self()})\r\n receive do\r\n {:stats, count} -> count\r\n end\r\n end\r\nend<\/pre>\ninit<\/code> function which doesn\u2019t do much in this case other than start the loop<\/code> function looping. The loop<\/code> function is in charge of receiving messages from the mailbox, performing the requested task and then looping again with the updated state.<\/p>\ndefmodule MyLogger.Server do\r\n def init(count \\\\ 0) do\r\n loop(count)\r\n end\r\n\r\n def loop(count) do\r\n new_count = receive do\r\n {:log, msg} ->\r\n IO.puts msg\r\n count + 1\r\n {:print_stats} ->\r\n IO.puts \"I've logged #{count} messages\"\r\n count\r\n {:return_stats, caller} ->\r\n send(caller, {:stats, count})\r\n count\r\n end\r\n loop(new_count)\r\n end\r\nend<\/pre>\nMyLogger.Client<\/code> various times.<\/p>\nalias MyLogger.Client, as: Logger\r\n\r\npid = Logger.start_link\r\nLogger.log(pid, \"First message\")\r\nLogger.log(pid, \"Another message\")\r\nLogger.print_stats(pid)\r\nstats = Logger.return_stats(pid)<\/pre>\n
Refactoring the server<\/h3>\n
handle_receive<\/code> functions with the correct arguments to test that they are working correctly.<\/p>\ndefmodule MyLogger.Server do\r\n def init(count \\\\ 0) do\r\n loop(count)\r\n end\r\n\r\n def loop(count) do\r\n new_count = receive do\r\n message -> handle_receive(message, count)\r\n end\r\n loop(new_count)\r\n end\r\n\r\n def handle_receive({:log, msg}, count) do\r\n IO.puts msg\r\n count + 1\r\n end\r\n\r\n def handle_receive({:print_stats}, count) do\r\n IO.puts \"I've logged #{count} messages\"\r\n count\r\n end\r\n\r\n def handle_receive({:return_stats, caller}, count) do\r\n send(caller, {:stats, count})\r\n count\r\n end\r\n\r\n def handle_receive(other, count) do\r\n IO.puts \"Unhandled message of #{inspect other} received by logger\"\r\n count\r\n end\r\nend<\/pre>\nParallel map<\/h3>\n
defmodule StatusesTest do\r\n use ExUnit.Case\r\n\r\n test \"parallel status map\" do\r\n urls = [\r\n url1 = \"http:\/\/www.fakeresponse.com\/api\/?sleep=2\",\r\n url2 = \"http:\/\/www.fakeresponse.com\/api\/?sleep=1\",\r\n url3 = \"http:\/\/www.fakeresponse.com\/api\/?status=500\",\r\n url4 = \"https:\/\/www.leighhalliday.com\",\r\n url5 = \"https:\/\/www.reddit.com\"\r\n ]\r\n assert Statuses.map(urls) == [\r\n {url1, 200},\r\n {url2, 200},\r\n {url3, 500},\r\n {url4, 200},\r\n {url5, 200}\r\n ]\r\n end\r\nend<\/pre>\ndefmodule Statuses do\r\n def map(urls) do\r\n # Put self into variable to send to spawned process\r\n caller = self()\r\n urls\r\n # Map the URLs to a spawns process. Remember a `pid` is returned.\r\n |> Enum.map(&(spawn(fn -> process(&1, caller) end)))\r\n # Map the returned pids\r\n |> Enum.map(fn pid ->\r\n # Receive the response from this pid\r\n receive do\r\n {^pid, url, status} -> {url, status}\r\n end\r\n end)\r\n end\r\n\r\n def process(url, caller) do\r\n status =\r\n case HTTPoison.get(url) do\r\n {:ok, %HTTPoison.Response{status_code: status_code}} ->\r\n status_code\r\n {:error, %HTTPoison.Error{reason: reason}} ->\r\n {:error, reason}\r\n end\r\n # Send message back to caller with result\r\n send(caller, {self(), url, status})\r\n end\r\nend<\/pre>\nWhere to go from here?<\/h2>\n
Agent<\/code> is a module for maintaining state in a process. Task<\/code> is a module for running code concurrently and optionally receiving its response. GenServer<\/code> handles both state and concurrent tasks in a long standing process. I plan on covering these topics in a second article in this series.<\/p>\n