ExecutionContext with RateLimiter

Question

Suppose I have an HTTP client to call a server with a request rate limit, e.g. 1000 requests/ sec. I implemented a rate limiter in ExecutionContext like this:

Created a bounded blocking queue with RateLimiter of Guava

class MyBlockingQueue[A](capacity: Int, permitsPerSecond: Int) 
  extends ArrayBlockingQueue[A](capacity) {

  private val rateLimiter = RateLimiter.create(permitsPerSecond.toDouble)

  override def take(): A = {
    rateLimiter.acquire()
    super.take()
  }

  override def poll(timeout: Long, unit: TimeUnit): A = {
    rateLimiter.tryAcquire(timeout, unit) // todo: fix it
    super.poll(timeout, unit)
  }
}

Created an ExecutionContext from a ThreadPoolExecutor with this queue.

def createRateLimitingExecutionContext(numThreads: Int,
                                       capacity: Int,
                                       permitsPerSecond: Int): ExecutionContext = {
  val queue = new MyBlockingQueue[Runnable](capacity, permitsPerSecond)
  val executor = new ThreadPoolExecutor(numThreads, numThreads, 0L, TimeUnit.MILLISECONDS, queue)
  ExecutionContext.fromExecutor(executor)
}

Now I can create an ExecutionContext with a rate limit and pass it to the client:

implicit val ec = createRateLimitingThreadPoolExecutionContext(
  numThreads = 100,
  capacity = 1000,
  permitsPerSecond = 1000
)
httpGet("http://myserver.com/xyz") // create Futures with "ec"  

Does it make sense ? How would you test this ExecutionContext ?

Answer 1

It seems kinda ok, except the custom execution context should be explicit and/or managed inside the httpGet or its enclosing class, not a global implicit.

Because otherwise, when you write something like this for example:

    httpGet(foo)
     .recover("")
     .map(_.split(","))
     .map(_.map(_.toInt))
     .map(_.max)
     .foreach(println)

You end up consuming 6(!) permits, not one - i.e., it counts as if you have made 6 requests, which is probably not what you want.

Answer 2

For testing this custom ExecutionContext, you should be able to create a test with a behavior similar to the following:

• schedule firing a bunch of Futures for few seconds
• each Future modify an atomic value (like a counter) on which you'll be able to assert later
• regularly check the value of the atomic value: assuming you allow 10 futures/second, then checking each second you should your counter less than or equal to previous value + 10

// Pseudo-code

implicit val ec: ExecutionContext = ??? // your custom ExecutionContext allowing only 10 futures/second

val counter = new AtomicInteger()

// Fire some Futures
val start = Instant.now
val futures = (1 to 100).map(_ => Future { counter.getAndIncrement() }) )

// Check every second
(0 to 10).foreach { i =>
  counter.get() shouldBe between (i-1)*10 and (i+1)*10
  Thread.sleep(1000)
}

// Final check
Await.result(Future.sequence(futures))
val end = Instant.now
(end - start) shouldBe > 10s

This is just a rough basic idea, you can adapt for different scenarios.

Maybe a counter is too basic and you'll want more fine-grained assertions. Also here the Futures complete almost instantly, you can also simulate operations lasting longer.

Keep in mind that, as always with timing sensitive operations, you'll probably not be able to assert on specific values but assert on some value with an acceptable error margin.

Finally, you can also rely on the Guava RateLimiter to have been tested extensively. Thus you might want to consider it as a boundary of your test and only test the different interactions with it but not all the possible timing scenarios.