A high-performance, back-pressured, generic concurrent queue implementation.

This is the pure and generic version of monix.execution.AsyncQueue.

Example

import cats.implicits._
import cats.effect._
import monix.execution.Scheduler.global

// For being able to do IO.start
implicit val cs: ContextShift[IO] = SchedulerEffect.contextShift[IO](global)(IO.ioEffect)
// We need a `Timer` for this to work
implicit val timer: Timer[IO] = SchedulerEffect.timer[IO](global)

def consumer(queue: ConcurrentQueue[IO, Int], index: Int): IO[Unit] =
  queue.poll.flatMap { a =>
    println(s"Worker $$index: $$a")
    consumer(queue, index)
  }

for {
  queue     <- ConcurrentQueue[IO].bounded[Int](capacity = 32)
  consumer1 <- consumer(queue, 1).start
  consumer2 <- consumer(queue, 1).start
  // Pushing some samples
  _         <- queue.offer(1)
  _         <- queue.offer(2)
  _         <- queue.offer(3)
  // Stopping the consumer loops
  _         <- consumer1.cancel
  _         <- consumer2.cancel
} yield ()

Back-Pressuring and the Polling Model

The initialized queue can be limited to a maximum buffer size, a size that could be rounded to a power of 2, so you can't rely on it to be precise. Such a bounded queue can be initialized via ConcurrentQueue.bounded. Also see BufferCapacity, the configuration parameter that can be passed in the ConcurrentQueue.withConfig builder.

On offer, when the queue is full, the implementation back-pressures until the queue has room again in its internal buffer, the task being completed when the value was pushed successfully. Similarly poll awaits the queue to have items in it. This works for both bounded and unbounded queues.

For both offer and poll, in case awaiting a result happens, the implementation does so asynchronously, without any threads being blocked.

Multi-threading Scenario

This queue supports a ChannelType configuration, for fine tuning depending on the needed multi-threading scenario. And this can yield better performance:

• MPMC: multi-producer, multi-consumer
• MPSC: multi-producer, single-consumer
• SPMC: single-producer, multi-consumer
• SPSC: single-producer, single-consumer

The default is MPMC, because that's the safest scenario.

import monix.execution.ChannelType.MPSC
import monix.execution.BufferCapacity.Bounded

val queue = ConcurrentQueue[IO].withConfig[Int](
  capacity = Bounded(128),
  channelType = MPSC
)

WARNING: default is MPMC, however any other scenario implies a relaxation of the internal synchronization between threads.

This means that using the wrong scenario can lead to severe concurrency bugs. If you're not sure what multi-threading scenario you have, then just stick with the default MPMC.