Beginner OTP/GenServer design question

Questions & Help Questions
,
3

When you say SYNCHRONOUSLY is it that you want to

  1. Just ensure errors are propagated
  2. Never execute two batches at the same time
  3. Return a value from executing the batch?

Ensure errors are handled (case 1)

Here we’re just using spawn_link() to run the work, so the batch processor is not blocked, but if something goes wrong with the batch the processor is killed too. Here it might happen that multiple batches are run in parallel at the same time.

def handle_call({:new_task, task}, _from, state) do
  current_batch = Enum.find(state, fn {k, v} -> length(v) < 5 end)

  case current_batch do
    nil ->
      batch_id = gen_random_id()
      Process.send_after(self(), {:batch_timeout, batch_id}, 10 * 1000)

     {:reply, :ok, Map.put(state, batch_id, [task]}

    {batch_id, tasks} when length(tasks) == 4 -> 
      # SYNCHRONOUSLY send batched API request
      state = Map.update!(state, batch_id, & [task | &1])
      {:noreply, state} = handle_info({:batch_timeout, batch_id}, state)
      {:reply, :ok, state}
    
    {batch_id, tasks} -> 
      {:reply, :ok, Map.update!(state, batch_id, & [task | &1])}
  end
end

def handle_info({:batch_timeout, batch_id}, state) do
  case Map.pop(state, batch_id) do
    {nil, _state} -> {:noreply, state}
    {batch, state} -> 
      spawn_link(fn -> work(batch) end)
      {:noreply, state}
  end
end

Never execute two batches at the same time (case 2)

For this I’m moving the start of the timer into the caller and change from a GenServer.call() to a GenServer.cast() – So now the processing is after each other and never blocks. But as a result the server might get overloaded when batch processing is slower than incoming batches…

def new_task(task) do
  batch_id = gen_random_id()
  Process.send_after(__MODULE__, {:batch_timeout, batch_id}, 10 * 1000)
  GenServer.cast(__MODULE__, {:new_task, task, batch_id}})
end

def handle_cast({:new_task, task, batch_id}, _from, state) do
  current_batch = Enum.find(state, fn {k, v} -> length(v) < 5 end)

  case current_batch do
    nil ->
     {:noreply, Map.put(state, batch_id, [task]}

    {batch_id, tasks} when length(tasks) == 4 -> 
      state = Map.update!(state, batch_id, & [task | &1])
      handle_info({:batch_timeout, batch_id}, state)

    {batch_id, tasks} -> 
      {:noreply, Map.update!(state, batch_id, & [task | &1])}
  end
end

def handle_info({:batch_timeout, batch_id}, state) do
  case Map.pop(state, batch_id) do
    {nil, _state} -> {:noreply, state}
    {batch, state} -> 
      # SYNCHRONOUSLY send batched API request
      work(...)
      {:noreply, state}
  end
end

Return a value from the batch (case 3)

In case you want to return a value from the batch processing / task processing back to the one who submitted the task you can combine that either with case 1 or two 2. Just in addition you have to use the GenServer.call() with {:noreply, …} and GenServer.reply() to reply after the batch submission is done. Left as an exercise for the reader :wink: