Rate limiting core.async channels in Clojure

Question

I'm using Clojure with core.async, and have a situation where I want to put a rate limit on the number of messages processed through a channel.

In particular I would like to:

• Define a rate limit e.g. 1,000 messages per second
• Handle messages normally (and promptly) as long as the number of messages is less than the rate limit
• Have some kind of sensible alternative handling of events if the rate limit is exceeded (e.g. telling a client to try again later)
• Have reasonably low overhead

What's the best way to achieve this?

Answer 1

Problem breakdown:

1. Define a rate limit e.g. 1,000 messages per second
2. Handle messages normally (and promptly) as long as the number of messages is less than the rate limit
3. Have some kind of sensible alternative handling of events if the rate limit is exceeded (e.g. telling a client to try again later)
4. Have reasonably low overhead

I'm approaching the problem with a solution that simply composes channels in loops.

A common rate limiting algorithm is called Token bucket. You have a fixed-size bucket of tokens and you add tokens at a fixed rate. As long as you have a token, you can send a message.

The size of the bucket determines the "burstiness" (how fast can you catch up to the maximum rate), and the rate determines the maximum average rate. These will be parameters to our code.

Let's make a channel that sends a message (doesn't matter what) at a given rate. (# 1)

(defn rate-chan [burstiness rate]
  (let [c (chan burstiness) ;; bucket size is buffer size
        delta (/ 1000 rate)]
    (go
      (while true
        (>! c :go) ;; send a token, will block if bucket is full
        (<! (timeout delta)))) ;; wait a little
    c))

Now we want a channel that limits another channel by rate. (# 2)

(defn limit-chan [in rc]
  (let [c (chan)]
    (go 
      (while true
        (<! rc) ;; wait for token
        (>! c (<! in)))) ;; pass message along
    c))

Now we can use these channels with a default if there's no message waiting:

(defn chan-with-default [in]
  (let [c (chan)]
    (go
      (while true
        ;; take from in, or if not available, pass useful message
        (>! c (alts! [in] :default :rate-exceeded))))
    c))

Now we have all of the pieces to solve the problem.

(def rchan (-> (chan)
               (limit-chan (rate-chan 100 1000))
               (chan-with-default)))

As far as #4 goes, this is not the absolute fastest solution. But it's one that uses composable parts and will probably be fast enough. If you want it faster, you could make one loop to do all of this (instead of decomposing it into smaller functions). The fastest would be to implement the interfaces yourself.

Answer 2

I wrote a little library to solve just this problem. Its implementation is eerily similar to Eric Normand's, but with some measures for high throughput channels (timeout is not precise for near-millisecond sleep times).

It also supports throttling a group of channels globablly, and function throttling.

Check it out here.