I am currently attempting to design a set of modules which uses batching under the hood while using a single ID function definition. In concrete terms, I would have a function definition like:
batcher = SomeBatcher.new()
{:ok, res} = SomeBatcher.load(batcher, "1")
Repeated instances of this call would batch them in groups and run some arbitrary code against said group. It’s some what similar to dataloaders in that respect. So far, I’ve go this working:
defmodule Batcher do
@moduledoc false
use GenServer
defmodule State do
defstruct [:batch_fn, :max_batch_count, :timeout, :timer, pending: %{}]
end
def start_link([batch_fn, max_batch_count]) do
start_link(batch_fn, max_batch_count)
end
def start_link([batch_fn, max_batch_count, timeout]) do
start_link(batch_fn, max_batch_count, timeout)
end
def start_link(batch_fn, max_batch_count, timeout \\ 1_000) do
GenServer.start_link(
__MODULE__,
{:ok, %State{batch_fn: batch_fn, max_batch_count: max_batch_count, timeout: timeout}}
)
end
def init({:ok, state}) do
{:ok, state}
end
def load(batcher_pid, id) do
GenServer.call(batcher_pid, {:add, id})
end
def handle_call({:add, id}, from, %State{pending: pending} = state) do
pending = Map.put(pending, id, from)
state = %{state | pending: pending}
IO.inspect("adding #{id} into pending stack")
{:noreply, schedule_timeout(state), {:continue, :flush_on_capacity}}
end
defp schedule_timeout(%State{timer: timer_ref} = state) do
case timer_ref do
nil ->
%{state | timer: create_timer(state)}
existing_timer ->
Process.cancel_timer(existing_timer)
%{state | timer: create_timer(state)}
end
end
def handle_continue(
:flush_on_capacity,
%State{pending: pending, max_batch_count: max_size} = state
)
when map_size(pending) >= max_size do
IO.inspect("flushing from max capacity")
{:noreply, flush(state)}
end
def handle_continue(:flush_on_capacity, state), do: {:noreply, state}
defp create_timer(%State{timeout: timeout}), do: Process.send_after(self(), :timeout, timeout)
def handle_info(:timeout, state) do
IO.inspect("flushing from timeout")
{:noreply, flush(state)}
end
defp flush(%State{batch_fn: batch_fn, pending: pending, timer: timer_ref} = state) do
IO.inspect("flushing jobs: #{inspect(pending)}")
res =
pending
|> Map.keys()
|> batch_fn.()
|> Enum.into(%{})
for {id, from} <- pending do
GenServer.reply(from, {:ok, res[id]})
end
Process.cancel_timer(timer_ref)
%{state | pending: %{}}
end
end
I haven’t used GenServer.reply/2 before, but reading the documentation seems to make it clear enough. The actual question I have is:
- Is it possible due to how I am wiping the pending state at
flush/1that I could remove pending jobs that have not been actioned?
Let’s say I have the following configuration on the batcher:
- 5 maximum capacity
- Timeout of 1 second
With this configured batcher, I send 7 load/2 calls, where the first argument is the batcher process and the second some string. I would assume that the first 5 that run are queued. Once the threshold is met, it will execute the batch function on said 5. What I don’t understand is the possibility of the sixth or seventh load/2 call being executed at the same, which may be lost when the pending map is reset. My tests don’t seem to indicate that this happens (thankfully), which leads me to believe that regardless of handle_continue and the like that the Erlang process is running each task one by one, which I think it does.
Does this assumption line up within other people’s expectations? I think it might based off my understanding of the concurrency model, but if there are more nuances I am overlooking, I would welcome any feedback.
