How do I make a GenServer that processes messages at a specific rate? (every n seconds)

Question

One of my services talks to an external API that is rate-limited, so I want to ensure that I send no more than 1 call every 10 seconds.

My naive approach would be to have a long-running API service, and time it out after each call:

def handle_cast({:call_api, data}, state) do
  send_to_external_api(data)
  :timer.sleep(10000)
  {:noreply, state}
end

I'm not sure if there's a proper way to do this.

You can also take advantage of `Process.send_after`. Take a look at this answer: http://stackoverflow.com/questions/32085258/how-to-run-some-code-every-few-hours-in-phoenix-framework — edmz, Sep 21 '16 at 23:46
@edmz hi, so this isn't really a periodic task in that it's a task that executes every 10 seconds. But rather it's a process that can be called by other processes, but which must execute *at most* in 10 second intervals. -- The purpose is solely to limit API calls to below a threshold. — o_o_o--, Sep 22 '16 at 00:09
Since this is `handle_cast`, this approach looks good to me. You might want to subtract the time taken for `send_to_external_api(data)` from 10000 if you want to send "a maximum of 1 request per 10 seconds" and not "10 second gap between requests" — Dogbert, Sep 22 '16 at 08:43
Do you want to drop other requests in those 10sec or you want to queue them so they can be sent when 10sec period is elapsed? — BurmajaM, Sep 22 '16 at 09:08

score 5 · Accepted Answer · edited May 23 '17 at 12:00

Edit: The original solution dropped messages in between 10s ticks, as burmajam suggested. The edit provides a more appropriate solution.

EDIT

Due to the fact that GenServer handle_* functions do not actually receive messages from the queue, but just process them, we cannot exploit pattern matching to receive selectively only each 10s from the processes queue.

Therefore, since we're picking up messages in order of their arrival, we need out internal queue as part of GenServer's state.

defmodule RateLimited do
  use GenServer

  def start_link do
    GenServer.start_link(__MODULE__, %{queue: []})
  end

  def init(state) do
    allow_work()
    {:ok, state}
  end

  def handle_cast({:call_api, data}, %{"queue" => queue} = state) do
    {:noreply, %{state | queue: queue ++ [data]}}
  end

  def handle_info(:work, %{"queue" => [data | queue]} = state) do
      send_to_external_api(data)
    allow_work()

    {:noreply, %{state | queue: queue}}
  end

  defp allow_work() do
    Process.send_after(self(), :work, 10000) # after 10s
  end

  defp send_to_external_api (data) do end
end

So we're just moving messages from process queue to state queue and we process the head when we signal ourselves that 10s have passed.

But in the end, we actually achieve the same result as if we put the process to sleep for 10 seconds. Your solution seems easier and achieves the same result.

The solution is based on How to run some code every few hours in Phoenix framework?

First, let your GenServer store a flag in its state (work = true/false).

Then let GenServer use Process.send_after to signal itself when it can work. You receive the signal in handle_info where you set the work state flag to true.

Now notice pattern matching on state in the handle_cast function: it will only pick up the message when work state flag equals true. Otherwise, messages will be put in the queue waiting.

And after you send the message to external service, you run Process.send_after again to schedule the next signal and return state which has work flag set to false to prevent next messages from being immediately picked up.

defmodule RateLimited do
  use GenServer

  def start_link do
    GenServer.start_link(__MODULE__, %{work: false})
  end

  def init(state) do
    allow_work()
    {:ok, state}
  end

  def handle_cast({:call_api, data}, %{"work" => true} = state) do
    send_to_external_api(data)
    allow_work()
    {:noreply, %{state | work = false}}
  end

  def handle_info(:work, state) do
    {:noreply, %{state | work = true}}
  end

  defp allow_work() do
    Process.send_after(self(), :work, 10000) # after 10s
  end
end

This is good solution if he wants to drop requests in those 10sec. However you need handle_cast when work is false :) — BurmajaM, Sep 23 '16 at 08:36
you Sir, are correct :) I've edited my answer to include a more appropriate solution. — usoban, Sep 25 '16 at 20:40
Marked this as correct because my solution, while simpler, has the problem that the caller has to wait until the timeout is over to receive the {:noreply} tuple, which is problematic for long timeouts since it blocks. — o_o_o--, Sep 27 '16 at 20:00

How do I make a GenServer that processes messages at a specific rate? (every n seconds)

1 Answers1