• Change the ^pid in pmap to _pid. This means that the receive block will take responses in the order the processes send them. Now run the code again. Do you see any difference in the output? If you’re like me, you don’t, but the program clearly contains a bug. Are you scared by this? Can you find a way to reveal the problem (perhaps by passing in a different function, or by sleeping, or increasing the number of propcesses)? Change it back to ^pid and make sure the order is now correct.

Pass the list through Enum.reverse to set up a situation in which it is likely that the processes will answer in the “wrong” order, like this:

defmodule Parallel do
  def pmap(collection, fun) do
    me = self

    collection
    |> Enum.map(&(spawn_link fn -> (send me, { self, fun.(&1) }) end))
    |> Enum.reverse
    |> Enum.map(&(receive do { ^&1, result } -> result end))
  end
end

Simply removing the ^ in the 8th line above will demonstrate the problem.

And, of course, if you still want the answers in the original order you can pass them through Enum.reverse again after the receive line.

Daniel, your solution doesn’t show the problem on my machine. Even more curious when changing the first time it showed the problem right away

iex(9)> Parallel2.pmap 1..10, &(&1 * &1) 
[64, 81, 100, 1, 4, 9, 16, 25, 36, 49]   
iex(10)> Parallel2.pmap 1..10, &(&1 * &1)
[64, 81, 100, 1, 4, 9, 16, 25, 36, 49]  
iex(11)> Parallel2.pmap 1..10, &(&1 * &1) 
[64, 81, 100, 1, 4, 9, 16, 25, 36, 49]

Then on next day the result was in sequence. I was increasing the count from 1..10 to 1..40 and it showed up again.

iex(34)> Parallel2.pmap 1..10, &(&1 * &1)
[1, 4, 9, 16, 25, 36, 49, 64, 81, 100]
iex(35)> Parallel2.pmap 1..40, &(&1 * &1)
[1, 4, 9, 16, 25, 36, 49, 64, 81, 100, 121, 144, 169, 196, 256, 289, 225,
 324, 361, 400, 441, 484, 529, 576, 625, 676, 729, 784, 841, 900, 961,
 1024, 1089, 1156, 1225, 1296, 1369, 1444, 1521, 1600]
iex(36)> Parallel2.pmap 1..10, &(&1 * &1)
[1, 4, 9, 16, 25, 36, 49, 64, 81, 100]

I think the result in case of such an error is completely unpredictable. I would like to know if there is a way to monitor the processes, wo when which process is conducting its task.

It’s enough to put some load on the process, to reveal the problem:

iex(15)> busy_guy = fn x ->
...(15)> Enum.map 1..10000, fn i -> :math.sin(i) end
...(15)> x
...(15)> end
#Function<6.90072148/1 in :erl_eval.expr/5>
iex(16)> busy_guy.(1)
1
iex(17)> Parallel.pmap [1,2,3,4], busy_guy
[1, 4, 2, 3]
iex(18)> Parallel.pmap [1,2,3,4], busy_guy
[4, 1, 3, 2]
iex(19)> Parallel.pmap [1,2,3,4], busy_guy
[1, 4, 2, 3]
iex(20)> Parallel.pmap [1,2,3,4], busy_guy
[4, 2, 3, 1]

Really like Dave’s answer https://forums.pragprog.com/users/8520

I went with a sleep based on the value of the element (assuming they’re all positive numbers below 10 which really only applies to this simple example).

defmodule Parallel do
  def pmap(collection, fun) do
    me = self
    collection
    |> Enum.map(fn (elem) ->
      spawn_link fn ->
        :timer.sleep(100 - (elem * 10))
        send me, {self, fun.(elem)}
      end
    end)
    |> Enum.map(fn (pid) ->
      receive do {^pid, result} -> result end
    end)
  end
end

Parallel.pmap [1, 2, 3, 4, 5], &(&1*&1)

What Dave is trying to show here is that using ^pid with receive we ensure that we process the messages in the correct order, it is requiring that we process the message for the next pid and thus process each message in order. However if we don’t do that it’s possible that we could process messages in a random order. The order we receive the messages is dependent on a lot of things and the length of time each process takes is often the largest factor. I think the best way to demonstrate this is to use sleep in each process to change the processing time, as Diogo mentioned. In order to make the processing times truly different for each process I used a random sleep time of 1 to 50 msec.

defmodule Parallel do
  def pmap(collection, fun) do
    me = self
    collection
    |> Enum.map(fn (elem) ->
       spawn_link fn ->(
           :timer.sleep round(:rand.uniform * 50)
           send me, {self, fun.(elem) })
         end
       end)
    |> Enum.map(fn (pid) ->
         receive do { ^pid, result } -> result end
       end)
  end
end

Running the code as it it produces the expected results in the expected order:

[1, 4, 9, 16, 25, 36, 49, 64, 81, 100]

However if I change ^pid to _pid (we won’t be using pid) our results will vary, here is what I saw after running it the first time:

[25, 4, 81, 100, 64, 1, 9, 49, 16, 36]

Again the reason the second set of results is out of order is because each process is taking some arbitrary length of time, some more than others, and we process the messages as they are received, unlike the first set where we required that the numbers be processed in order.

Got this to go out of order by having each process wait a random amount of time before sending the calculation to the parent.

defmodule Parallel do
  def calculate(parent, elem, fun) do
    :timer.sleep(:rand.uniform(500))
    send parent, {self, fun.(elem)}
  end

  def pmap(collection, fun) do
    me = self
    collection
    |> Enum.map(fn (elem) ->
        spawn_link(Parallel, :calculate, [me, elem, fun])
       end)
    |> Enum.map(fn (_pid) ->
        receive do { _pid, result } -> result end
       end)
  end
end

I went with the same approach as Stefan Chrobot, just putting a little bit of a load with a :math function to get it out of order.

What I find interesting is that if I run the pinned version and then right after run the unpinned version, the return value is always the same [1, 4, 2, 3] but subsequent return values are not always the same.