From 53c88d22626e009afa19f5cdeedfa80b2268bd8d Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 15 Jul 2025 09:06:40 -0500 Subject: [PATCH] postgres optimization --- config/config.exs | 13 +- config/dev.exs | 24 +++- config/runtime.exs | 40 +++++- lib/aprsme/db_optimizer.ex | 226 ++++++++++++++++++++++++++++++++++ lib/aprsme/packet_consumer.ex | 20 ++- 5 files changed, 307 insertions(+), 16 deletions(-) create mode 100644 lib/aprsme/db_optimizer.ex diff --git a/config/config.exs b/config/config.exs index 4ad2210..685f3a7 100644 --- a/config/config.exs +++ b/config/config.exs @@ -58,11 +58,16 @@ config :aprsme, # Packet retention period in days (default: 365 days = 1 year) packet_retention_days: String.to_integer(System.get_env("PACKET_RETENTION_DAYS", "365")), # GenStage packet processing configuration + # Optimized for PostgreSQL with work_mem=16MB and synchronous_commit=off packet_pipeline: [ - max_buffer_size: 1000, - batch_size: 100, - batch_timeout: 1000, - max_demand: 50 + # Larger buffer since inserts are async + max_buffer_size: 5000, + # ~32KB per packet, stays within work_mem + batch_size: 500, + # 2 seconds - longer timeout for larger batches + batch_timeout: 2000, + # Higher demand to keep pipeline flowing + max_demand: 250 ] config :error_tracker, diff --git a/config/dev.exs b/config/dev.exs index 1574c54..9c2dc15 100644 --- a/config/dev.exs +++ b/config/dev.exs @@ -14,11 +14,27 @@ config :aprsme, Aprsme.Repo, database: "aprsme_dev", stacktrace: true, show_sensitive_data_on_connection_error: true, - pool_size: 10, - pool_timeout: 5000, - timeout: 15_000, + # Optimized pool settings for development + pool_size: 15, + pool_timeout: 10_000, + timeout: 30_000, + queue_target: 100, + queue_interval: 1_000, log: :debug, - types: Aprsme.PostgresTypes + types: Aprsme.PostgresTypes, + # Use unnamed prepared statements for better development flexibility + prepare: :unnamed, + # Socket options for better performance + socket_options: [ + keepalive: true, + nodelay: true + ], + # Development-specific parameters matching PostgreSQL config + parameters: [ + application_name: "aprsme_dev", + work_mem: "16MB", + statement_timeout: "60s" + ] config :aprsme, AprsmeWeb.Endpoint, # Binding to loopback ipv4 address prevents access from other machines. diff --git a/config/runtime.exs b/config/runtime.exs index 34b4c59..27a9d75 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -49,13 +49,41 @@ if config_env() == :prod do config :aprsme, Aprsme.Repo, # ssl: true, url: database_url, - pool_size: String.to_integer(System.get_env("POOL_SIZE") || "20"), - pool_timeout: String.to_integer(System.get_env("POOL_TIMEOUT") || "5000"), - timeout: String.to_integer(System.get_env("DB_TIMEOUT") || "15000"), - socket_options: maybe_ipv6, + # Optimized for max_connections=100 with other apps on server + pool_size: String.to_integer(System.get_env("POOL_SIZE") || "25"), + # Increased timeout for ARM system under load + pool_timeout: String.to_integer(System.get_env("POOL_TIMEOUT") || "10000"), + # Match PostgreSQL statement timeout capabilities + timeout: String.to_integer(System.get_env("DB_TIMEOUT") || "30000"), + socket_options: + maybe_ipv6 ++ + [ + # TCP optimizations for better connection handling + keepalive: true, + nodelay: true, + recbuf: 8192, + sndbuf: 8192 + ], types: Aprsme.PostgresTypes, - queue_target: 10_000, - queue_interval: 20_000 + # Reduced queue target for faster response + queue_target: 100, + # Check queue more frequently + queue_interval: 1_000, + # Optimize for unnamed prepared statements (better for dynamic queries) + prepare: :unnamed, + # Connection parameters to leverage PostgreSQL settings + parameters: [ + # Application name for monitoring + application_name: "aprsme", + # Leverage synchronous_commit=off in postgresql.conf + synchronous_commit: "off", + # Match work_mem setting + work_mem: "16MB", + # Statement timeout as safety net + statement_timeout: "30s", + # Prevent idle transactions + idle_in_transaction_session_timeout: "60s" + ] config :aprsme, AprsmeWeb.Endpoint, url: [host: host, port: 443, scheme: "https"], diff --git a/lib/aprsme/db_optimizer.ex b/lib/aprsme/db_optimizer.ex new file mode 100644 index 0000000..c022503 --- /dev/null +++ b/lib/aprsme/db_optimizer.ex @@ -0,0 +1,226 @@ +defmodule Aprsme.DbOptimizer do + @moduledoc """ + Database optimization utilities that leverage PostgreSQL server configuration + for better performance on ARM RK3588 with 16GB RAM. + + PostgreSQL key settings we're optimizing for: + - max_connections = 100 + - shared_buffers = 1GB + - work_mem = 16MB + - synchronous_commit = off + - effective_io_concurrency = 100 (SSD optimized) + """ + + alias Aprsme.Repo + alias Ecto.Adapters.SQL + + require Logger + + @doc """ + Execute a large batch insert using PostgreSQL COPY command for maximum performance. + This is significantly faster than INSERT for large datasets. + """ + def copy_insert(table_name, columns, rows) when length(rows) > 1000 do + # Convert rows to CSV format for COPY + csv_data = rows_to_csv(rows, columns) + + # Use COPY command which bypasses much of the overhead of INSERT + query = """ + COPY #{table_name} (#{Enum.join(columns, ", ")}) + FROM STDIN WITH (FORMAT csv, HEADER false) + """ + + SQL.query!(Repo, query, [csv_data]) + length(rows) + rescue + error -> + Logger.error("COPY insert failed: #{inspect(error)}") + # Fall back to regular insert + regular_batch_insert(table_name, columns, rows) + end + + def copy_insert(table_name, columns, rows) do + # For smaller batches, use regular insert + regular_batch_insert(table_name, columns, rows) + end + + @doc """ + Optimized batch insert that leverages PostgreSQL configuration + """ + def optimized_batch_insert(schema, entries, opts \\ []) do + # Calculate optimal batch size based on work_mem + optimal_batch_size = calculate_optimal_batch_size(entries) + + # Split into optimal chunks + entries + |> Enum.chunk_every(optimal_batch_size) + |> Enum.map(fn batch -> + insert_opts = + Keyword.merge( + [ + returning: false, + on_conflict: :nothing, + timeout: 60_000, + placeholders: length(batch) > 100 + ], + opts + ) + + case Repo.insert_all(schema, batch, insert_opts) do + {count, _} -> {:ok, count} + error -> {:error, error} + end + end) + |> Enum.reduce({0, 0}, fn + {:ok, count}, {success, errors} -> {success + count, errors} + {:error, _}, {success, errors} -> {success, errors + 1} + end) + end + + @doc """ + Calculate optimal batch size based on PostgreSQL work_mem setting (16MB) + and estimated row size + """ + def calculate_optimal_batch_size(entries) when is_list(entries) do + # Estimate size of one entry (rough approximation) + sample = List.first(entries) + estimated_size = estimate_entry_size(sample) + + # work_mem is 16MB, leave some headroom + # 14MB in bytes + available_memory = 14 * 1024 * 1024 + + # Calculate how many entries fit in work_mem + max_batch = div(available_memory, estimated_size) + + # Cap at reasonable limits + max_batch |> min(2000) |> max(100) + end + + @doc """ + Run ANALYZE on a table after bulk inserts to update statistics + This helps PostgreSQL make better query plans + """ + def analyze_table(table_name) do + SQL.query!(Repo, "ANALYZE #{table_name}", []) + :ok + rescue + error -> + Logger.warning("Failed to analyze table #{table_name}: #{inspect(error)}") + :error + end + + @doc """ + Vacuum a table to reclaim space and update visibility map + Use this after large delete operations + """ + def vacuum_table(table_name, opts \\ []) do + full = Keyword.get(opts, :full, false) + analyze = Keyword.get(opts, :analyze, true) + + vacuum_type = if full, do: "VACUUM FULL", else: "VACUUM" + analyze_clause = if analyze, do: " ANALYZE", else: "" + + query = "#{vacuum_type}#{analyze_clause} #{table_name}" + + # Vacuum operations can take a long time + SQL.query!(Repo, query, [], timeout: :infinity) + :ok + rescue + error -> + Logger.error("Failed to vacuum table #{table_name}: #{inspect(error)}") + :error + end + + @doc """ + Get current database statistics for monitoring + """ + def get_connection_stats do + query = """ + SELECT + count(*) as total_connections, + count(*) FILTER (WHERE state = 'active') as active_connections, + count(*) FILTER (WHERE state = 'idle') as idle_connections, + count(*) FILTER (WHERE state = 'idle in transaction') as idle_in_transaction, + count(*) FILTER (WHERE wait_event_type IS NOT NULL) as waiting_connections + FROM pg_stat_activity + WHERE datname = current_database() + """ + + case SQL.query(Repo, query, []) do + {:ok, %{rows: [[total, active, idle, idle_tx, waiting]]}} -> + %{ + total: total, + active: active, + idle: idle, + idle_in_transaction: idle_tx, + waiting: waiting + } + + _ -> + %{} + end + end + + # Private functions + + defp regular_batch_insert(table_name, columns, rows) do + # Convert to maps for Ecto.insert_all + entries = + Enum.map(rows, fn row -> + columns |> Enum.zip(row) |> Map.new() + end) + + {count, _} = + Repo.insert_all(table_name, entries, + returning: false, + on_conflict: :nothing, + timeout: 60_000 + ) + + count + end + + defp rows_to_csv(rows, _columns) do + Enum.map_join(rows, "\n", fn row -> + Enum.map_join(row, ",", &escape_csv_value/1) + end) + end + + defp escape_csv_value(nil), do: "" + + defp escape_csv_value(value) when is_binary(value) do + if String.contains?(value, [",", "\"", "\n"]) do + "\"#{String.replace(value, "\"", "\"\"")}\"" + else + value + end + end + + defp escape_csv_value(value), do: to_string(value) + + # Default 1KB + defp estimate_entry_size(nil), do: 1024 + + defp estimate_entry_size(entry) when is_map(entry) do + # Rough estimation of entry size in bytes + entry + |> Map.values() + |> Enum.map(&estimate_value_size/1) + |> Enum.sum() + # Add overhead for structure + |> Kernel.+(100) + end + + defp estimate_entry_size(_), do: 1024 + + defp estimate_value_size(nil), do: 4 + defp estimate_value_size(value) when is_binary(value), do: byte_size(value) + defp estimate_value_size(value) when is_integer(value), do: 8 + defp estimate_value_size(value) when is_float(value), do: 8 + defp estimate_value_size(value) when is_boolean(value), do: 1 + defp estimate_value_size(%DateTime{}), do: 8 + defp estimate_value_size(%Date{}), do: 4 + # Conservative estimate for complex types + defp estimate_value_size(_), do: 50 +end diff --git a/lib/aprsme/packet_consumer.ex b/lib/aprsme/packet_consumer.ex index 857a13f..6eaa2ce 100644 --- a/lib/aprsme/packet_consumer.ex +++ b/lib/aprsme/packet_consumer.ex @@ -100,9 +100,13 @@ defmodule Aprsme.PacketConsumer do {memory_before, _} = :erlang.statistics(:runtime) start_time = System.monotonic_time(:millisecond) + # Chunk size optimized for PostgreSQL work_mem=16MB + # With ~32KB per packet, we can fit ~500 packets in work_mem + chunk_size = Application.get_env(:aprsme, :packet_pipeline)[:batch_size] || 500 + results = packets - |> Enum.chunk_every(50) + |> Enum.chunk_every(chunk_size) |> Enum.map(&process_chunk/1) {success_count, error_count} = @@ -172,7 +176,19 @@ defmodule Aprsme.PacketConsumer do {valid_packets, invalid_packets} = Enum.split_with(packet_attrs, &valid_packet?/1) # Insert valid packets in batch - case Repo.insert_all(Aprsme.Packet, valid_packets, returning: [:id]) do + # Optimized for PostgreSQL with synchronous_commit=off + insert_opts = [ + # Don't return IDs for better performance + returning: false, + # Skip conflicts to avoid blocking + on_conflict: :nothing, + # Increased timeout for large batches + timeout: 60_000, + # Use placeholders for better performance with large batches + placeholders: length(valid_packets) > 100 + ] + + case Repo.insert_all(Aprsme.Packet, valid_packets, insert_opts) do {:error, error} -> Logger.error("Batch insert failed: #{inspect(error)}") {0, length(packets)}