Task.start vs TaskSupervisor

Agree.

To compare, I started to write an alternative with GenServer as you described. I had also a registry to be able to cast by name during the lifetime of the channel, and the solution based on Task.Supervisor.async_stream_nolink/6 wins in simplicity.

The cost is less throghtput, but as a first implementation of this service I prefer to keep it simple first, and introduce the GenServer if usage shows it is justified. On the other side, the synchronous approach would allow us to send an ACK status back to the client (not a requirement right now, though).

Thanks man!

So I would go with the current approach, where the channel is blocked. This means that actions on the client won’t receive a reply until the publishing is done - but other than that it is generally fine. I assume the publishing to the message bus is super fast, so that’s how I would roll honestly.

Or I would create a GenServer on every new channel with the purpose of keeping ordering. Every time you have to publish, you just cast the GenServer to do it for you. The GenServer would be started in its own supervisor (a DynamicSupervisor) and it would monitor the channel. Once the channel is DOWN, it exits.

https://github.com/elixir-lang/elixir/blob/v1.9.4/lib/elixir/lib/task.ex#L602-L603

Kernel.exit/1

See also:

Kernel.raise/1 is an Elixir style exception - exit/1 is the original Erlang style process exception which can be caught within the process but will terminate that process if it is not caught.

Therefore, it receives exit signals from the task as messages.

Signals relate to process links (and convert to :EXIT messages when trapped) - :DOWN messages relate to monitors - two entirely separate mechanisms.

it is not clear to me if Task.await/1 is correct in a Phoenix Channel at all because it enters a receive block to consume messages directly from the process mailbox. What do you think?

That is the fundamental issue with await - it blocks the process outside of the OTP code. But keep in mind that even calls with GenServer.call/3 to another process are blocking. So there are situations where it can be OK - it’s more of an issue of awareness of blocking behaviour.

On the other hand, Task.await/1 emits exit signals from within that receive block

Your channel/topic process is executing await/2 and therefore exit/1 - so it’s not a signal yet - it can still be stopped with a try..catch.

How is the supervisor of the WS going to react to those signals?

If uncaught the channel/topic process will be terminated. I’m not sure how the supervisor is configured, so I can’t predict its behaviour.

Further update:

https://github.com/phoenixframework/phoenix/blob/master/lib/phoenix/channel.ex#L407-L414

So by default the channel/topic process won’t be restarted (from the server end) - “any termination (even abnormal) is considered successful”.

https://github.com/phoenixframework/phoenix/blob/master/lib/phoenix/socket/pool_supervisor.ex#L42-L58

And if I’m reading this right the supervisor is running with default max_restarts: (3) and :max_seconds (5) i.e. if there are more than 3 restarts within 5 seconds the supervisor will terminate.

So it looks like intensely crashing channel/topic processes will in fact terminate the pool supervisor easily.

It is not recommended to await a long-running task inside an OTP behaviour such as GenServer . Instead, you should match on the message coming from a task inside your GenServer.handle_info/2 callback. For more information on the format of the message, see the documentation for async/1 .

In this case, the WS stores to DynamoDB, and publishes to Kafka. Those are fast operations in normal circumstances.

Could you folks help me understand what could happen in the Task.await/1 call in the original code?

The WS is not linked to the tasks because of the async_nolink call, but it still monitors them (the documentation of async_nolink does not say otherwise). Therefore, it receives exit signals from the task as messages. Channels have a handle_info callback, and it is not clear to me if Task.await/1 is correct in a Phoenix Channel at all because it enters a receive block to consume messages directly from the process mailbox. What do you think?

On the other hand, Task.await/1 emits exit signals from within that receive block, whose origin is the WS process itself. How is the supervisor of the WS going to react to those signals? Is this why it has been said in the thread that Task.await/1 would still need a rescue if I want the WS to not exit at all due to failures in those operations?

Just freehand but you may want to try something like this:

def task_fun(:store, event),
  do: store(event)

def task_fun(:publish, event),
  do: publish(event)

def run_tasks(event) do
  # blocks until list is reified
  # :max_concurrency defaults to System.schedulers_online/0
  MyApp.TaskSupervisor
  |> Task.Supervisor.async_stream_nolink([:store, :publish], __MODULE__, :task_fun, [event], [])
  |> Enum.to_list()
end

def handle_in(message, payload, socket) do
  event = build_event(message, payload, socket)
  # blocks until done
  run_tasks(event)
  {:reply, :ok, socket}
end

And just as reference for the topic as a whole:

Deleted a comment, I saw flaws in the questions. I’ll revise.

Your first (async/await) version was fine if that’s all that channel/topic is ever going to do - my concerns were based on the expectation that this process could accumulate more and more responsibilities over time which may not happen. I’d definitely still stick a Task.Supervisor in there and use it.

I would also hope that the actual logic behind the store and publish functions resides in (similarly named) modules separate from the channel/topic itself if only to make it easier to locate that store and publish functionality in the future (think of it as analogous to pushing functionality out of the controller into the context).

That is a consequence of using Task.await/2 to retrieve the result (which you mention).

An alternative would be to use Task.supervisor.async_stream_nolink/6 - successes are returned as :ok result tuples and crashes as :exit reason tuples. By default the stream runs as many tasks in parallel as there are schedulers online.

Hello,

If you need to await your tasks in your handle_in because you want to avoid race conditions, you must use the “async” family of task functions, to be able to call Task.await afterwise. With a “start” function you would have to send yourself a “done” message, which is ok but it’s simpler to “await”.

If you do not want to crash your socket then you must use the “nolink” family of task functions.

The Task module does not provide an async_nolink function, but Task.Supervisor does, therefore you need to use Task.Supervisor. And it comes with other benefits as José said.

So I’d say your code is fine, but note that Task.Supervisor.async_nolink/3 will still raise an exception if the task fails. So you have to handle that. (edit: actually the exception will be raised by Task.await.)

Oh and another edit: You say that « Those two operations are independent of each other, so I want to run them in parallel », but if at least one of the two operations is very fast (for example publish/1 is just sending a message and you do not need to wait for the response as your event bus handles messages order as you have told us), I would not bother with optimisation and just call the two functions from the WS process. In either case you will have to put a try/rescue block.

OK! In this approach I have the feeling of kinda emulating a mailbox by hand in the assigns. Also, the mailbox of the channel itself would get genuine messages and task messages intertwined and would need to handle :DOWN and :timeout (I think) by hand.

Oh man, need to develop fluency thinking about this, feel very clumsy!

Code organization wasn’t my primary concern.

Though I have to admit I somehow lost track of this fact:

One channel server process is created per client, per topic.

Thus by keeping topics relatively simple, complexity can be kept in check.

My main concern revolved around the long term maintainability of a single process (in this case channel/topic server) as more types of messages get added over time. Eventually the quantity of these variations and their flows will become difficult to reason about. So I would have the tendency to delegate work out of a process as it deals with more types of messages (and if there isn’t an opportunity to split out a subset of these messages and move them to an entirely separate process) so that its role could focus more and more on just handling messages in a responsive and reliable manner. It comes down to what the role of the topic is.

Also when there already is a requirement to preserve the order of events - I have to wonder whether tasks are the right tool for the job and whether this is the thin edge of the wedge towards a publishing service (which could be anything between a GenServer and an OTP application).

• You could queue pending events in the socket’s assigns.
• Wrap the two tasks in a third task which does the actual await.
• On that third task’s :DOWN message check if there are pending events to start the cycle all over again.