defmodule Ark.Drip do use GenServer def start_link(args, opts) do args = read_args(args) GenServer.start_link(__MODULE__, args, opts) end def start(args, opts) do args = read_args(args ) GenServer.start(__MODULE__, args, opts) end defp read_args({max_drips, range_ms}), do: [ max_drips: max_drips, range_ms: range_ms ] defp read_args(args), do: args def stop(bucket) do GenServer.stop(bucket) end def await(bucket, timeout \\ :infinity) def await(bucket, timeout) do GenServer.call(bucket, :await, timeout) end def init(args) do max_drips = Keyword.fetch!(args, :max_drips) range_ms = Keyword.fetch!(args, :range_ms) if max_drips < 1, do: raise "Minimum drip per period is 1, got: #{inspect max_drips}" if range_ms < 1, do: raise "Minimum drip per period is 1, got: #{inspect range_ms}" # Small optimisation : if we can lower the max_drips to `1` by dividing # the range by the given max_drips, we can skip all time calculations # But this is not equivalent because if you can do 10 calls in 1 second, # then you can actually make your 10 calls at the beginning of the second. # If we divide, we will do each call after 1/10 second and this will take # the whole second. state = %{ last_calls: [], max_drips: max_drips, range_ms: range_ms, } {:ok, state} end # We receive a drip demand. We satisfy it immediately, because we have waited # beforehand. # After replying, we count the time of the request and wait if needed before # being authorized to satisfy another demand def handle_call(:await, from, state) do GenServer.reply(from, :ok) {:noreply, count_drip(state)} end # count_drip # # We take the last demand in account : First, as we can handle N demands in T # timespan, we keep track of the times of the demands. So we discard the # oldest one. Then we push the new one on top of the stack, and calculate the # time we have to wait before satisfying another demand. # # If we have not reached the max calls yet (list length < max_drips - 1), we # can just say that it's ok, no need to wait. # Notice: We use (max_drips - 1) because we wait before *new* accepting calls. # So the call that happened right before is counted and we wait for a *future* # call. Example, if max_drips is 2, we want to enter the function clause that # does not wait only when there is only 1 last_call registered, so when the # last calls lenght is < (2 - 1), i.e. zero. If the list has already a value, # the call that triggered count_drip must trigger a wait and will be added # to the list to calculate wait. So after a call, a list of length # (max_drips - 1) plus the last call hits the limit. # # But first, edge case with only 1 drip per timespan. defp count_drip(%{max_drips: 1, range_ms: range_ms} = state) do Process.sleep(range_ms) # tracking last call is useless when max_drips is 1. But we may want to # change max_drips amount during runtime ? %{state | last_calls: [now()]} end defp count_drip(%{last_calls: last_calls, max_drips: max_drips} = state) when length(last_calls) < max_drips - 1 do %{state | last_calls: [now() | last_calls]} end defp count_drip(state) do %{last_calls: last_calls, max_drips: max_drips, range_ms: range_ms} = state # Push the last drip onto the stack and discard the old one last = now() last_calls = [last | :lists.sublist(last_calls, max_drips - 1)] last_call_in_range = List.last(last_calls) # Calculate the time at which a new call is authorized. Since our call list # is full, we just calculate the time after which the last remaining call # in our list goes out of range. next_call_at = last_call_in_range + range_ms # Now, sleep just the amount of time required before another call. # Process.sleep will not accept negative values wait = (next_call_at - last) |> max(0) Process.sleep(wait) %{state | last_calls: last_calls} end def now() do :erlang.monotonic_time(:millisecond) # :erlang.system_time(:millisecond) end end