From 8811298eed62f14e42cbc1ac4e6c0bbd2998edeb Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 15 Jul 2025 08:16:11 -0500 Subject: [PATCH] Fix packet buffer overflow and system load adjustment issues MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Fixed InsertOptimizer returning map instead of keyword list for Ecto - Increased batch size ranges to handle high load (100-800 packets) - Improved consumer responsiveness: - Reduced batch timeout from 1000ms to 500ms - Process batches at 80% capacity for better throughput - Reduced adjustment interval from 10s to 5s - Added comprehensive logging for debugging buffer status - Created integration tests for packet pipeline - Coordinated batch sizing between SystemMonitor and InsertOptimizer 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude --- lib/aprsme/packet_consumer.ex | 16 +++-- lib/aprsme/packet_producer.ex | 24 ++++++- lib/aprsme/performance/insert_optimizer.ex | 14 ++--- lib/aprsme/system_monitor.ex | 6 +- .../packet_pipeline_integration_test.exs | 62 +++++++++++++++++++ 5 files changed, 106 insertions(+), 16 deletions(-) create mode 100644 test/aprsme/packet_pipeline_integration_test.exs diff --git a/lib/aprsme/packet_consumer.ex b/lib/aprsme/packet_consumer.ex index 4a4c8c3..808acb2 100644 --- a/lib/aprsme/packet_consumer.ex +++ b/lib/aprsme/packet_consumer.ex @@ -19,7 +19,7 @@ defmodule Aprsme.PacketConsumer do def init(opts) do # Use dynamic batch sizing from system monitor initial_batch_size = Aprsme.SystemMonitor.get_recommended_batch_size() - batch_timeout = opts[:batch_timeout] || 1000 + batch_timeout = opts[:batch_timeout] || 500 # Maximum batch size to prevent unbounded memory growth max_batch_size = opts[:max_batch_size] || 1000 @@ -27,7 +27,7 @@ defmodule Aprsme.PacketConsumer do timer = Process.send_after(self(), :process_batch, batch_timeout) # Schedule periodic batch size adjustment - Process.send_after(self(), :adjust_batch_size, 10_000) + Process.send_after(self(), :adjust_batch_size, 5_000) {:consumer, %{ @@ -46,6 +46,11 @@ defmodule Aprsme.PacketConsumer do current_batch_size = Aprsme.SystemMonitor.get_recommended_batch_size() state = %{state | batch_size: current_batch_size} + # Debug logging + Logger.debug( + "PacketConsumer received #{length(events)} events, current batch: #{length(batch)}, batch_size threshold: #{current_batch_size}" + ) + new_batch = batch ++ events new_batch_length = length(new_batch) @@ -69,7 +74,8 @@ defmodule Aprsme.PacketConsumer do {:noreply, [], %{state | batch: []}} - new_batch_length >= current_batch_size -> + # Process immediately if we reach 80% of target batch size to improve responsiveness + new_batch_length >= current_batch_size * 0.8 -> # Process the batch immediately process_batch(new_batch) {:noreply, [], %{state | batch: []}} @@ -120,7 +126,7 @@ defmodule Aprsme.PacketConsumer do end # Schedule next adjustment - Process.send_after(self(), :adjust_batch_size, 10_000) + Process.send_after(self(), :adjust_batch_size, 5_000) {:noreply, [], %{state | batch_size: new_batch_size}} end @@ -135,6 +141,8 @@ defmodule Aprsme.PacketConsumer do # Use optimized batch size for INSERT performance batch_size = InsertOptimizer.get_optimal_batch_size() + Logger.debug("Processing batch of #{length(packets)} packets with insert chunk size: #{batch_size}") + results = packets |> Enum.chunk_every(batch_size) diff --git a/lib/aprsme/packet_producer.ex b/lib/aprsme/packet_producer.ex index f8ce11c..2d12ad2 100644 --- a/lib/aprsme/packet_producer.ex +++ b/lib/aprsme/packet_producer.ex @@ -35,11 +35,31 @@ defmodule Aprsme.PacketProducer do # No demand, buffer the packet new_buffer = [packet_data | buffer] - if length(new_buffer) > max_size do + buffer_size = length(new_buffer) + + if buffer_size > max_size do # Buffer is full, drop oldest packet - Logger.warning("Packet buffer full, dropping oldest packet") + Logger.warning("Packet buffer full, dropping oldest packet", + buffer_status: %{ + current_size: buffer_size, + max_size: max_size, + dropped: 1 + } + ) + {:noreply, [], %{state | buffer: Enum.take(new_buffer, max_size)}} else + # Log when buffer is getting full + if buffer_size > max_size * 0.8 do + Logger.warning("Packet buffer approaching capacity", + buffer_status: %{ + current_size: buffer_size, + max_size: max_size, + utilization: Float.round(buffer_size / max_size * 100, 1) + } + ) + end + {:noreply, [], %{state | buffer: new_buffer}} end end diff --git a/lib/aprsme/performance/insert_optimizer.ex b/lib/aprsme/performance/insert_optimizer.ex index 5a32c25..c73c662 100644 --- a/lib/aprsme/performance/insert_optimizer.ex +++ b/lib/aprsme/performance/insert_optimizer.ex @@ -14,9 +14,9 @@ defmodule Aprsme.Performance.InsertOptimizer do require Logger # Configuration for INSERT optimization - @base_batch_size 100 - @max_batch_size 500 - @min_batch_size 50 + @base_batch_size 200 + @max_batch_size 800 + @min_batch_size 100 # 30 seconds @optimization_check_interval 30_000 @@ -163,24 +163,24 @@ defmodule Aprsme.Performance.InsertOptimizer do # If INSERTs are taking too long, use more aggressive optimizations if avg_duration > 3000 do - Map.merge(base_options, %{ + Keyword.merge(base_options, returning: false, on_conflict: :nothing, timeout: 30_000 - }) + ) else base_options end end defp default_insert_options do - %{ + [ # Don't return IDs unless needed returning: false, # Skip conflicts instead of raising on_conflict: :nothing, # 15 second timeout timeout: 15_000 - } + ] end end diff --git a/lib/aprsme/system_monitor.ex b/lib/aprsme/system_monitor.ex index d1cfad0..1c1e015 100644 --- a/lib/aprsme/system_monitor.ex +++ b/lib/aprsme/system_monitor.ex @@ -7,9 +7,9 @@ defmodule Aprsme.SystemMonitor do require Logger @check_interval 5_000 - @min_batch_size 50 - @max_batch_size 500 - @default_batch_size 100 + @min_batch_size 100 + @max_batch_size 800 + @default_batch_size 200 def start_link(opts \\ []) do GenServer.start_link(__MODULE__, opts, name: __MODULE__) diff --git a/test/aprsme/packet_pipeline_integration_test.exs b/test/aprsme/packet_pipeline_integration_test.exs new file mode 100644 index 0000000..e38c19d --- /dev/null +++ b/test/aprsme/packet_pipeline_integration_test.exs @@ -0,0 +1,62 @@ +defmodule Aprsme.PacketPipelineIntegrationTest do + use Aprsme.DataCase, async: false + + import ExUnit.CaptureLog + + alias Aprsme.PacketProducer + + describe "packet pipeline under load" do + test "system adjusts batch sizes based on load" do + # Test the actual running system without starting new processes + initial_batch_size = Aprsme.SystemMonitor.get_recommended_batch_size() + assert initial_batch_size >= 100 + assert initial_batch_size <= 800 + + # Submit some packets to the running system + log_output = + capture_log(fn -> + for i <- 1..10 do + packet = %{ + sender: "TEST-#{i}", + destination: "APRS", + path: "WIDE1-1", + information_field: "Integration test packet #{i}", + data_type: "position", + lat: 40.0 + i / 1000, + lon: -74.0 + i / 1000 + } + + PacketProducer.submit_packet(packet) + end + + # Wait a bit for processing + Process.sleep(100) + end) + + # Verify no errors occurred + refute log_output =~ "error" + end + + test "insert optimizer provides reasonable batch sizes" do + batch_size = Aprsme.Performance.InsertOptimizer.get_optimal_batch_size() + assert batch_size >= 100 + assert batch_size <= 800 + end + + test "system monitor provides metrics" do + metrics = Aprsme.SystemMonitor.get_metrics() + + assert is_map(metrics) + assert Map.has_key?(metrics, :memory) + assert Map.has_key?(metrics, :cpu) + assert Map.has_key?(metrics, :processes) + assert Map.has_key?(metrics, :db_pool) + + # Check pressure values are reasonable + assert metrics.memory.pressure >= 0.0 + assert metrics.memory.pressure <= 1.0 + assert metrics.cpu.pressure >= 0.0 + assert metrics.cpu.pressure <= 1.0 + end + end +end