Exploring rate limiting techniques in Clojure and Redis Lua
In this post I show how to implement a concurrent-safe and performant rate limiter in both Clojure and Redis using the token bucket algorithm.
An instinctive approach first
Let’s imagine you want to prevent excessive use of your API, either globally or per client: you could count the incoming requests and reject them once a certain threshold is reached, that counter state could be maintained over time windows, using fixed intervals, or sliding ones:
Both can be implemented by counting requests over time intervals, we can take the example of a rate limiting of 10 requests per minute.
We simply have to count over fixed periods of time, a naive implementation (but concurrent-safe) of this could be:
(def max-hits 10)
;; a state of the count of requests by minute
(def counters* (atom {}))
(defn update-counters [counters minute]
;; keep only the minute in the counters map
{minute (-> (get counters minute)
(or 0)
inc
(min max-hits))})
(defn throttle!
"returns true if the request should be throttled"
[counters* timestamp]
(let [minute (int (/ timestamp (* 60 1000)))
[old new] (swap-vals! counters* #(update-counters % minute))]
;; if the atom is unchanged, it means the hit has been rejected
(= old new)))
(comment
(def now (System/currentTimeMillis))
@counters*
(dotimes [_ 10] (throttle! counters* now))
(throttle! counters* now)
;;=> true
(throttle! counters* (+ now (* 60000)))
;;=> false
)Note that it is not a very rigorous approach, the client can make 10 requests at :59, and 10 requests just 1 second after since the counter is on a new minute.
For a more correct implementation we can maintain a queue of events timestamps (at the cost of more memory). We can represent each request as a timestamp, and append them to the rear of a queue, while popping the old requests at the front. If there are still more elements than max-hits after the popping, the request should be rejected:
;; 10 hits max every 10 seconds
(def max-hits 10)
(def interval-ms (* 10 1000))
;; we store requests ids with timestamps in an ordered queue
;; of [{:id ? :timestamp ?}]
(def requests-queue* (atom (PersistentQueue/EMPTY)))
(defn update-queue [q request]
(let [q-without-old
(loop [q q]
(if-let [{:keys [timestamp]} (first q)]
(if (<= timestamp (- (:timestamp request) interval-ms))
(recur (pop q))
q)
q))]
(if (>= (count q-without-old) max-hits)
q-without-old
(conj q-without-old request))))
(defn throttle!
"returns true if the request should be throttled"
[requests-queue* request-timestamp]
(let [request-id (UUID/randomUUID)
new-queue (swap! requests-queue*
#(update-queue % {:id request-id
:timestamp request-timestamp}))]
;; if the last value of the atom does not contain the request id,
;; it should be throttled.
;; Note that this "last" call is O(n),
;; better use a double-ended queue instead
(not= request-id (:id (last new-queue)))))
(comment
(def now (System/currentTimeMillis))
(dotimes [_ 10] (throttle! requests-queue* now))
(throttle! requests-queue* now)
;;=> true
(throttle! requests-queue* (+ now (* 10000)))
;;=> false
)Instead of counting what happened in the past to decide if the current request should be throttled, we can use a more efficient technique both in time complexity and memory space.
The token bucket algorithm
We can consider the "rate" of the rate limit as a speed at which tokens (or credits) are released, each token allowing one hit.
For instance if we want to limit the bandwidth to 60 reqs/minute, after 30s we would have 30 new tokens available. For each incoming request, if there is a token available, we burn it and let the request pass, if not, we reject the request.
We can implement that with a Clojure atom again, swapping it in one step to avoid race conditions
;; the rate limit time window (10s)
(def interval-ms (* 10 1000))
;; how many hits can be made during that interval at max
(def max-hits 10)
;; the token bucket state
(def token-bucket* (atom {;; the last time the bucket has been updated
:last-checked-ts (System/currentTimeMillis)
;; the tokens available at that time
:tokens-count max-hits}))
(defn update-bucket [bucket request-timestamp]
(let [{:keys [last-checked-ts tokens-count]} bucket
elapsed-ms (- request-timestamp last-checked-ts)
tokens-to-add (/ (float (* max-hits elapsed-ms)) interval-ms)
new-tokens-count (dec (+ tokens-count tokens-to-add))]
(if (pos? new-tokens-count)
{:last-checked-ts request-timestamp
:tokens-count (min new-tokens-count
(dec max-hits))}
;; if no token is available in the bucket, leave the atom unchanged
bucket)))
(defn throttle! [token-bucket* request-timestamp]
(let [[old new] (swap-vals! token-bucket* #(update-bucket % request-timestamp))]
;; if the atom is unchanged, it means the hit has been rejected
(= old new)))
(comment
(def now (System/currentTimeMillis))
(dotimes [_ 10] (throttle! token-bucket* now))
(throttle! token-bucket* now)
;=> true
(throttle! token-bucket* (+ now 10000))
;=> false
)Going remote with Redis and Lua
If the requests handlers are distributed, we can centralize the token bucket state on Redis instead. A Lua script provides the equivalent semantics as the Clojure atom (the Lua script is also blocking, but guarantees the order of execution) so we avoid race conditions. The algorithm stays the same.
We use two Redis keys in the Lua script:
- KEYS[1] = rate-limiter:last-checked-ts
- KEYS[2] = rate-limiter:tokens-count
And three arguments
- ARGV[1] = max-hits
- ARGV[2] = interval-ms
- ARGV[3] = request-timestamp
local last_checked_ts = tonumber(redis.call('get', KEYS[1]));
local tokens_count = tonumber(redis.call('get', KEYS[2]));
local max_hits = tonumber(ARGV[1]);
if last_checked_ts then
local elapsed_ms = tonumber(ARGV[3]) - last_checked_ts;
local tokens_to_add = math.floor((max_hits * elapsed_ms / tonumber(ARGV[2])) + 0.5);
local new_tokens_count = math.min(tokens_count + tokens_to_add - 1, max_hits - 1);
redis.log(redis.LOG_WARNING, new_tokens_count)
if new_tokens_count >= 0 then
redis.call('set', KEYS[2], new_tokens_count);
redis.call('set', KEYS[1], ARGV[3]);
end
return new_tokens_count < 0;
else
redis.call('set', KEYS[2], max_hits - 1);
redis.call('set', KEYS[1], ARGV[3]);
return false;
endUsing the Clojure redis client Carmine:
;; 10 hits max every 10 seconds
(def max-hits 10)
(def interval-ms (* 10 1000))
(defn throttle! [redis-conn request-timestamp]
(= 1
(car/wcar redis-conn
(car/eval*
(str "
local last_checked_ts = tonumber(redis.call('get', KEYS[1]));
local tokens_count = tonumber(redis.call('get', KEYS[2]));
local max_hits = tonumber(ARGV[1]);
if last_checked_ts then
local elapsed_ms = tonumber(ARGV[3]) - last_checked_ts;
local tokens_to_add = math.floor((max_hits * elapsed_ms / tonumber(ARGV[2])) + 0.5);
local new_tokens_count = math.min(tokens_count + tokens_to_add - 1, max_hits - 1);
redis.log(redis.LOG_WARNING, new_tokens_count)
if new_tokens_count >= 0 then
redis.call('set', KEYS[2], new_tokens_count);
redis.call('set', KEYS[1], ARGV[3]);
end
return new_tokens_count < 0;
else
redis.call('set', KEYS[2], max_hits - 1);
redis.call('set', KEYS[1], ARGV[3]);
return false;
end")
2
"rate-limiter:last-checked-ts"
"rate-limiter:tokens-count"
max-hits
interval-ms
request-timestamp))))
(comment
(def now (System/currentTimeMillis))
(dotimes [_ 10] (throttle! {} now))
(throttle! {} now)
;=> true
(throttle! {} (+ now 10000))
;=> false
(car/wcar {} (car/get "rate-limiter:last-checked-ts") (car/get "rate-limiter:tokens-count"))
)You can find an implementation of this algorithm on top of core.async channels in this library: https://github.com/brunoV/throttler