I’m fairly new to Elixir, and so far I’m finding it very exciting. So many of my concurrency problems seem to be solved in a simple, elegant way. Now i’ve learnt the basics, I’m trying to write some applications and am having trouble working out the right patterns.
First I was using Wabbit as a Genstage producer (consuming from a rabbitmq queue). I was doing some processing on the messages and then inserting into a DB. It worked great locally, around ~8000/s. When I moved it to a production environment, after excessive logging, I worked out that it was choking up at the Sink end (backpressure was working!) because of latency of writes. Locally, i would get say, 500 events to the consumer at a time, but in production i would get 1 at a time, which would make for higher latency on the write per message.
My solution was to create a stage which buffered upto x events, or for x ms (whichever came first). To be more concrete, here is the code.
# This will handle the acks back to rabbitmq
defmodule RabbitAckker do
use GenStage
require Logger
def start_link() do
GenStage.start_link(__MODULE__, :ok)
end
def init(:ok) do
Process.send_after(self(), {:flush}, 500) # first message is fairly soon after startup
{:producer_consumer, []}
end
# handle events coming from rabbit
def handle_events(events, _from, state) do
event_buffer = state
new_event_buffer = event_buffer ++ events
messages_before_purge = 400
if (Enum.count(new_event_buffer) >= messages_before_purge) do
for {event, meta} <- new_event_buffer do
:ok = ack(meta.channel, meta.delivery_tag)
end
# pass all events down to the next stage
{:noreply, new_event_buffer, []}
else
# don't pass any events down, just store them in the state
{:noreply, [], new_event_buffer}
end
end
# when this flush message comes, ack the events and pass them down.
# TODO: only flush when we haven't acked anything in flush_time ms
def handle_info({:flush}, state) do
flush_time = 2_000 # ms
event_buffer = state
event_count = Enum.count(state)
if event_count > 0 do
Logger.debug("ackker: timer flush called for #{event_count} events")
end
# ack all the events we will flush
for {event, meta} <- event_buffer do
:ok = ack(meta.channel, meta.delivery_tag)
end
# call this again in another flush_time ms
Process.send_after(self(), {:flush}, flush_time)
# move event_buffer to events, and reset the state
{:noreply, event_buffer, []}
end
defp ack(channel, delivery_tag) do
try do
Wabbit.Basic.ack(channel, delivery_tag)
catch
_, _ ->
:ok
end
end
end
This works great! I realise that I have potential for dataloss if the process is killed before flush_time but you have that problem with any sort of buffering.
My first question is, is there a better way to do this?
Now on to my next problem!
In addition to writing to a datastore, I would like to spit out the messages to multiple websockets. Sometimes the consumers on the end of the websocket are slow. Each of them have different performance characteristics. How can I drop messages to them if they aren’t keeping up? I am trying to avoid an unbounded buffer. I am also trying to write to the datastore at the same time, but i want the datastore to get every message and apply backpressure (via demand) as it is in the above example.
Any suggestions on a good pattern for this?
Thanks for taking the time to read through my whole question!
