dingosky

dingosky

The (excellent) GenStage documentation includes a QueueBroadcaster module as an example for “a more robust implementation” with “tighter control over the events and demand by tracking … data locally”. I’ve included the example code in its entirety below for reference. My question concerns the inclusion of from in the queued data.

Following GenServer best practices, QueueBroadcaster provides sync_notify/2 as public API for placing an event into this GenStage producer. That function simply does a GenStage.call/3 which, when handled by handle_call/3, enqueues the tuple {from, event} in the internal queue held in state and calls dispatch_events/3 with the current state pending_demand. The handle_call return results from the dispatch_events/3 function, which when there is positive demand, dequeues {from, event} tuples from the queue in state, and for each tuple calls GenStage.reply(from, :ok) and adds the event to a list to be delivered in the returned :no_reply tuple.

All is well so far, but here’s the question. Why enqueue the value from and make the GenStage.reply(from, :ok) call at all? The reasons this seems superfluous are:

  • Since the public API sync_notify/2 function properly encapsulated the GenStage.call/3 for :notify messaging, any call from sync_notify/2 will have from == self(). So GenStage.reply(from, :ok) will send a message to itself, and since the QueueBroadcaster module does not provide a handle_info/2 function to actually process that message, the process effectively sends itself a message that it ignores.

  • Even if there were a handle_info/2 function in QueueBroadcaster, sending a simple :ok without any indication of the event that was handled would be of very limited value.

I would suggest the example code could be simplified by dispensing of tracking from in state (via queue) and removing the ignored GenStage.reply/2 call altogether. There would be no change in functionality and there would be no need for those learning GenStage to unwind the question: Why is from being kept in state?

Am I missing something?

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

Showing Posts 1 to 2

dingosky

dingosky OP

Okay, pardon the noise. I see now why the GenStage.reply(from, :ok) is necessary. The sync GenStage.call needs a corresponding GenStage.reply to “complete” the call (o/w a timeout occurs).

LostKobrakai

LostKobrakai

That’s not what it does. Say pid A calls sync_notify then the GenStage.call will lookup the pid registered with the name QueueBroadcaster, say pid B and queue an event.

Pid B being the GenStage producer will receive an opaque value from in handle_call, which does later allow GenStage.reply to reply with an answer to the call, because the initial handle_call doesn’t provide an answer with its return value.

This means the sync_notify doesn’t return for pid A until the event has been dispatched even if there hasn’t been any demand at the time the {:notify, event} message was received by pid B/the producer.

— All posts loaded —

Where Next? Top

Trending in Questions Top

stjefim
Hello! Suppose you are building workflow (order / task / payment) processing system with the following requirements: Each workflow con...
New
Blokh
Hey guys, I’ve got a huge CSV ( around 10 GB ) that needs to be processed hourly Do you guys have any suggestions what is the best prac...
New
kszambelanczyk
Hello! Could someone please give me a help/sample code, how to delete a file from s3 using waffle/waffle_ecto from Phoenix app. I creat...
New
Onor.io
I have what I’ve heard referred to as a “lookup table” in my database. This is a way of assigning codes to common values. One common lo...
New
jaybe78
Hello, I’m developing a online persistent chat system (what’s app) like using elixir/dynamodb/aws for a mobile app(flutter). The diffic...
New
Trolleger
What approach to take when sending live updates to “random” users Hi! I have a question, I have a little chat app, and when I create a DM...
New
widianto
I think I’ve found a small improvement I could contribute to <%= web_namespace %>.CoreComponents (installer/templates/phx_web/compo...
New

Other Trending Topics Top

garrison
Hobbes is a low-level distributed database for the Elixir programming language. Hobbes provides a simple, safe, and scalable storage lay...
New
mcass19
ExRatatui lets you cook up rich terminal UIs in Elixir, powered by Rust’s ratatui via Rustler NIFs. Build interactive terminal applicatio...
New
Damirados
Hello everyone. After busy few months I am happy to announce v0.1.0 of Emerge & Solve. They are GUI (Emerge) and State management (S...
New
netoum
Corex is an accessible, unstyled UI component library for Phoenix that integrates Zag.js state machines using Vanilla JavaScript and Live...
New
wintermeyer
There are three potential reasons for members of this forum to have a look at https://vutuv.de You are tired or annoyed of LinkedIn. Yo...
New
webofbits
Aludel - LLM Evaluation Workbench Aludel is an embeddable Phoenix LiveView dashboard for evaluating and comparing LLM prompts across mult...
New

We're in Beta

About us Mission Statement

Options

Thread Display Mode




Thread Preview

Skip Thread Previews