Stages are computation steps that send and/or receive data from other stages.

When a stage sends data, it acts as a producer. When it receives data, it acts as a consumer. Stages may take both producer and consumer roles at once.

Stage types

Besides taking both producer and consumer roles, a stage may be called “source” if it only produces items or called “sink” if it only consumes items.

For example, imagine the stages below where A sends data to B that sends data to C:

[A] -> [B] -> [C]

we conclude that:

• A is only a producer (and therefore a source)
• B is both producer and consumer
• C is only a consumer (and therefore a sink)

As we will see in the upcoming Examples section, we must specify the type of the stage when we implement each of them.

To start the flow of events, we subscribe consumers to producers. Once the communication channel between them is established, consumers will ask the producers for events. We typically say the consumer is sending demand upstream. Once demand arrives, the producer will emit items, never emitting more items than the consumer asked for. This provides a back-pressure mechanism.

A consumer may have multiple producers and a producer may have multiple consumers. When a consumer asks for data, each producer is handled separately, with its own demand. When a producer receives demand and sends data to multiple consumers, the demand is tracked and the events are sent by a dispatcher. This allows producers to send data using different “strategies”. See GenStage.Dispatcher for more information.

Example

Let’s define the simple pipeline below:

[A] -> [B] -> [C]

where A is a producer that will emit items starting from 0, B is a producer-consumer that will receive those items and multiply them by a given number and C will receive those events and print them to the terminal.

Let’s start with A. Since A is a producer, its main responsibility is to receive demand and generate events. Those events may be in memory or an external queue system. For simplicity, let’s implement a simple counter starting from a given value of counter received on init/1:

defmodule A do
  use GenStage

  def start_link(number) do
    GenStage.start_link(A, number)
  end

  def init(counter) do
    {:producer, counter}
  end

  def handle_demand(demand, counter) when demand > 0 do
    # If the counter is 3 and we ask for 2 items, we will
    # emit the items 3 and 4, and set the state to 5.
    events = Enum.to_list(counter..counter+demand-1)
    {:noreply, events, counter + demand}
  end
end

B is a producer-consumer. This means it does not explicitly handle the demand because the demand is always forwarded to its producer. Once A receives the demand from B, it will send events to B which will be transformed by B as desired. In our case, B will receive events and multiply them by a number given on initialization and stored as the state:

defmodule B do
  use GenStage

  def start_link(number) do
    GenStage.start_link(B, number)
  end

  def init(number) do
    {:producer_consumer, number}
  end

  def handle_events(events, _from, number) do
    events = Enum.map(events, & &1 * number)
    {:noreply, events, number}
  end
end

C will finally receive those events and print them every second to the terminal:

defmodule C do
  use GenStage

  def start_link() do
    GenStage.start_link(C, :ok)
  end

  def init(:ok) do
    {:consumer, :the_state_does_not_matter}
  end

  def handle_events(events, _from, state) do
    # Wait for a second.
    :timer.sleep(1000)

    # Inspect the events.
    IO.inspect(events)

    # We are a consumer, so we would never emit items.
    {:noreply, [], state}
  end
end

Now we can start and connect them:

{:ok, a} = A.start_link(0)  # starting from zero
{:ok, b} = B.start_link(2)  # multiply by 2
{:ok, c} = C.start_link()   # state does not matter

GenStage.sync_subscribe(c, to: b)
GenStage.sync_subscribe(b, to: a)

Notice we typically subscribe from bottom to top. Since A will start producing items only when B connects to it, we want this subscription to happen when the whole pipeline is ready. After you subscribe all of them, demand will start flowing upstream and events downstream.

When implementing consumers, we often set the :max_demand and :min_demand on subscription. The :max_demand specifies the maximum amount of events that must be in flow while the :min_demand specifies the minimum threshold to trigger for more demand. For example, if :max_demand is 1000 and :min_demand is 500 (the default values), the consumer will ask for 1000 events initially and ask for more only after it receives at least 500.

In the example above, B is a :producer_consumer and therefore acts as a buffer. Getting the proper demand values in B is important: making the buffer too small may make the whole pipeline slower, making the buffer too big may unnecessarily consume memory.

When such values are applied to the stages above, it is easy to see the producer works in batches. The producer A ends-up emitting batches of 50 items which will take approximately 50 seconds to be consumed by C, which will then request another batch of 50 items.

init and subscribe_to

In the example above, we have started the processes A, B and C independently and subscribed them later on. But most often it is simpler to subscribe a consumer to its producer on its init/1 callback. This way, if the consumer crashes, restarting the consumer will automatically re-invoke its init/1 callback and resubscribe it to the supervisor.

This approach works as long as the producer can be referenced when the consumer starts—such as by name (for a named process) or by pid for a running unnamed process. For example, assuming the process A and B are started as follows:

# Let's call the stage in module A as A
GenStage.start_link(A, 0, name: A)
# Let's call the stage in module B as B
GenStage.start_link(B, 2, name: B)
# No need to name consumers as they won't be subscribed to
GenStage.start_link(C, :ok)

We can now change the init/1 callback for C to the following:

def init(:ok) do
  {:consumer, :the_state_does_not_matter, subscribe_to: [B]}
end

Or:

def init(:ok) do
  {:consumer, :the_state_does_not_matter, subscribe_to: [{B, options}]}
end

And we will no longer need to call sync_subscribe/2.

Another advantage of this approach is that it makes it straight-forward to leverage concurrency by simply starting multiple consumers that subscribe to its producer (or producer_consumer). This can be done in the example above by simply calling start link multiple times:

# Start 4 consumers
GenStage.start_link(C, :ok)
GenStage.start_link(C, :ok)
GenStage.start_link(C, :ok)
GenStage.start_link(C, :ok)

In a supervision tree, this is often done by starting multiple workers:

children = [
  worker(A, [0]),
  worker(B, [2]),
  worker(C, []),
  worker(C, []),
  worker(C, []),
  worker(C, [])
]

Supervisor.start_link(children, strategy: :one_for_one)

In fact, multiple consumers is often the easiest and simplest way to leverage concurrency in a GenStage pipeline, especially if events can be processed out of order. For example, imagine a scenario where you have a stream of incoming events and you need to access a number of external services per event. Instead of building complex stages that route events through those services, one simple mechanism to leverage concurrency is to start a producer and N consumers and invoke the external services directly for each event in each consumer. N is typically the number of cores (as returned by System.schedulers_online/0) but can likely be increased if the consumers are mostly waiting on IO.

Another alternative to the scenario above, is to use a ConsumerSupervisor for consuming the events instead of N consumers. The ConsumerSupervisor will start a separate supervised process per event in a way you have at most max_demand children and the average amount of children is (max_demand - min_demand) / 2.

Buffering

In many situations, producers may attempt to emit events while no consumers have yet subscribed. Similarly, consumers may ask producers for events that are not yet available. In such cases, it is necessary for producers to buffer events until a consumer is available or buffer the consumer demand until events arrive, respectively. As we will see next, buffering events can be done automatically by GenStage, while buffering the demand is a case that must be explicitly considered by developers implementing producers.

Buffering events

Due to the concurrent nature of Elixir software, sometimes a producer may dispatch events without consumers to send those events to. For example, imagine a :consumer B subscribes to :producer A. Next, the consumer B sends demand to A, which uses to start producing events. Now, if the consumer B crashes, the producer may attempt to dispatch the now produced events but it no longer has a consumer to send those events to. In such cases, the producer will automatically buffer the events until another consumer subscribes.

The buffer can also be used in cases external sources only send events in batches larger than asked for. For example, if you are receiving events from an external source that only sends events in batches of 1000 and the internal demand is smaller than that, the buffer allows you to always emit batches of 1000 events even when the consumer has asked for less.

In all of those cases when an event cannot be sent immediately by a producer, the event will be automatically stored and sent the next time consumers ask for events. The size of the buffer is configured via the :buffer_size option returned by init/1 and the default value is 10000. If the buffer_size is exceeded, an error is logged.

Buffering demand

In case consumers send demand and the producer is not yet ready to fill in the demand, producers must buffer the demand until data arrives.

As an example, let’s implement a producer that broadcasts messages to consumers. For producers, we need to consider two scenarios:

1. what if events arrive and there are no consumers?
2. what if consumers send demand and there are not enough events?

One way to implement such a broadcaster is to simply rely on the internal buffer available in GenStage, dispatching events as they arrive, as explained in the previous section:

defmodule Broadcaster do
  use GenStage

  @doc "Starts the broadcaster."
  def start_link() do
    GenStage.start_link(__MODULE__, :ok, name: __MODULE__)
  end

  @doc "Sends an event and returns only after the event is dispatched."
  def sync_notify(pid, event, timeout \\ 5000) do
    GenStage.call(__MODULE__, {:notify, event}, timeout)
  end

  def init(:ok) do
    {:producer, :ok, dispatcher: GenStage.BroadcastDispatcher}
  end

  def handle_call({:notify, event}, _from, state) do
    {:reply, :ok, [event], state} # Dispatch immediately
  end

  def handle_demand(_demand, state) do
    {:noreply, [], state} # We don't care about the demand
  end
end

By always sending events as soon as they arrive, if there is any demand, we will serve the existing demand, otherwise the event will be queued in GenStage’s internal buffer. In case events are being queued and not being consumed, a log message will be emitted when we exceed the :buffer_size configuration.

While the implementation above is enough to solve the constraints above, a more robust implementation would have tighter control over the events and demand by tracking this data locally, leaving the GenStage internal buffer only for cases where consumers crash without consuming all data.

To handle such cases, we will make the broadcaster state a tuple with two elements: a queue and the pending demand. When events arrive and there are no consumers, we store the event in the queue alongside the process information that broadcasted the event. When consumers send demand and there are not enough events, we increase the pending demand. Once we have both the data and the demand, we acknowledge the process that has sent the event to the broadcaster and finally broadcast the event downstream.

defmodule QueueBroadcaster do
  use GenStage

  @doc "Starts the broadcaster."
  def start_link() do
    GenStage.start_link(__MODULE__, :ok, name: __MODULE__)
  end

  @doc "Sends an event and returns only after the event is dispatched."
  def sync_notify(event, timeout \\ 5000) do
    GenStage.call(__MODULE__, {:notify, event}, timeout)
  end

  ## Callbacks

  def init(:ok) do
    {:producer, {:queue.new, 0}, dispatcher: GenStage.BroadcastDispatcher}
  end

  def handle_call({:notify, event}, from, {queue, pending_demand}) do
    queue = :queue.in({from, event}, queue)
    dispatch_events(queue, pending_demand, [])
  end

  def handle_demand(incoming_demand, {queue, pending_demand}) do
    dispatch_events(queue, incoming_demand + pending_demand, [])
  end

  defp dispatch_events(queue, 0, events) do
    {:noreply, Enum.reverse(events), {queue, 0}}
  end
  defp dispatch_events(queue, demand, events) do
    case :queue.out(queue) do
      {{:value, {from, event}}, queue} ->
        GenStage.reply(from, :ok)
        dispatch_events(queue, demand - 1, [event | events])
      {:empty, queue} ->
        {:noreply, Enum.reverse(events), {queue, demand}}
    end
  end
end

Let’s also implement a consumer that automatically subscribes to the broadcaster on init/1. The advantage of doing so on initialization is that, if the consumer crashes while it is supervised, the subscription is automatically reestablished when the supervisor restarts it.

defmodule Printer do
  use GenStage

  @doc "Starts the consumer."
  def start_link() do
    GenStage.start_link(__MODULE__, :ok)
  end

  def init(:ok) do
    # Starts a permanent subscription to the broadcaster
    # which will automatically start requesting items.
    {:consumer, :ok, subscribe_to: [QueueBroadcaster]}
  end

  def handle_events(events, _from, state) do
    for event <- events do
      IO.inspect {self(), event}
    end
    {:noreply, [], state}
  end
end

With the broadcaster in hand, now let’s start the producer as well as multiple consumers:

# Start the producer
QueueBroadcaster.start_link()

# Start multiple consumers
Printer.start_link()
Printer.start_link()
Printer.start_link()
Printer.start_link()

At this point, all consumers must have sent their demand which we were not able to fulfill. Now by calling sync_notify, the event shall be broadcasted to all consumers at once as we have buffered the demand in the producer:

QueueBroadcaster.sync_notify(:hello_world)

If we had called QueueBroadcaster.sync_notify(:hello_world) before any consumer was available, the event would also be buffered in our own queue and served only when demand is received.

By having control over the demand and queue, the Broadcaster has full control on how to behave when there are no consumers, when the queue grows too large, and so forth.

Asynchronous work and handle_subscribe

Both producer_consumer and consumer have been designed to do their work in the handle_events/3 callback. This means that, after handle_events/3 is invoked, both producer_consumer and consumer will immediately send demand upstream and ask for more items, as it assumes events have been fully processed by handle_events/3.

Such default behaviour makes producer_consumer and consumer unfeasible for doing asynchronous work. However, given GenStage was designed to run with multiple consumers, it is not a problem to perform synchronous or blocking actions inside handle_events/3 as you can then start multiple consumers in order to max both CPU and IO usage as necessary.

On the other hand, if you must perform some work asynchronously, GenStage comes with an option that manually controls how demand is sent upstream, avoiding the default behaviour where demand is sent after handle_events/3. Such can be done by implementing the handle_subscribe/4 callback and returning {:manual, state} instead of the default {:automatic, state}. Once the producer mode is set to :manual, developers must use GenStage.ask/3 to send demand upstream when necessary.

For example, the ConsumerSupervisor module processes events asynchronously by starting child process and such is done by manually sending demand to producers. The ConsumerSupervisor can be used to distribute work to a limited amount of processes, behaving similar to a pool where a new process is started per event. The minimum amount of concurrent children per producer is specified by min_demand and the maximum is given by max_demand. See the ConsumerSupervisor docs for more information.

Setting the demand to :manual in handle_subscribe/4 is not only useful for asynchronous work but also for setting up other mechanisms for back-pressure. As an example, let’s implement a consumer that is allowed to process a limited number of events per time interval. Those are often called rate limiters:

defmodule RateLimiter do
  use GenStage

  def init(_) do
    # Our state will keep all producers and their pending demand
    {:consumer, %{}}
  end

  def handle_subscribe(:producer, opts, from, producers) do
    # We will only allow max_demand events every 5000 miliseconds
    pending = opts[:max_demand] || 1000
    interval = opts[:interval] || 5000

    # Register the producer in the state
    producers = Map.put(producers, from, {pending, interval})
    # Ask for the pending events and schedule the next time around
    producers = ask_and_schedule(producers, from)

    # Returns manual as we want control over the demand
    {:manual, producers}
  end

  def handle_cancel(_, from, producers) do
    # Remove the producers from the map on unsubscribe
    {:noreply, [], Map.delete(producers, from)}
  end

  def handle_events(events, from, producers) do
    # Bump the amount of pending events for the given producer
    producers = Map.update!(producers, from, fn {pending, interval} ->
      {pending + length(events), interval}
    end)

    # Consume the events by printing them.
    IO.inspect(events)

    # A producer_consumer would return the processed events here.
    {:noreply, [], producers}
  end

  def handle_info({:ask, from}, producers) do
    # This callback is invoked by the Process.send_after/3 message below.
    {:noreply, [], ask_and_schedule(producers, from)}
  end

  defp ask_and_schedule(producers, from) do
    case producers do
      %{^from => {pending, interval}} ->
        # Ask for any pending events
        GenStage.ask(from, pending)
        # And let's check again after interval
        Process.send_after(self(), {:ask, from}, interval)
        # Finally, reset pending events to 0
        Map.put(producers, from, {0, interval})
      %{} ->
        producers
    end
  end
end

With the RateLimiter implemented, let’s subscribe it to the producer we have implemented at the beginning of the module documentation:

{:ok, a} = GenStage.start_link(A, 0)
{:ok, b} = GenStage.start_link(RateLimiter, :ok)

# Ask for 10 items every 2 seconds
GenStage.sync_subscribe(b, to: a, max_demand: 10, interval: 2000)

Although the rate limiter above is a consumer, it could be made a producer_consumer by changing init/1 to return a :producer_consumer and then forwarding the events in handle_events/3.

Notifications

GenStage also supports the ability to send notifications to all consumers. Those notifications are sent as regular messages outside of the demand-driven protocol but respecting the event ordering. See sync_notify/3 and async_notify/2.

Notifications are useful for out-of-band information, for example, to notify consumers the producer has sent all events it had to process or that a new batch of events is starting.

Note the notification system should not be used for broadcasting events, for such, consider using GenStage.BroadcastDispatcher.

Callbacks

GenStage is implemented on top of a GenServer with two additions. Besides exposing all of the GenServer callbacks, it also provides handle_demand/2 to be implemented by producers and handle_events/3 to be implemented by consumers, as shown above. Furthermore, all the callback responses have been modified to potentially emit events. See the callbacks documentation for more information.

By adding use GenStage to your module, Elixir will automatically define all callbacks for you except the following:

• init/1 - must be implemented to choose between :producer, :consumer or :producer_consumer
• handle_demand/2 - must be implemented by :producer types
• handle_events/3 - must be implemented by :producer_consumer and :consumer types

Although this module exposes functions similar to the ones found in the GenServer API, like call/3 and cast/2, developers can also rely directly on GenServer functions such as GenServer.multi_call/4 and GenServer.abcast/3 if they wish to.

Name Registration

GenStage is bound to the same name registration rules as a GenServer. Read more about it in the GenServer docs.

Message-protocol overview

This section will describe the message-protocol implemented by stages. By documenting these messages, we will allow developers to provide their own stage implementations.

Back-pressure

When data is sent between stages, it is done by a message protocol that provides back-pressure. The first step is for the consumer to subscribe to the producer. Each subscription has a unique reference.

Once subscribed, the consumer may ask the producer for messages for the given subscription. The consumer may demand more items whenever it wants to. A consumer must never receive more data than it has asked for from any given producer stage.

A consumer may have multiple producers, where each demand is managed individually. A producer may have multiple consumers, where the demand and events are managed and delivered according to a GenStage.Dispatcher implementation.

Producer messages

The producer is responsible for sending events to consumers based on demand.

• {:"$gen_producer", from :: {consumer_pid, subscription_tag}, {:subscribe, current, options}} - sent by the consumer to the producer to start a new subscription.

  Before sending, the consumer MUST monitor the producer for clean-up purposes in case of crashes. The subscription_tag is unique to identify the subscription. It is typically the subscriber monitoring reference although it may be any term.

  Once sent, the consumer MAY immediately send demand to the producer. The subscription_tag is unique to identify the subscription.

  The current field, when not nil, is a two-item tuple containing a subscription that must be cancelled with the given reason before the current one is accepted.

  Once received, the producer MUST monitor the consumer. However, if the subscription reference is known, it MUST send a :cancel message to the consumer instead of monitoring and accepting the subscription.

• {:"$gen_producer", from :: {pid, subscription_tag}, {:cancel, reason}} - sent by the consumer to cancel a given subscription.

  Once received, the producer MUST send a :cancel reply to the registered consumer (which may not necessarily be the one received in the tuple above). Keep in mind, however, there is no guarantee such messages can be delivered in case the producer crashes before. If the pair is unknown, the producer MUST send an appropriate cancel reply.

• {:"$gen_producer", from :: {pid, subscription_tag}, {:ask, count}} - sent by consumers to ask data in a given subscription.

  Once received, the producer MUST send data up to the demand. If the pair is unknown, the producer MUST send an appropriate cancel reply.

Consumer messages

The consumer is responsible for starting the subscription and sending demand to producers.

• {:"$gen_consumer", from :: {producer_pid, subscription_tag}, {:notification, msg}} - notifications sent by producers.

• {:"$gen_consumer", from :: {producer_pid, subscription_tag}, {:cancel, reason}} - sent by producers to cancel a given subscription.

  It is used as a confirmation for client cancellations OR whenever the producer wants to cancel some upstream demand.

• {:"$gen_consumer", from :: {producer_pid, subscription_tag}, [event]} - events sent by producers to consumers.

  subscription_tag identifies the subscription. The third argument is a non-empty list of events. If the subscription is unknown, the events must be ignored and a cancel message sent to the producer.