Packages

snakepit

0.5.1
0.13.0 0.12.0 0.11.1 0.11.0 0.10.1 0.10.0 0.9.1 0.9.0 0.8.9 0.8.8 0.8.7 0.8.6 0.8.5 0.8.4 0.8.3 0.8.2 0.8.1 0.8.0 0.7.7 0.7.6 0.7.5 0.7.4 0.7.3 0.7.2 0.7.1 0.7.0 0.6.11 0.6.10 0.6.9 0.6.8 0.6.7 0.6.6 0.6.5 0.6.4 0.6.3 0.6.2 0.6.1 0.6.0 0.5.1 0.5.0 0.4.3 0.4.2 0.4.1 0.4.0 0.3.3 0.3.2 0.3.1 0.3.0 0.2.1 0.2.0 0.1.2 0.1.1 0.1.0

High-performance pooler and session manager for external language integrations. Supports Python, Node.js, Ruby, and more with gRPC streaming, session management, and production-ready process cleanup.

HexDocs

Current section

53 Versions

Jump to
Readme
53 Versions
3 Dependencies
Dependants
Files
Activity
Show whitespace

Compare versions

10 files changed
+923 additions
-48 deletions
  @@ -7,6 +7,70 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
7 7
8 8 ## [Unreleased]
9 9
10 + ## [0.5.1] - 2025-10-11
11 +
12 + ### Added
13 + - **Diagnostic Tools**
14 + - New `mix diagnose.scaling` task (613 lines) for comprehensive bottleneck analysis
15 + - Captures resource metrics (ports, processes, TCP connections, memory usage)
16 + - Enhanced error logging with port buffer drainage
17 +
18 + - **Configuration Enhancements**
19 + - Explicit gRPC port range constraint documentation and validation
20 + - Batched worker startup configuration (`startup_batch_size: 8`, `startup_batch_delay_ms: 750`)
21 + - Resource limit safeguards with `max_workers: 1000` hard limit
22 +
23 + ### Changed
24 + - **Worker Pool Scaling Improvements**
25 + - Pool now reliably scales to 250+ workers (previously limited to ~105)
26 + - Resolved thread explosion during concurrent startup (fixed "fork bomb" issue)
27 + - Dynamic port allocation using OS-assigned ports (port=0) eliminates port collision races
28 + - Batched worker startup prevents system resource exhaustion during concurrent initialization
29 +
30 + - **Performance Optimizations**
31 + - Aggressive thread limiting via environment variables for optimal pool-level parallelism:
32 + - `OPENBLAS_NUM_THREADS=1` (numpy/scipy)
33 + - `OMP_NUM_THREADS=1` (OpenMP)
34 + - `MKL_NUM_THREADS=1` (Intel MKL)
35 + - `NUMEXPR_NUM_THREADS=1` (NumExpr)
36 + - `GRPC_POLL_STRATEGY=poll` (single-threaded)
37 + - Increased GRPC server connection backlog to 512
38 + - Extended worker ready timeout to 30s for large pools
39 +
40 + - **Configuration Updates**
41 + - Increased `port_range` to 1000 (accommodates `max_workers`)
42 + - Enhanced configuration comments explaining each tuning parameter
43 + - Resource usage tracking during pool initialization
44 +
45 + ### Fixed
46 + - **Concurrent Startup Issues**
47 + - Fixed "Cannot fork" / EAGAIN errors from thread explosion during worker spawn
48 + - Eliminated port collision races with dynamic port allocation
49 + - Resolved fork bomb caused by Python scientific libraries spawning excessive threads (6,000+ threads from OpenBLAS, gRPC, MKL)
50 +
51 + - **Resource Management**
52 + - Better port binding error handling in Python gRPC server
53 + - Improved error diagnostics during pool initialization
54 + - Enhanced connection management in GRPC server
55 +
56 + ### Performance
57 + - Successfully tested with 250 workers (2.5x previous limit)
58 + - Startup time increases with pool size (~60s for 250 workers vs ~10s for 100 workers)
59 + - Eliminated port collision races and fork resource exhaustion
60 + - Dynamic port allocation provides reliable scaling
61 +
62 + ### Notes
63 + - Thread limiting optimizes for high concurrency with many small tasks
64 + - CPU-intensive workloads that perform heavy numerical computation within a single task may need different threading configuration
65 + - For computationally intensive per-task workloads, consider:
66 + - Workload-specific environment variables passed per task
67 + - Separate worker pools with different threading profiles
68 + - Dynamic thread limit adjustment based on task type
69 + - Allowing higher OpenBLAS threads but reducing max_workers accordingly
70 + - See commit dc67572 for detailed technical analysis and future considerations
71 +
72 + ---
73 +
10 74 ## [0.5.0] - 2025-10-10
11 75
12 76 ### Added
  @@ -401,6 +465,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
401 465 - Configurable pool sizes and timeouts
402 466 - Built-in bridge scripts for Python and JavaScript
403 467
468 + [0.5.1]: https://github.com/nshkrdotcom/snakepit/releases/tag/v0.5.1
404 469 [0.5.0]: https://github.com/nshkrdotcom/snakepit/releases/tag/v0.5.0
405 470 [0.4.3]: https://github.com/nshkrdotcom/snakepit/releases/tag/v0.4.3
406 471 [0.4.2]: https://github.com/nshkrdotcom/snakepit/releases/tag/v0.4.2
  @@ -8,7 +8,13 @@
8 8
9 9 [![CI](https://github.com/nshkrdotcom/snakepit/actions/workflows/ci.yml/badge.svg)](https://github.com/nshkrdotcom/snakepit/actions/workflows/ci.yml)
10 10 [![Hex Version](https://img.shields.io/hexpm/v/snakepit.svg)](https://hex.pm/packages/snakepit)
11 + [![Hex Docs](https://img.shields.io/badge/hex-docs-lightgreen.svg)](https://hexdocs.pm/snakepit)
12 + [![Downloads](https://img.shields.io/hexpm/dt/snakepit.svg)](https://hex.pm/packages/snakepit)
11 13 [![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](https://opensource.org/licenses/MIT)
14 + [![Erlang/OTP](https://img.shields.io/badge/Erlang%2FOTP-27%2B-red.svg)](https://www.erlang.org)
15 + [![Elixir](https://img.shields.io/badge/Elixir-1.18%2B-purple.svg)](https://elixir-lang.org)
16 + [![Last Commit](https://img.shields.io/github/last-commit/nshkrdotcom/snakepit)](https://github.com/nshkrdotcom/snakepit/commits/main)
17 + [![GitHub Stars](https://img.shields.io/github/stars/nshkrdotcom/snakepit?style=social)](https://github.com/nshkrdotcom/snakepit)
12 18
13 19 ## 🚀 What is Snakepit?
14 20
  @@ -25,6 +31,7 @@ Snakepit is a battle-tested Elixir library that provides a robust pooling system
25 31 ## 📋 Table of Contents
26 32
27 33 - [Breaking Changes (v0.5.0)](#️-breaking-changes-v050)
34 + - [What's New in v0.5.1](#whats-new-in-v051)
28 35 - [What's New in v0.5](#whats-new-in-v05)
29 36 - [Quick Start](#quick-start)
30 37 - [Installation](#installation)
  @@ -95,6 +102,35 @@ For **non-DSPex users**, if you're using these classes directly:
95 102
96 103 ---
97 104
105 + ## 🆕 What's New in v0.5.1
106 +
107 + ### Worker Pool Scaling Enhancements
108 + - **Fixed worker pool scaling limits** - Pool now reliably scales to 250+ workers (previously limited to ~105)
109 + - **Resolved thread explosion during concurrent startup** - Fixed "fork bomb" caused by Python scientific libraries spawning excessive threads
110 + - **Dynamic port allocation** - Workers now use OS-assigned ports (port=0) eliminating port collision races
111 + - **Batched worker startup** - Configurable batch size and delay prevents system resource exhaustion
112 + - **Enhanced resource limits** - Added max_workers safeguard (1000) with comprehensive warnings
113 + - **New diagnostic tools** - Added `mix diagnose.scaling` task for bottleneck analysis
114 +
115 + ### Configuration Improvements
116 + - **Aggressive thread limiting** - Set `OPENBLAS_NUM_THREADS=1`, `OMP_NUM_THREADS=1`, `MKL_NUM_THREADS=1` for optimal pool-level parallelism
117 + - **Batched startup configuration** - `startup_batch_size: 8`, `startup_batch_delay_ms: 750`
118 + - **Increased resource limits** - Extended `port_range: 1000`, GRPC backlog: 512, worker timeout: 30s
119 + - **Explicit port range constraints** - Added configuration documentation and validation
120 +
121 + ### Performance & Reliability
122 + - **Successfully tested with 250 workers** - Validated reliable operation at 2.5x previous limit
123 + - **Eliminated port collision races** - Dynamic port allocation prevents startup failures
124 + - **Improved error diagnostics** - Better logging and resource tracking during pool initialization
125 + - **Enhanced GRPC server** - Better port binding error handling and connection management
126 +
127 + ### Notes
128 + - Startup time increases with large pools (~60s for 250 workers vs ~10s for 100 workers)
129 + - Thread limiting optimizes for high concurrency; CPU-intensive tasks per worker may need adjustment
130 + - See commit dc67572 for detailed technical analysis and future considerations
131 +
132 + ---
133 +
98 134 ## 🆕 What's New in v0.5.0
99 135
100 136 ### Breaking Changes
  @@ -231,7 +267,7 @@ For **non-DSPex users**, if you're using these classes directly:
231 267 # In your mix.exs
232 268 def deps do
233 269 [
234 - {:snakepit, "~> 0.5.0"}
270 + {:snakepit, "~> 0.5.1"}
235 271 ]
236 272 end
237 273
  @@ -266,7 +302,7 @@ end)
266 302 ```elixir
267 303 def deps do
268 304 [
269 - {:snakepit, "~> 0.5.0"}
305 + {:snakepit, "~> 0.5.1"}
270 306 ]
271 307 end
272 308 ```
  @@ -2066,7 +2102,15 @@ Snakepit is released under the MIT License. See the [LICENSE](https://github.com
2066 2102
2067 2103 ## 📊 Development Status
2068 2104
2069 - **v0.5.0 (Current Release)**
2105 + **v0.5.1 (Current Release)**
2106 + - **Worker pool scaling fixed** - Reliably scales to 250+ workers (previously ~105 limit)
2107 + - **Thread explosion resolved** - Fixed fork bomb from Python scientific libraries
2108 + - **Dynamic port allocation** - OS-assigned ports eliminate collision races
2109 + - **Batched startup** - Configurable batching prevents resource exhaustion
2110 + - **New diagnostic tools** - Added `mix diagnose.scaling` for bottleneck analysis
2111 + - **Enhanced configuration** - Thread limiting and resource management improvements
2112 +
2113 + **v0.5.0**
2070 2114 - **DSPy integration removed** - Clean architecture separation achieved
2071 2115 - **Test infrastructure enhanced** - 89% increase in test coverage (27→51 tests)
2072 2116 - **Code cleanup complete** - ~1,500 LOC of dead code removed
  @@ -8,7 +8,7 @@
8 8 {<<"gRPC Streaming Guide">>,
9 9 <<"https://hexdocs.pm/snakepit/README_GRPC.html">>}]}.
10 10 {<<"name">>,<<"snakepit">>}.
11 - {<<"version">>,<<"0.5.0">>}.
11 + {<<"version">>,<<"0.5.1">>}.
12 12 {<<"description">>,
13 13 <<"High-performance pooler and session manager for external language integrations.\nSupports Python, Node.js, Ruby, and more with gRPC streaming, session management,\nand production-ready process cleanup.">>}.
14 14 {<<"elixir">>,<<"~> 1.18">>}.
  @@ -37,10 +37,12 @@
37 37 <<"lib/snakepit/pool/application_cleanup.ex">>,
38 38 <<"lib/snakepit/pool/process_registry.ex">>,
39 39 <<"lib/snakepit/pool/worker_starter.ex">>,<<"lib/snakepit/telemetry.ex">>,
40 - <<"lib/snakepit/run_id.ex">>,<<"lib/snakepit.ex">>,<<"priv/proto">>,
41 - <<"priv/proto/README.md">>,<<"priv/proto/snakepit_bridge.proto">>,
42 - <<"priv/python/generate_proto.py">>,<<"priv/python/grpc_server.py">>,
43 - <<"priv/python/setup.py">>,<<"priv/python/snakepit_bridge_pb2.py">>,
40 + <<"lib/snakepit/run_id.ex">>,<<"lib/mix">>,<<"lib/mix/tasks">>,
41 + <<"lib/mix/tasks/diagnose.scaling.ex">>,<<"lib/snakepit.ex">>,
42 + <<"priv/proto">>,<<"priv/proto/README.md">>,
43 + <<"priv/proto/snakepit_bridge.proto">>,<<"priv/python/generate_proto.py">>,
44 + <<"priv/python/grpc_server.py">>,<<"priv/python/setup.py">>,
45 + <<"priv/python/snakepit_bridge_pb2.py">>,
44 46 <<"priv/python/snakepit_bridge_pb2_grpc.py">>,
45 47 <<"priv/python/requirements.txt">>,<<"priv/python/snakepit_bridge">>,
46 48 <<"priv/python/snakepit_bridge/__init__.py">>,
  @@ -0,0 +1,610 @@
1 + defmodule Mix.Tasks.Diagnose.Scaling do
2 + @moduledoc """
3 + Comprehensive scaling diagnostics to find the bottleneck preventing >105 workers.
4 +
5 + Usage:
6 + mix diagnose.scaling
7 + """
8 +
9 + use Mix.Task
10 + require Logger
11 +
12 + @shortdoc "Find scaling bottleneck preventing >105 workers"
13 +
14 + @impl Mix.Task
15 + def run(_args) do
16 + Mix.Task.run("app.start")
17 +
18 + IO.puts("\n" <> String.duplicate("=", 80))
19 + IO.puts("🔬 SNAKEPIT SCALING DIAGNOSTICS")
20 + IO.puts(String.duplicate("=", 80) <> "\n")
21 +
22 + # Run all diagnostics
23 + check_erlang_limits()
24 + check_erlang_process_limits()
25 + check_dets_contention()
26 + check_python_spawn_rate()
27 + check_grpc_limits()
28 + find_exact_breaking_point()
29 +
30 + IO.puts("\n" <> String.duplicate("=", 80))
31 + IO.puts("✅ DIAGNOSTICS COMPLETE")
32 + IO.puts(String.duplicate("=", 80) <> "\n")
33 + end
34 +
35 + defp check_erlang_limits do
36 + IO.puts("📊 TEST 1: ERLANG PORT LIMITS")
37 + IO.puts(String.duplicate("-", 80))
38 +
39 + port_limit = :erlang.system_info(:port_limit)
40 + port_count = :erlang.system_info(:port_count)
41 + port_available = port_limit - port_count
42 +
43 + IO.puts("Port Limit: #{format_number(port_limit)}")
44 + IO.puts("Current Ports: #{format_number(port_count)}")
45 + IO.puts("Available: #{format_number(port_available)}")
46 + IO.puts("Usage: #{Float.round(port_count / port_limit * 100, 2)}%")
47 +
48 + # Estimate ports per worker
49 + if port_count > 0 do
50 + # Try to count actual workers
51 + worker_count = count_workers()
52 +
53 + if worker_count > 0 do
54 + ports_per_worker = port_count / worker_count
55 + IO.puts("\nEstimated ports per worker: #{Float.round(ports_per_worker, 2)}")
56 + max_workers = floor(port_limit / ports_per_worker)
57 + IO.puts("Theoretical max workers: #{max_workers}")
58 +
59 + if max_workers < 110 do
60 + IO.puts("\n⚠️ WARNING: Port limit would prevent 110 workers!")
61 +
62 + IO.puts(
63 + " At #{Float.round(ports_per_worker, 2)} ports/worker, you can only spawn ~#{max_workers} workers"
64 + )
65 + end
66 + end
67 + end
68 +
69 + # Check if we're close to the limit
70 + if port_available < 1000 do
71 + IO.puts("\n🔴 CRITICAL: Less than 1000 ports available!")
72 + else
73 + if port_count / port_limit > 0.8 do
74 + IO.puts("\n⚠️ WARNING: Using >80% of port limit")
75 + else
76 + IO.puts("\n✅ Port usage looks healthy")
77 + end
78 + end
79 +
80 + IO.puts("\n")
81 + end
82 +
83 + defp check_erlang_process_limits do
84 + IO.puts("📊 TEST 2: ERLANG PROCESS LIMITS")
85 + IO.puts(String.duplicate("-", 80))
86 +
87 + process_limit = :erlang.system_info(:process_limit)
88 + process_count = :erlang.system_info(:process_count)
89 + process_available = process_limit - process_count
90 +
91 + IO.puts("Process Limit: #{format_number(process_limit)}")
92 + IO.puts("Current Procs: #{format_number(process_count)}")
93 + IO.puts("Available: #{format_number(process_available)}")
94 + IO.puts("Usage: #{Float.round(process_count / process_limit * 100, 2)}%")
95 +
96 + # Count snakepit-related processes
97 + snakepit_procs =
98 + Process.list()
99 + |> Enum.count(fn pid ->
100 + case Process.info(pid, :registered_name) do
101 + {:registered_name, name} ->
102 + name_str = to_string(name)
103 + String.contains?(name_str, "snakepit") || String.contains?(name_str, "Snakepit")
104 +
105 + _ ->
106 + # Check initial call
107 + case Process.info(pid, :initial_call) do
108 + {:initial_call, {mod, _fun, _arity}} ->
109 + mod_str = to_string(mod)
110 + String.contains?(mod_str, "Snakepit")
111 +
112 + _ ->
113 + false
114 + end
115 + end
116 + end)
117 +
118 + IO.puts("\nSnakepit processes: #{snakepit_procs}")
119 +
120 + worker_count = count_workers()
121 +
122 + if worker_count > 0 do
123 + procs_per_worker = process_count / worker_count
124 + IO.puts("Estimated procs per worker: #{Float.round(procs_per_worker, 2)}")
125 + max_workers = floor(process_limit / procs_per_worker)
126 + IO.puts("Theoretical max workers: #{max_workers}")
127 +
128 + if max_workers < 110 do
129 + IO.puts("\n⚠️ WARNING: Process limit would prevent 110 workers!")
130 + end
131 + end
132 +
133 + if process_count / process_limit > 0.8 do
134 + IO.puts("\n⚠️ WARNING: Using >80% of process limit")
135 + else
136 + IO.puts("\n✅ Process usage looks healthy")
137 + end
138 +
139 + IO.puts("\n")
140 + end
141 +
142 + defp check_dets_contention do
143 + IO.puts("📊 TEST 3: DETS LOCK CONTENTION")
144 + IO.puts(String.duplicate("-", 80))
145 +
146 + # Check if DETS table exists
147 + dets_table = :snakepit_process_registry_dets
148 +
149 + case :dets.info(dets_table) do
150 + :undefined ->
151 + IO.puts("⚠️ DETS table not initialized yet")
152 +
153 + _info ->
154 + size = :dets.info(dets_table, :size)
155 + IO.puts("DETS table size: #{size} entries")
156 +
157 + # Test concurrent write performance
158 + IO.puts("\nTesting concurrent DETS writes (simulating 110 workers)...")
159 +
160 + {time_us, results} =
161 + :timer.tc(fn ->
162 + 1..110
163 + |> Task.async_stream(
164 + fn i ->
165 + start = System.monotonic_time(:microsecond)
166 +
167 + result =
168 + :dets.insert(
169 + dets_table,
170 + {"test_worker_#{i}_#{:rand.uniform(999_999)}",
171 + %{pid: self(), started_at: System.system_time(:second)}}
172 + )
173 +
174 + elapsed = System.monotonic_time(:microsecond) - start
175 + {result, elapsed}
176 + end,
177 + max_concurrency: 110,
178 + timeout: 30_000
179 + )
180 + |> Enum.to_list()
181 + end)
182 +
183 + time_s = time_us / 1_000_000
184 +
185 + successes =
186 + Enum.count(results, fn
187 + {:ok, {:ok, _}} -> true
188 + _ -> false
189 + end)
190 +
191 + failures = 110 - successes
192 +
193 + # Get individual write times
194 + write_times =
195 + Enum.map(results, fn
196 + {:ok, {_result, elapsed}} -> elapsed
197 + _ -> nil
198 + end)
199 + |> Enum.reject(&is_nil/1)
200 +
201 + if length(write_times) > 0 do
202 + avg_write_us = Enum.sum(write_times) / length(write_times)
203 + max_write_us = Enum.max(write_times)
204 + min_write_us = Enum.min(write_times)
205 +
206 + IO.puts("\nResults:")
207 + IO.puts(" Total time: #{Float.round(time_s, 3)}s")
208 + IO.puts(" Successes: #{successes}/110")
209 + IO.puts(" Failures: #{failures}/110")
210 + IO.puts(" Avg write time: #{Float.round(avg_write_us / 1000, 2)}ms")
211 + IO.puts(" Min write time: #{Float.round(min_write_us / 1000, 2)}ms")
212 + IO.puts(" Max write time: #{Float.round(max_write_us / 1000, 2)}ms")
213 +
214 + # Diagnosis
215 + if time_s > 5.0 do
216 + IO.puts("\n🔴 CRITICAL: DETS serialization is likely your bottleneck!")
217 + IO.puts(" 110 concurrent writes took #{Float.round(time_s, 1)}s")
218 + IO.puts(" DETS locks on every write, causing extreme serialization")
219 + IO.puts("\n💡 SOLUTION: Replace DETS with ETS or remove registry entirely")
220 + else
221 + if time_s > 2.0 do
222 + IO.puts("\n⚠️ WARNING: DETS writes are slow (#{Float.round(time_s, 1)}s)")
223 + IO.puts(" This could contribute to timeouts during worker startup")
224 + else
225 + IO.puts("\n✅ DETS performance acceptable (#{Float.round(time_s, 3)}s)")
226 + end
227 + end
228 + end
229 + end
230 +
231 + IO.puts("\n")
232 + end
233 +
234 + defp check_python_spawn_rate do
235 + IO.puts("📊 TEST 4: PYTHON PROCESS SPAWN RATE")
236 + IO.puts(String.duplicate("-", 80))
237 +
238 + IO.puts("Testing Python process spawn rate (110 processes)...")
239 +
240 + # Test python3 spawn rate
241 + {time_us, {output, exit_code}} =
242 + :timer.tc(fn ->
243 + System.cmd(
244 + "sh",
245 + [
246 + "-c",
247 + """
248 + for i in {1..110}; do
249 + python3 -c "import sys; sys.exit(0)" &
250 + done
251 + wait
252 + """
253 + ],
254 + stderr_to_stdout: true
255 + )
256 + end)
257 +
258 + time_s = time_us / 1_000_000
259 +
260 + IO.puts("Time to spawn 110 Python processes: #{Float.round(time_s, 2)}s")
261 + IO.puts("Exit code: #{exit_code}")
262 +
263 + if exit_code != 0 do
264 + IO.puts("Output: #{String.slice(output, 0, 500)}")
265 + end
266 +
267 + if time_s > 30.0 do
268 + IO.puts("\n🔴 CRITICAL: Python spawn rate is very slow!")
269 + IO.puts(" System may be CPU or fork() constrained")
270 + else
271 + if time_s > 10.0 do
272 + IO.puts("\n⚠️ WARNING: Python spawn rate is slow (#{Float.round(time_s, 1)}s)")
273 + else
274 + IO.puts("\n✅ Python spawn rate looks good (#{Float.round(time_s, 2)}s)")
275 + end
276 + end
277 +
278 + IO.puts("\n")
279 + end
280 +
281 + defp check_grpc_limits do
282 + IO.puts("📊 TEST 5: GRPC CONNECTION LIMITS")
283 + IO.puts(String.duplicate("-", 80))
284 +
285 + # Check for Ranch (used by GRPC.Server)
286 + try do
287 + ranch_info = :ranch.info()
288 + IO.puts("Ranch info:")
289 + IO.inspect(ranch_info, limit: :infinity, pretty: true)
290 + rescue
291 + _ ->
292 + IO.puts("Ranch not available or not running")
293 + end
294 +
295 + # Count TCP connections
296 + case System.cmd("sh", ["-c", "ss -tan | grep ESTAB | wc -l"], stderr_to_stdout: true) do
297 + {output, 0} ->
298 + tcp_count = String.trim(output) |> String.to_integer()
299 + IO.puts("\nEstablished TCP connections: #{tcp_count}")
300 +
301 + worker_count = count_workers()
302 +
303 + if worker_count > 0 do
304 + tcp_per_worker = tcp_count / worker_count
305 + IO.puts("TCP connections per worker: #{Float.round(tcp_per_worker, 2)}")
306 + end
307 +
308 + {output, _} ->
309 + IO.puts("Could not count TCP connections: #{output}")
310 + end
311 +
312 + # Check gRPC port status
313 + case System.cmd("sh", ["-c", "ss -tlnp | grep -E ':(50051|50052)'"], stderr_to_stdout: true) do
314 + {output, 0} ->
315 + IO.puts("\ngRPC port status:")
316 + IO.puts(output)
317 +
318 + {_output, _} ->
319 + IO.puts("\ngRPC ports not listening or ss command failed")
320 + end
321 +
322 + IO.puts("\n")
323 + end
324 +
325 + defp find_exact_breaking_point do
326 + IO.puts("📊 TEST 6: BINARY SEARCH FOR EXACT BREAKING POINT")
327 + IO.puts(String.duplicate("-", 80))
328 + IO.puts("\n⚠️ This test spawns real workers and may fail!")
329 + IO.puts("Testing from 100 to 400 workers in increments...\n")
330 +
331 + # Capture baseline metrics
332 + baseline = capture_metrics()
333 + IO.puts("Baseline metrics:")
334 + print_metrics(baseline)
335 +
336 + # Test increasing worker counts
337 + results =
338 + for count <- [100, 250, 400, 500] do
339 + IO.puts("\n" <> String.duplicate("-", 60))
340 + IO.puts("🔬 Testing #{count} workers...")
341 +
342 + # Small delay between tests
343 + :timer.sleep(2000)
344 +
345 + before_metrics = capture_metrics()
346 +
347 + # Try to start a worker
348 + worker_id = "diagnostic_#{count}_#{:rand.uniform(999_999)}"
349 +
350 + start_time = System.monotonic_time(:millisecond)
351 +
352 + result =
353 + try do
354 + # Check if WorkerSupervisor exists
355 + case Process.whereis(Snakepit.Pool.WorkerSupervisor) do
356 + nil ->
357 + {:error, :supervisor_not_running}
358 +
359 + _pid ->
360 + Snakepit.Pool.WorkerSupervisor.start_worker(
361 + worker_id,
362 + Snakepit.GRPCWorker,
363 + Snakepit.Adapters.GRPCPython,
364 + Snakepit.Pool
365 + )
366 + end
367 + rescue
368 + e -> {:error, e}
369 + catch
370 + :exit, reason -> {:error, {:exit, reason}}
371 + end
372 +
373 + elapsed = System.monotonic_time(:millisecond) - start_time
374 +
375 + after_metrics = capture_metrics()
376 + delta = calculate_delta(before_metrics, after_metrics)
377 +
378 + status =
379 + case result do
380 + {:ok, _pid} -> :success
381 + {:error, reason} -> {:failed, reason}
382 + end
383 +
384 + # Clean up if successful
385 + if status == :success do
386 + try do
387 + Snakepit.Pool.WorkerSupervisor.stop_worker(worker_id)
388 + rescue
389 + _ -> :ok
390 + end
391 + end
392 +
393 + result_data = %{
394 + count: count,
395 + status: status,
396 + elapsed_ms: elapsed,
397 + before: before_metrics,
398 + after: after_metrics,
399 + delta: delta
400 + }
401 +
402 + print_test_result(result_data)
403 +
404 + result_data
405 + end
406 +
407 + # Analyze results
408 + IO.puts("\n" <> String.duplicate("=", 60))
409 + IO.puts("📈 ANALYSIS")
410 + IO.puts(String.duplicate("=", 60))
411 +
412 + first_failure =
413 + Enum.find(results, fn r ->
414 + match?({:failed, _}, r.status)
415 + end)
416 +
417 + case first_failure do
418 + nil ->
419 + IO.puts("\n✅ All tests passed up to 500 workers!")
420 + IO.puts(" The bottleneck may be higher than 500.")
421 +
422 + %{count: count, status: {:failed, reason}} ->
423 + IO.puts("\n🔴 FIRST FAILURE AT #{count} WORKERS")
424 + IO.puts("\nReason: #{inspect(reason, pretty: true)}")
425 +
426 + # Check what metric grew most
427 + analyze_metric_growth(results, count)
428 + end
429 +
430 + IO.puts("\n")
431 + end
432 +
433 + defp capture_metrics do
434 + %{
435 + erlang_ports: :erlang.system_info(:port_count),
436 + erlang_procs: :erlang.system_info(:process_count),
437 + tcp_connections: count_tcp_connections(),
438 + worker_count: count_workers(),
439 + memory_bytes: :erlang.memory(:total)
440 + }
441 + end
442 +
443 + defp calculate_delta(before, after_metrics) do
444 + %{
445 + ports: after_metrics.erlang_ports - before.erlang_ports,
446 + procs: after_metrics.erlang_procs - before.erlang_procs,
447 + tcp: after_metrics.tcp_connections - before.tcp_connections,
448 + workers: after_metrics.worker_count - before.worker_count,
449 + memory_mb: (after_metrics.memory_bytes - before.memory_bytes) / (1024 * 1024)
450 + }
451 + end
452 +
453 + defp print_metrics(metrics) do
454 + IO.puts(" Erlang ports: #{metrics.erlang_ports}")
455 + IO.puts(" Erlang procs: #{metrics.erlang_procs}")
456 + IO.puts(" TCP conns: #{metrics.tcp_connections}")
457 + IO.puts(" Workers: #{metrics.worker_count}")
458 + IO.puts(" Memory: #{Float.round(metrics.memory_bytes / (1024 * 1024), 1)} MB")
459 + end
460 +
461 + defp print_test_result(result) do
462 + case result.status do
463 + :success ->
464 + IO.puts("✅ SUCCESS (#{result.elapsed_ms}ms)")
465 +
466 + {:failed, reason} ->
467 + IO.puts("🔴 FAILED (#{result.elapsed_ms}ms)")
468 + IO.puts(" Reason: #{inspect(reason)}")
469 + end
470 +
471 + IO.puts("\nMetric deltas:")
472 + IO.puts(" Ports: #{format_delta(result.delta.ports)}")
473 + IO.puts(" Procs: #{format_delta(result.delta.procs)}")
474 + IO.puts(" TCP: #{format_delta(result.delta.tcp)}")
475 + IO.puts(" Workers: #{format_delta(result.delta.workers)}")
476 + IO.puts(" Memory: #{format_delta(result.delta.memory_mb)} MB")
477 + end
478 +
479 + defp analyze_metric_growth(results, failure_count) do
480 + successful = Enum.filter(results, fn r -> r.status == :success end)
481 +
482 + if length(successful) > 0 do
483 + # Calculate average resource usage per worker from successful runs
484 + avg_delta = %{
485 + ports: avg(Enum.map(successful, & &1.delta.ports)),
486 + procs: avg(Enum.map(successful, & &1.delta.procs)),
487 + tcp: avg(Enum.map(successful, & &1.delta.tcp))
488 + }
489 +
490 + IO.puts("\nAverage resource usage per worker (from successful runs):")
491 + IO.puts(" Ports per worker: #{Float.round(avg_delta.ports, 2)}")
492 + IO.puts(" Procs per worker: #{Float.round(avg_delta.procs, 2)}")
493 + IO.puts(" TCP per worker: #{Float.round(avg_delta.tcp, 2)}")
494 +
495 + # Extrapolate to limits
496 + port_limit = :erlang.system_info(:port_limit)
497 + proc_limit = :erlang.system_info(:process_limit)
498 +
499 + max_by_ports = floor(port_limit / avg_delta.ports)
500 + max_by_procs = floor(proc_limit / avg_delta.procs)
501 +
502 + IO.puts("\n📊 PROJECTED MAXIMUMS:")
503 + IO.puts(" Max by ports: #{max_by_ports} workers")
504 + IO.puts(" Max by processes: #{max_by_procs} workers")
505 +
506 + bottleneck =
507 + cond do
508 + max_by_ports < failure_count ->
509 + IO.puts("\n🔴 SMOKING GUN: ERLANG PORT LIMIT")
510 +
511 + IO.puts(
512 + " At #{Float.round(avg_delta.ports, 2)} ports/worker, you hit the #{port_limit} port limit at ~#{max_by_ports} workers"
513 + )
514 +
515 + :ports
516 +
517 + max_by_procs < failure_count ->
518 + IO.puts("\n🔴 SMOKING GUN: ERLANG PROCESS LIMIT")
519 +
520 + IO.puts(
521 + " At #{Float.round(avg_delta.procs, 2)} procs/worker, you hit the #{proc_limit} process limit at ~#{max_by_procs} workers"
522 + )
523 +
524 + :procs
525 +
526 + true ->
527 + IO.puts("\n🤔 UNCLEAR: Neither port nor process limit explains the failure")
528 + IO.puts(" Check the error message above for clues")
529 + :unknown
530 + end
531 +
532 + # Suggest solutions
533 + suggest_solutions(bottleneck, avg_delta)
534 + end
535 + end
536 +
537 + defp suggest_solutions(:ports, avg_delta) do
538 + IO.puts("\n💡 SOLUTIONS:")
539 + IO.puts(" 1. Increase Erlang port limit in vm.args:")
540 + IO.puts(" +Q 1000000")
541 + IO.puts(" 2. Each worker uses #{Float.round(avg_delta.ports, 1)} ports - investigate why:")
542 + IO.puts(" - 1 Port for stdin/stdout")
543 + IO.puts(" - 1 TCP socket for gRPC")
544 + IO.puts(" - Additional ports for...?")
545 + IO.puts(" 3. Consider reducing port usage per worker")
546 + end
547 +
548 + defp suggest_solutions(:procs, avg_delta) do
549 + IO.puts("\n💡 SOLUTIONS:")
550 + IO.puts(" 1. Increase Erlang process limit in vm.args:")
551 + IO.puts(" +P 5000000")
552 +
553 + IO.puts(
554 + " 2. Each worker uses #{Float.round(avg_delta.procs, 1)} processes - investigate supervision tree depth"
555 + )
556 +
557 + IO.puts(" 3. Consider flattening the supervision tree")
558 + end
559 +
560 + defp suggest_solutions(:unknown, _avg_delta) do
561 + IO.puts("\n💡 NEXT STEPS:")
562 + IO.puts(" 1. Check the error message for clues")
563 + IO.puts(" 2. Look for timeouts in worker startup")
564 + IO.puts(" 3. Check DETS test results above")
565 + IO.puts(" 4. Monitor system resources (CPU, memory, file descriptors)")
566 + end
567 +
568 + defp count_workers do
569 + try do
570 + case Process.whereis(Snakepit.Pool.Registry) do
571 + nil ->
572 + 0
573 +
574 + _pid ->
575 + Registry.count(Snakepit.Pool.Registry)
576 + end
577 + rescue
578 + _ -> 0
579 + end
580 + end
581 +
582 + defp count_tcp_connections do
583 + case System.cmd("sh", ["-c", "ss -tan | grep ESTAB | wc -l"], stderr_to_stdout: true) do
584 + {output, 0} ->
585 + output |> String.trim() |> String.to_integer()
586 +
587 + _ ->
588 + 0
589 + end
590 + end
591 +
592 + defp avg(list) when length(list) > 0 do
593 + Enum.sum(list) / length(list)
594 + end
595 +
596 + defp avg(_), do: 0.0
597 +
598 + defp format_number(num) when num >= 1_000_000 do
599 + "#{Float.round(num / 1_000_000, 2)}M"
600 + end
601 +
602 + defp format_number(num) when num >= 1_000 do
603 + "#{Float.round(num / 1_000, 1)}K"
604 + end
605 +
606 + defp format_number(num), do: to_string(num)
607 +
608 + defp format_delta(num) when num >= 0, do: "+#{Float.round(num * 1.0, 2)}"
609 + defp format_delta(num), do: Float.round(num * 1.0, 2) |> to_string()
610 + end
  @@ -161,37 +161,18 @@ defmodule Snakepit.Adapters.GRPCPython do
161 161 @doc """
162 162 Get the gRPC port for this adapter instance.
163 163
164 - CRITICAL FIX: Uses atomic counter instead of random allocation to prevent
165 - port collisions during concurrent worker startup (birthday paradox issue).
166 - With 48 concurrent workers and 100 random ports, collision probability ≈ 90%.
164 + ROBUST FIX: Use port 0 to let the OS dynamically assign an available port.
165 + This completely eliminates:
166 + - Port collision races
167 + - TIME_WAIT conflicts
168 + - Manual port range management
169 + - Port leak tracking
170 +
171 + Python will bind to an OS-assigned port and report it back via GRPC_READY.
167 172 """
168 173 def get_port do
169 - config = Application.get_env(:snakepit, :grpc_config, %{})
170 - # Start at 50052 to avoid Elixir server
171 - base_port = Map.get(config, :base_port, 50052)
172 - port_range = Map.get(config, :port_range, 100)
173 -
174 - # Atomic counter ensures sequential port assignment with zero collisions
175 - counter = :atomics.add_get(get_port_counter(), 1, 1)
176 - # Wrap around to reuse ports after workers terminate
177 - offset = rem(counter, port_range)
178 - base_port + offset
179 - end
180 -
181 - defp get_port_counter do
182 - # Lazy initialization of atomic counter (shared across all adapter instances)
183 - # Using persistent_term for process-independent storage
184 - case :persistent_term.get({__MODULE__, :port_counter}, nil) do
185 - nil ->
186 - # Create new atomic counter
187 - counter = :atomics.new(1, signed: false)
188 - :atomics.put(counter, 1, 0)
189 - :persistent_term.put({__MODULE__, :port_counter}, counter)
190 - counter
191 -
192 - counter ->
193 - counter
194 - end
174 + # Port 0 = "OS, please assign me any available port"
175 + 0
195 176 end
196 177
197 178 @doc """
Loading more files…