ConcurrentChannel can be used to model complex producer-consumer communication channels.

It exposes these fundamental operations:

• push for pushing single events to consumers (producer side)
• pushMany for pushing event sequences to consumers (producer side)
• halt for pushing the final completion event to all consumers (producer side)
• consume for creating a ConsumerF value that can consume the incoming events from the channel

Example

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

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

// Completion event
sealed trait Complete
object Complete extends Complete

def logLines(consumer: ConsumerF[IO, Complete, String], index: Int): IO[Unit] =
  consumer.pull.flatMap {
    case Right(message) =>
      IO(println("Worker $$index: $$message"))
        // continue loop
        .flatMap(_ => logLines(consumer, index))
    case Left(Complete) =>
      IO(println("Worker $$index is done!"))
  }

for {
  channel <- ConcurrentChannel[IO].of[Complete, String]
  // Workers 1 & 2, sharing the load between them
  task_1_2 = channel.consume.use { ref =>
    (logLines(ref, 1), logLines(ref, 2)).parSequence_
  }
  consumers_1_2 <- task_1_2.start // fiber
  // Workers 3 & 4, receiving the same events as workers 1 & 2,
  // but sharing the load between them
  task_3_4 = channel.consume.use { ref =>
    (logLines(ref, 3), logLines(ref, 4)).parSequence_
  }
  consumers_3_4 <- task_3_4.start // fiber
  // Pushing some samples
  _ <- channel.push("Hello, ")
  _ <- channel.push("World!")
  // Signal there are no more events
  _ <- channel.halt(Complete)
  // Await for the completion of the consumers
  _ <- consumers_1_2.join
  _ <- consumers_3_4.join
} yield ()

Unicasting vs Broadcasting vs Multicasting

Unicasting: A communication channel between one producer and one ConsumerF. Multiple workers can share the load of processing incoming events. For example in case we want to have 8 workers running in parallel, you can create one ConsumerF, via consume and then use it for multiple workers. Internally this setup uses a single queue and whatever workers you have will share it.

Broadcasting: the same events can be sent to multiple consumers, thus duplicating the load, as a broadcasting setup can be created by creating and consuming from multiple ConsumerF via multiple calls to consume. Internally each ConsumerF gets its own queue and hence messages are duplicated.

Multicasting: multiple producers can push events at the same time, provided the channel's type is configured as a MultiProducer.

Back-Pressuring and the Polling Model

When consumers get created via consume, a buffer gets created and assigned per consumer.

Depending on what the BufferCapacity is configured to be, the initialized consumer can work with 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. See consumeWithConfig for customizing this buffer on a per-consumer basis, or the ConcurrentChannel.withConfig builder for setting the default used in consume.

On push, when the queue is full, the implementation back-pressures until the channel has room again in its internal buffer(s), the task being completed when the value was pushed successfully. Similarly ConsumerF.pull (returned by consume) awaits the channel to have items in it. This works for both bounded and unbounded channels.

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

Multi-threading Scenario

This channel supports the fine-tuning of the concurrency scenario via ChannelType.ProducerSide (see ConcurrentChannel.withConfig) and the ChannelType.ConsumerSide that can be specified per consumer (see consumeWithConfig).

The default is set to MultiProducer and MultiConsumer, which is always the safe choice, however these can be customized for better performance.

These scenarios are available:

• MPMC: multi-producer, multi-consumer, when MultiProducer is selected on the channel's creation and MultiConsumer is selected when consuming; this is the safe scenario and should be used as the default, especially when in doubt
• MPSC: multi-producer, single-consumer, when MultiProducer is selected on the channel's creation and SingleConsumer is selected when consuming; this scenario should be selected when there are multiple producers, but a single worker that consumes data sequentially (per ConsumerF); note that this means a single worker per ConsumerF instance, but you can still have multiple ConsumerF instances created, , because each ConsumerF gets its own buffer anyway
• SPMC: single-producer, multi-consumer, when SingleProducer is selected on the channel's creation and MultiConsumer is selected when consuming; this scenario should be selected when there are multiple workers processing data in parallel (e.g. pulling from the same ConsumerF), but a single producer that pushes data on the channel sequentially
• SPSC: single-producer, single-consumer, when SingleProducer is selected on the channel's creation and SingleConsumer is selected when consuming; this scenario should be selected when there is a single producer that pushes data on the channel sequentially and a single worker per ConsumerF instance that pulls data from the channel sequentially; note you can still have multiple ConsumerF instances running in parallel, because each ConsumerF gets its own buffer anyway

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

import cats.implicits._
import cats.effect.IO
import monix.execution.ChannelType.{SingleProducer, SingleConsumer}
import monix.execution.BufferCapacity.Bounded

val channel = ConcurrentChannel[IO].withConfig[Int, Int](
  producerType = SingleProducer
)

val consumerConfig = ConsumerF.Config(
  consumerType = Some(SingleConsumer)
)

for {
  producer  <- channel
  consumer1 =  producer.consumeWithConfig(consumerConfig)
  consumer2 =  producer.consumeWithConfig(consumerConfig)
  fiber1    <- consumer1.use { ref => ref.pull }.start
  fiber2    <- consumer2.use { ref => ref.pull }.start
  _         <- producer.push(1)
  value1    <- fiber1.join
  value2    <- fiber2.join
} yield {
  (value1, value2)
}

Note that in this example, even if we used SingleConsumer as the type passed in consumeWithConfig, we can still consume from two ConsumerF instances at the same time, because each one gets its own internal buffer. But you cannot have multiple workers per ConsumerF in this scenario, because this would break the internal synchronization / visibility guarantees.

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.

Credits

Inspired by Haskell's Control.Concurrent.ConcurrentChannel, but note that this isn't a straight port — e.g. the Monix implementation has a cleaner, non-leaky interface, is back-pressured and allows for termination (via halt), which changes its semantics significantly.