In a distributed system, it is often necessary to handle multiple concurrent requests for the same or similar resources. However, processing each request individually can lead to unnecessary overhead, especially if the requests are disjointed and do not benefit from sharing resources or operations. Additionally, in certain scenarios, the system may experience spikes in traffic that can overwhelm the available resources and cause performance issues.

The Batcher library aims to address these issues by providing a mechanism to batch similar requests together, reducing the overall number of operations and improving performance. The library offers a simple API for processing requests in batches. It allows developers to configure the batch size, maximum concurrency, and the duration to wait for additional requests before processing a batch.

The library can reduce the operations needed to process requests and optimize resource utilization. This ability is handy when the system receives many small, disjointed requests, such as HTTP APIs. The Batcher library provides a solution to these problems by enabling developers to improve the performance of their distributed systems while keeping the code simple and concise.

• Batching requests: The library allows you to batch together requests of the same type, improving efficiency by reducing the overhead of making individual requests.

• Concurrency control: You can specify the maximum number of concurrent requests that can be executed, which allows you to control the load on the system.

• Lingering: The library allows you to specify a duration for how long the Batcher should wait before sending a batch of requests to the server. This can help reduce the number of unnecessary requests by allowing time for other requests to be added to the batch.

• Result caching: The library caches the results of in-flight requests, so if the same request is made again, it can be returned immediately without needing to execute the request again.

Add the following dependency to your build to use the Batcher library in your project build.sbt file:

libraryDependencies += "com.filippodeluca" %%% "batcher" % "<latest-version>"

It is available in Scala-JS, Scala-Native, and Scala (2.12, 2.13, 3).

Replace latest-version with the latest version of the library available on Maven Central.

To use Batcher, you need to create an instance of the Batcher trait, which has a single method:

trait Batcher[F[_], K, V] {
  def single(key: K): F[V]
}

This method takes a key of type K and returns a F[V], where F is some effect type, like IO.

To create a Batcher instance, you can use the following method in the Batcher companion object:

def resource[F[_]: Async, K, V](
      maxConcurrency: Int,
      maxBatchSize: Int,
      linger: FiniteDuration
  )(f: IndexedSeq[K] => F[IndexedSeq[V]]): Resource[F, Batcher[F, K, V]]

This method takes three parameters:

• maxConcurrency: The maximum number of concurrent instances of f that can be run in parallel.
• maxBatchSize: The maximum number of requests to collect before calling f.
• linger: The amount of time to wait before calling f if the batch hasn't been filled yet.

The fourth parameter, f, is a function that takes an IndexedSeq[K] of keys and returns an F[IndexedSeq[V]] of results in the same order.

"The behavior of the batcher is heavily influenced by three key configuration parameters:

• maxConcurrency: This parameter determines the maximum number of parallel processes the batcher can handle simultaneously. It should be set based on the underlying system's capacity to manage concurrent operations efficiently. A good value here is usually between 16 and 500 but can hugely vary with your system capabilities.
• maxBatchSize: Specifies the maximum number of elements to include in each batch request. This value should be aligned with the capabilities of the underlying system to process batch requests efficiently. A good value here is usually in the order of thousands.
• linger: Defines the maximum duration the batcher waits before sending a batch request after receiving the last element. It should reflect the acceptable latency for the client or system utilizing the batcher. A good value here is usually between 50 and 300 milliseconds.

For example, when interfacing with AWS DynamoDB, setting maxConcurrency to the maximum allowed connections by the DynamoDB HTTP client, maxBatchSize to the DynamoDB BatchGet limit (typically 100), and linger to a value representing acceptable latency for the client can optimize the batcher's performance for such use cases."

Here's an example of how to use Batcher:

import scala.concurrent.duration.*

import cats.effect.*
import cats.syntax.all.*
import cats.effect.std.SecureRandom
import cats.effect.std.Console
import java.util.concurrent.atomic.AtomicInteger

object Example extends IOApp.Simple {

  val counter = new AtomicInteger(0)
  class SumApi {
    def batched(requests: Vector[(Int, Int)]): IO[Vector[Int]] = {
      requests
        .traverse { case (l, r) =>
          (l + r).pure[IO]
        }
        .delayBy(750.milliseconds)
    }
  }

  override def run = {

    val api = new SumApi

    val batcher = Batcher.resource[IO, (Int, Int), Int](
      maxConcurrency = 2,
      maxBatchSize = 5,
      linger = 125.milliseconds
    ) { items =>
      api.batched(items.toVector)
    }

    batcher.use { batcher =>
      SecureRandom.javaSecuritySecureRandom[IO].flatMap { random =>
        val fls = Vector.fill(100)(random.betweenInt(0, 10)).sequence
        val frs = Vector.fill(100)(random.betweenInt(0, 10)).sequence
        (fls, frs)
          .mapN { (ls, rs) =>
            ls.zip(rs)
          }
          .flatMap { pairs =>
            pairs
              .parTraverse_ { pair =>
                batcher.single(pair).flatMap { result =>
                  IO.println(s"${pair._1} + ${pair._2} = $result")
                }
              }
          }
      }
    }
  }
}

This example first generates two vectors of 100 random integers. Then, it pairs up the corresponding elements of both vectors using the zip method and passes them to the Batcher.apply method, which sums them up. Finally, it prints the result.

There is also an integration test based on DynamoDb, based on a real-world use case, that provides a practical demonstration of how to utilize the AWS SDK v2 to efficiently handle GetItem and PutItem requests together, which can be particularly useful in real-world scenarios with high data volumes. Utilizing DynamoDb as the underlying storage engine, the test also highlights the benefits of leveraging cloud-based services for scalable and performant data processing.