diff --git a/config/config.exs b/config/config.exs index 685f3a7..06c4c00 100644 --- a/config/config.exs +++ b/config/config.exs @@ -67,7 +67,9 @@ config :aprsme, # 2 seconds - longer timeout for larger batches batch_timeout: 2000, # Higher demand to keep pipeline flowing - max_demand: 250 + max_demand: 750, + # Number of parallel consumers for better throughput + num_consumers: 3 ] config :error_tracker, diff --git a/lib/aprsme/packet_consumer.ex b/lib/aprsme/packet_consumer.ex index cb0ff15..731b710 100644 --- a/lib/aprsme/packet_consumer.ex +++ b/lib/aprsme/packet_consumer.ex @@ -11,7 +11,14 @@ defmodule Aprsme.PacketConsumer do require Logger def start_link(opts \\ []) do - GenStage.start_link(__MODULE__, opts, name: __MODULE__) + # Allow unnamed consumers for pool usage + name = opts[:name] + + if name do + GenStage.start_link(__MODULE__, opts, name: name) + else + GenStage.start_link(__MODULE__, opts) + end end @impl true @@ -24,6 +31,9 @@ defmodule Aprsme.PacketConsumer do # Start a timer for batch processing timer = Process.send_after(self(), :process_batch, batch_timeout) + # Extract subscription options if provided + subscribe_to = opts[:subscribe_to] || [{Aprsme.PacketProducer, max_demand: opts[:max_demand] || 250}] + {:consumer, %{ batch: [], @@ -31,11 +41,15 @@ defmodule Aprsme.PacketConsumer do batch_timeout: batch_timeout, max_batch_size: max_batch_size, timer: timer - }} + }, subscribe_to: subscribe_to} end @impl true - def handle_events(events, _from, %{batch: batch, batch_size: batch_size, max_batch_size: max_batch_size} = state) do + def handle_events( + events, + _from, + %{batch: batch, batch_size: batch_size, max_batch_size: max_batch_size, timer: timer} = state + ) do new_batch = batch ++ events new_batch_length = length(new_batch) @@ -57,12 +71,19 @@ defmodule Aprsme.PacketConsumer do ) end - {:noreply, [], %{state | batch: []}} + # Cancel and restart timer + Process.cancel_timer(timer) + new_timer = Process.send_after(self(), :process_batch, state.batch_timeout) + {:noreply, [], %{state | batch: [], timer: new_timer}} new_batch_length >= batch_size -> # Process the batch immediately process_batch(new_batch) - {:noreply, [], %{state | batch: []}} + + # Cancel and restart timer + Process.cancel_timer(timer) + new_timer = Process.send_after(self(), :process_batch, state.batch_timeout) + {:noreply, [], %{state | batch: [], timer: new_timer}} true -> # Add to batch and wait for more diff --git a/lib/aprsme/packet_consumer_pool.ex b/lib/aprsme/packet_consumer_pool.ex new file mode 100644 index 0000000..f805fb2 --- /dev/null +++ b/lib/aprsme/packet_consumer_pool.ex @@ -0,0 +1,41 @@ +defmodule Aprsme.PacketConsumerPool do + @moduledoc """ + Manages a pool of packet consumers for parallel processing. + Each consumer subscribes to the producer with its own demand. + """ + use Supervisor + + def start_link(opts \\ []) do + Supervisor.start_link(__MODULE__, opts, name: __MODULE__) + end + + @impl true + def init(opts) do + config = Application.get_env(:aprsme, :packet_pipeline, []) + + # Number of parallel consumers (default to 3 for better throughput) + num_consumers = opts[:num_consumers] || config[:num_consumers] || 3 + + # Each consumer gets a portion of the max demand + max_demand_per_consumer = div(config[:max_demand] || 250, num_consumers) + + children = + for index <- 1..num_consumers do + %{ + id: {Aprsme.PacketConsumer, index}, + start: + {Aprsme.PacketConsumer, :start_link, + [ + [ + batch_size: config[:batch_size] || 100, + batch_timeout: config[:batch_timeout] || 1000, + max_demand: max_demand_per_consumer, + subscribe_to: [{Aprsme.PacketProducer, max_demand: max_demand_per_consumer}] + ] + ]} + } + end + + Supervisor.init(children, strategy: :one_for_one) + end +end diff --git a/lib/aprsme/packet_pipeline_supervisor.ex b/lib/aprsme/packet_pipeline_supervisor.ex index afb6316..826670a 100644 --- a/lib/aprsme/packet_pipeline_supervisor.ex +++ b/lib/aprsme/packet_pipeline_supervisor.ex @@ -12,10 +12,13 @@ defmodule Aprsme.PacketPipelineSupervisor do def init(_opts) do config = Application.get_env(:aprsme, :packet_pipeline, []) - children = [ - {Aprsme.PacketProducer, max_buffer_size: config[:max_buffer_size] || 1000}, - {Aprsme.PacketConsumer, batch_size: config[:batch_size] || 100, batch_timeout: config[:batch_timeout] || 1000} - ] + # Configure producer with correct buffer size from config + producer_spec = {Aprsme.PacketProducer, max_buffer_size: config[:max_buffer_size] || 1000} + + # Use consumer pool for better throughput + consumer_pool_spec = {Aprsme.PacketConsumerPool, num_consumers: config[:num_consumers] || 3} + + children = [producer_spec, consumer_pool_spec] Supervisor.init(children, strategy: :one_for_one) end