DEV Community: rafaquelhodev

Controlling outgoing rate limit

rafaquelhodev — Sun, 28 Jul 2024 21:17:36 +0000

Let's imagine a scenario that one has a distributed application that interacts with a third party API. Usually, third party APIs have a rate-limit control mechanism in order to avoid their clients from bursting requests and causing down-time on their services. In such a scenario, how can the caller control the rate of outgoing requests to the third party API in a distributed environment? This post discuss a possible strategy for this problem.

There a multiple algorithms to control the rate of requests, but here we'll focus on the token bucket algorithm, because it is relatively easy to understand and to implement. This algorithm states that: a bucket can hold a maximum of T tokens, and when an application wants to make a request to the third party API, it has to take 1 token from the bucket. If the bucket is empty, it has to wait until there is a least 1 token in the bucket. Also, the bucket is refilled with 1 token at a fixed rate of R tokens/milliseconds.

The token bucket algorithm is very straightforward to understand, but how can someone use it in a distributed environment to control the outgoing request to third party APIs?

If one wants to control the outgoing rate limit in a distributed environment, a centralized source of truth for the current rate limit is necessary. There are multiple ways to implement the source of truth and I've idealized the following diagram with a possible implementation:

In the figure above, we have a distributed application in multiple pods, and each pod can make requests to a third party API. In the application infrastructure, there is a TCP server that controls the rate limit by using the token bucket algorithm. Before making a request to the third party API, the pod asks the TCP server for a new token, and the pod waits for a response from the TCP server until there is at least one available token. After a token is available, the pod makes the request to the third party API.

The TCP server implementation can be found in this repository https://github.com/rafaquelhodev/rlimit/ and in the next section I'll discuss briefly the token bucket implementation in golang.

Token bucket implementation

Below, I'm showing the main ideas behind the token bucket implementation. Please, take a look at the https://github.com/rafaquelhodev/rlimit/ repository to understand the detailed implementation.

The rate limit control is centralized in the TokenBucket struct:

type TokenBucket struct {
    id           string
    mu           sync.Mutex
    tokens       int64
    maxTokens    int64
    refillPeriod int64
    cron         chan bool
    subs         []chan bool
}

You can notice that there is a subs property in the TokenBucket struct. Basically, this is an array of subscribers for a specific token bucket: every time a token is requested from a client, the client is added to the subs array and the client is notified when a new token is added to the bucket.

When starting the bucket, we need to provide a maximum number of tokens the bucket can support (maxTokens) and the amount of time a token is added to the bucket (refillPeriod):

func newTokenBucket(id string, maxTokens int64, refillPeriod int64) *TokenBucket {
    bucket := &TokenBucket{
        id:           id,
        tokens:       0,
        maxTokens:    maxTokens,
        refillPeriod: refillPeriod,
        cron:         make(chan bool),
        subs:         make([]chan bool, 0),
    }
    fmt.Printf("refill period  = %d\n", refillPeriod)
    bucket.startCron()
    return bucket
}

Now, you might wonder, "how a token is added to bucket?". For that, when a bucket is created, a cron job is started, and at every refillPeriod milliseconds, a new token is added to the bucket:

func (tb *TokenBucket) startCron() {
    ticker := time.NewTicker(time.Duration(tb.refillPeriod) * time.Millisecond)

    go func() {
        for {
            select {
            case <-tb.cron:
                ticker.Stop()
                return
            case <-ticker.C:
                if tb.tokens < tb.maxTokens {
                    tb.tokens += 1
                    fmt.Printf("[TOKEN REFIL] | currTokens = %d\n", tb.tokens)

                    if len(tb.subs) > 0 {
                        sub := tb.subs[0]
                        tb.subs = tb.subs[1:]
                        sub <- true
                    }
                }
            }
        }
    }()
}

Finally, when a client wants a token from the bucket, the waitAvailable function must be called:

func (tb *TokenBucket) waitAvailable() bool {
    tb.mu.Lock()

    if tb.tokens > 0 {
        fmt.Printf("[CONSUMING TOKEN] - id = %s\n", tb.id)
        tb.tokens -= 1
        tb.mu.Unlock()
        return true
    }

    fmt.Printf("[WAITING TOKEN] - id %s\n", tb.id)

    ch := tb.tokenSubscribe()

    tb.mu.Unlock()

    <-ch

    fmt.Printf("[NEW TOKEN AVAILABLED] - id %s\n", tb.id)

    tb.tokens -= 1

    return true
}

Inspired by https://github.com/Mohamed-khattab/Token-bucket-rate-limiter

Contributing to a Go open source repository

rafaquelhodev — Mon, 04 Apr 2022 11:56:46 +0000

In this post we are going to list the first steps to contribute to a Go lang open source project.

In this example, we'll simulate a contribution to Bubbletea repository.

1) Fork the project

The first step is to fork the project you want to contribute. Search the project on GitHub and click on Fork: you'll get your own version of the repository <your_name>/bubbletea.

2) Clone the fork project into your computer

Clone the forked version of bubbletea into your computer. For the sake of this example, let's say you've cloned the repository at /home/forked_bubbletea on your computer.

3) Use your local forked repository

We'll create a new folder at /home/test_bubble. Create a file main.go with a simple example:

package main

import (
    "fmt"
    "os"
    tea "github.com/charmbracelet/bubbletea"
)

.
.
.

func main() {
    p := tea.NewProgram(initialModel())
    if err := p.Start(); err != nil {
        fmt.Printf("Alas, there's been an error: %v", err)
        os.Exit(1)
    }
}

Create a go.mod file by doing:

go mod init test_bubble

Then, we need to replace the library we want to contribute by our own version of the library. For this, we need to append this line at the end of the go.mod file:

...
replace github.com/charmbracelet/bubbletea v0.20.0 => /home/forked_bubbletea

By doing this, when we call in main.go the library bubbletea, we are going to call our local version of the library. Therefore, we can change our local version of the library and instantly see the effects in our code.

4) Raise a pull request

Finally, after we changed and committed our local version of the repository, all we need to do is raise a pull request to the repository we want to contribute. For this, go to the repository you want to contribute and click Compare & pull request.

Adding periodic task in Elixir with OTP

rafaquelhodev — Thu, 31 Mar 2022 13:01:44 +0000

In this post, I'll show how to create a simple periodic task using Elixir OTP.

In this example, we'll create a simple server calculator. For somehow, we want to keep adding numbers periodically over time. The initial structure for this process is:

defmodule Calculator do
  use GenServer

  @doc """
  Starts the calculator.
  """
  def start_link(opts) do
    GenServer.start_link(__MODULE__, :ok, opts)
  end

  @impl true
  def init(:ok) do
    IO.puts("starting calculator")
    {:ok, 0}
  end

  @impl true
  def handle_call({:add, num}, _from, state) do
    result = state + num
    IO.puts(result)
    {:reply, {:ok, result}, result}
  end

  def add(server, num) do
    {:ok, result} = GenServer.call(server, {:add, num})
    result
  end
end

Here, we can add a number by sending a message {:add, num} to the server. The client function for this purpose is add/2.

Then, we need to create a new process that will handle the periodic tasks.

defmodule Periodic do
  use GenServer

  defstruct freq: nil, server: nil, callback: nil

  def start_link(opts) do
    GenServer.start_link(__MODULE__, :ok, opts)
  end

  @impl true
  def init(:ok) do
    {:ok, %{}}
  end

  @impl true
  def handle_call({:periodic, data}, _from, state) do
    Process.send_after(data.server, {:tick, data}, data.freq)

    {:reply, data, state}
  end

  @impl true
  def handle_info({:tick, data}, state) do
    data.callback.()

    Process.send_after(data.server, {:tick, data}, data.freq)

    {:noreply, state}
  end

  def retry(server, data) do
    data = Map.put(data, :server, server)
    GenServer.call(data.server, {:periodic, data})
  end
end

The periodic task is created using the Process.send_after/3 function. By doing this, a message {:tick, data} is sent to the server after a time of data.freq ms.

The idea behind this server is that the client must provide a callback function that is run periodically in a frequency (in ms) also defined by the client. Therefore, the client must provide two inputs: callback and freq.

In order to run a periodic task, a client must call the retry/2 function, passing the callback and freq information.

Finally, we need to define the periodic task that must be run the in Calculator server. Remember, we want to keep adding numbers periodically. For this, we define a new function add_periodic and a constant freq (set to be 1000 ms) at the Calculator module:

defmodule Calculator do
  ...
  alias Periodic

  @freq 1000

  def add_periodic(server, num, freq \\ @freq) do
    {:ok, pid} = Periodic.start_link([])

    callback = fn -> add(server, num) end

    Periodic.retry(pid, %Periodic{freq: freq, callback: callback})
  end

The entire code can be found HERE .

Usage example:

First open the iex shell by:

iex -S mix

Then,

import Calculator

{:ok, pid} = Calculator.start_link([])

Calculator.add_periodic(pid, 2)

And voila, we'll see something like this:

2
4
6
8
...