Packages
snakepit
0.1.0
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.
Current section
Files
Jump to
Current section
Files
README.md
# Snakepit 🐍
<div align="center">
<img src="img/snakepit-logo.svg" alt="Snakepit Logo" width="200" height="200">
</div>
> A high-performance, generalized process pooler and session manager for external language integrations in Elixir
[](https://github.com/nshkrdotcom/snakepit/actions/workflows/ci.yml)
[](https://hex.pm/packages/snakepit)
[](https://opensource.org/licenses/MIT)
## 🚀 What is Snakepit?
Snakepit is a battle-tested Elixir library that provides a robust pooling system for managing external processes (Python, Node.js, Ruby, R, etc.). Born from the need for reliable ML/AI integrations, it offers:
- **Lightning-fast concurrent initialization** - 1000x faster than sequential approaches
- **Session-based execution** with automatic worker affinity
- **Adapter pattern** for any external language/runtime
- **Built on OTP primitives** - DynamicSupervisor, Registry, GenServer
- **Production-ready** with telemetry, health checks, and graceful shutdowns
## 📋 Table of Contents
- [Quick Start](#-quick-start)
- [Installation](#-installation)
- [Core Concepts](#-core-concepts)
- [Configuration](#-configuration)
- [Usage Examples](#-usage-examples)
- [Built-in Adapters](#-built-in-adapters)
- [Creating Custom Adapters](#-creating-custom-adapters)
- [Session Management](#-session-management)
- [Monitoring & Telemetry](#-monitoring--telemetry)
- [Architecture Deep Dive](#-architecture-deep-dive)
- [Performance](#-performance)
- [Troubleshooting](#-troubleshooting)
- [Contributing](#-contributing)
## 🏃 Quick Start
```elixir
# In your mix.exs
def deps do
[
{:snakepit, "~> 0.0.1"}
]
end
# Configure and start
Application.put_env(:snakepit, :pooling_enabled, true)
Application.put_env(:snakepit, :adapter_module, Snakepit.Adapters.GenericPython)
Application.put_env(:snakepit, :pool_config, %{pool_size: 4})
{:ok, _} = Application.ensure_all_started(:snakepit)
# Execute commands
{:ok, result} = Snakepit.execute("ping", %{test: true})
{:ok, result} = Snakepit.execute("compute", %{operation: "add", a: 5, b: 3})
# Session-based execution (maintains state)
{:ok, result} = Snakepit.execute_in_session("user_123", "echo", %{message: "hello"})
```
## 📦 Installation
### Hex Package
```elixir
def deps do
[
{:snakepit, "~> 0.0.1"}
]
end
```
### GitHub (Latest)
```elixir
def deps do
[
{:snakepit, github: "nshkrdotcom/snakepit"}
]
end
```
### Requirements
- Elixir 1.18+
- Erlang/OTP 27+
- External runtime (Python 3.8+, Node.js 16+, etc.) depending on adapter
## 🎯 Core Concepts
### 1. **Adapters**
Adapters define how Snakepit communicates with external processes. They specify:
- The runtime executable (python3, node, ruby, etc.)
- The bridge script to execute
- Supported commands and validation
- Request/response transformations
### 2. **Workers**
Each worker is a GenServer that:
- Owns one external process via Erlang Port
- Handles request/response communication
- Manages health checks and metrics
- Auto-restarts on crashes
### 3. **Pool**
The pool manager:
- Starts workers concurrently on initialization
- Routes requests to available workers
- Handles queueing when all workers are busy
- Supports session affinity for stateful operations
### 4. **Sessions**
Sessions provide:
- State persistence across requests
- Worker affinity (same session prefers same worker)
- TTL-based expiration
- Centralized storage in ETS
## ⚙️ Configuration
### Basic Configuration
```elixir
# config/config.exs
config :snakepit,
pooling_enabled: true,
adapter_module: Snakepit.Adapters.GenericPython,
pool_config: %{
pool_size: 8 # Default: System.schedulers_online() * 2
}
```
### Advanced Configuration
```elixir
config :snakepit,
# Pool settings
pooling_enabled: true,
pool_config: %{
pool_size: 16
},
# Adapter
adapter_module: MyApp.CustomAdapter,
# Timeouts (milliseconds)
pool_startup_timeout: 10_000, # Max time for worker initialization
pool_queue_timeout: 5_000, # Max time in request queue
worker_init_timeout: 20_000, # Max time for worker to respond to init
worker_health_check_interval: 30_000, # Health check frequency
worker_shutdown_grace_period: 2_000, # Grace period for shutdown
# Cleanup settings
cleanup_retry_interval: 100, # Retry interval for cleanup
cleanup_max_retries: 10, # Max cleanup retries
# Queue management
pool_max_queue_size: 1000 # Max queued requests before rejection
```
### Runtime Configuration
```elixir
# Override configuration at runtime
Application.put_env(:snakepit, :adapter_module, Snakepit.Adapters.GenericJavaScript)
Application.stop(:snakepit)
Application.start(:snakepit)
```
## 📖 Usage Examples
### Basic Stateless Execution
```elixir
# Simple computation
{:ok, %{"result" => 8}} = Snakepit.execute("compute", %{
operation: "add",
a: 5,
b: 3
})
# With timeout
{:ok, result} = Snakepit.execute("long_running_task", %{data: "..."}, timeout: 60_000)
# Error handling
case Snakepit.execute("risky_operation", %{}) do
{:ok, result} -> handle_success(result)
{:error, :worker_timeout} -> handle_timeout()
{:error, :pool_saturated} -> handle_overload()
{:error, reason} -> handle_error(reason)
end
```
### Session-Based Execution
```elixir
# Create a session and maintain state
session_id = "user_#{user.id}"
# First request - initializes session
{:ok, _} = Snakepit.execute_in_session(session_id, "initialize", %{
user_id: user.id,
preferences: user.preferences
})
# Subsequent requests use same worker when possible
{:ok, recommendations} = Snakepit.execute_in_session(session_id, "get_recommendations", %{
category: "books"
})
# Session data persists across requests
{:ok, history} = Snakepit.execute_in_session(session_id, "get_history", %{})
```
### ML/AI Workflow Example
```elixir
# Using SessionHelpers for ML program management
alias Snakepit.SessionHelpers
# Create an ML program/model
{:ok, response} = SessionHelpers.execute_program_command(
"ml_session_123",
"create_program",
%{
signature: "question -> answer",
model: "gpt-3.5-turbo",
temperature: 0.7
}
)
program_id = response["program_id"]
# Execute the program multiple times
{:ok, result} = SessionHelpers.execute_program_command(
"ml_session_123",
"execute_program",
%{
program_id: program_id,
input: %{question: "What is the capital of France?"}
}
)
```
### Parallel Processing
```elixir
# Process multiple items in parallel across the pool
items = ["item1", "item2", "item3", "item4", "item5"]
tasks = Enum.map(items, fn item ->
Task.async(fn ->
Snakepit.execute("process_item", %{item: item})
end)
end)
results = Task.await_many(tasks, 30_000)
```
## 🔌 Built-in Adapters
### Python Adapter
```elixir
# Configure
Application.put_env(:snakepit, :adapter_module, Snakepit.Adapters.GenericPython)
# Available commands
{:ok, _} = Snakepit.execute("ping", %{})
{:ok, _} = Snakepit.execute("echo", %{message: "hello"})
{:ok, _} = Snakepit.execute("compute", %{operation: "multiply", a: 10, b: 5})
{:ok, _} = Snakepit.execute("info", %{})
```
### JavaScript/Node.js Adapter
```elixir
# Configure
Application.put_env(:snakepit, :adapter_module, Snakepit.Adapters.GenericJavaScript)
# Additional commands
{:ok, _} = Snakepit.execute("random", %{type: "uniform", min: 0, max: 100})
{:ok, _} = Snakepit.execute("compute", %{operation: "sqrt", a: 16})
```
## 🛠️ Creating Custom Adapters
### Elixir Adapter Implementation
```elixir
defmodule MyApp.RubyAdapter do
@behaviour Snakepit.Adapter
@impl true
def executable_path do
System.find_executable("ruby")
end
@impl true
def script_path do
Path.join(:code.priv_dir(:my_app), "ruby/bridge.rb")
end
@impl true
def script_args do
["--mode", "pool-worker"]
end
@impl true
def supported_commands do
["ping", "process_data", "generate_report"]
end
@impl true
def validate_command("process_data", args) do
if Map.has_key?(args, :data) do
:ok
else
{:error, "Missing required field: data"}
end
end
def validate_command("ping", _args), do: :ok
def validate_command(cmd, _args), do: {:error, "Unsupported command: #{cmd}"}
# Optional callbacks
@impl true
def prepare_args("process_data", args) do
# Transform args before sending
Map.update(args, :data, "", &String.trim/1)
end
@impl true
def process_response("generate_report", %{"report" => report} = response) do
# Post-process the response
{:ok, Map.put(response, "processed_at", DateTime.utc_now())}
end
@impl true
def command_timeout("generate_report", _args), do: 120_000 # 2 minutes
def command_timeout(_command, _args), do: 30_000 # Default 30 seconds
end
```
### External Bridge Script (Ruby Example)
```ruby
#!/usr/bin/env ruby
# priv/ruby/bridge.rb
require 'json'
class BridgeHandler
def initialize
@commands = {
'ping' => method(:handle_ping),
'process_data' => method(:handle_process_data),
'generate_report' => method(:handle_generate_report)
}
end
def run
STDERR.puts "Ruby bridge started"
loop do
# Read 4-byte length header
length_bytes = STDIN.read(4)
break unless length_bytes
# Unpack length (big-endian)
length = length_bytes.unpack('N')[0]
# Read JSON payload
json_data = STDIN.read(length)
request = JSON.parse(json_data)
# Process command
response = process_command(request)
# Send response
json_response = JSON.generate(response)
length_header = [json_response.bytesize].pack('N')
STDOUT.write(length_header)
STDOUT.write(json_response)
STDOUT.flush
end
end
private
def process_command(request)
command = request['command']
args = request['args'] || {}
handler = @commands[command]
if handler
result = handler.call(args)
{
'id' => request['id'],
'success' => true,
'result' => result,
'timestamp' => Time.now.iso8601
}
else
{
'id' => request['id'],
'success' => false,
'error' => "Unknown command: #{command}",
'timestamp' => Time.now.iso8601
}
end
rescue => e
{
'id' => request['id'],
'success' => false,
'error' => e.message,
'timestamp' => Time.now.iso8601
}
end
def handle_ping(args)
{ 'status' => 'ok', 'message' => 'pong' }
end
def handle_process_data(args)
data = args['data'] || ''
{ 'processed' => data.upcase, 'length' => data.length }
end
def handle_generate_report(args)
# Simulate report generation
sleep(1)
{
'report' => {
'title' => args['title'] || 'Report',
'generated_at' => Time.now.iso8601,
'data' => args['data'] || {}
}
}
end
end
# Handle signals gracefully
Signal.trap('TERM') { exit(0) }
Signal.trap('INT') { exit(0) }
# Run the bridge
BridgeHandler.new.run
```
## 🗃️ Session Management
### Session Store API
```elixir
alias Snakepit.Bridge.SessionStore
# Create a session
{:ok, session} = SessionStore.create_session("session_123", ttl: 7200)
# Store data in session
:ok = SessionStore.store_program("session_123", "prog_1", %{
model: "gpt-4",
temperature: 0.8
})
# Retrieve session data
{:ok, session} = SessionStore.get_session("session_123")
{:ok, program} = SessionStore.get_program("session_123", "prog_1")
# Update session
{:ok, updated} = SessionStore.update_session("session_123", fn session ->
Map.put(session, :last_activity, DateTime.utc_now())
end)
# Check if session exists
true = SessionStore.session_exists?("session_123")
# List all sessions
session_ids = SessionStore.list_sessions()
# Manual cleanup
SessionStore.delete_session("session_123")
# Get session statistics
stats = SessionStore.get_stats()
```
### Global Program Storage
```elixir
# Store programs accessible by any worker
:ok = SessionStore.store_global_program("template_1", %{
type: "qa_template",
prompt: "Answer the following question: {question}"
})
# Retrieve from any worker
{:ok, template} = SessionStore.get_global_program("template_1")
```
## 📊 Monitoring & Telemetry
### Available Events
```elixir
# Worker request completed
[:snakepit, :worker, :request]
# Measurements: %{duration: milliseconds}
# Metadata: %{result: :ok | :error}
# Worker initialized
[:snakepit, :worker, :initialized]
# Measurements: %{initialization_time: seconds}
# Metadata: %{worker_id: string}
```
### Setting Up Monitoring
```elixir
# In your application startup
:telemetry.attach_many(
"snakepit-metrics",
[
[:snakepit, :worker, :request],
[:snakepit, :worker, :initialized]
],
&MyApp.Metrics.handle_event/4,
%{}
)
defmodule MyApp.Metrics do
require Logger
def handle_event([:snakepit, :worker, :request], measurements, metadata, _config) do
# Log slow requests
if measurements.duration > 5000 do
Logger.warning("Slow request: #{measurements.duration}ms")
end
# Send to StatsD/Prometheus/DataDog
MyApp.Metrics.Client.histogram(
"snakepit.request.duration",
measurements.duration,
tags: ["result:#{metadata.result}"]
)
end
def handle_event([:snakepit, :worker, :initialized], measurements, metadata, _config) do
Logger.info("Worker #{metadata.worker_id} started in #{measurements.initialization_time}s")
end
end
```
### Pool Statistics
```elixir
stats = Snakepit.get_stats()
# Returns:
# %{
# workers: 8, # Total workers
# available: 6, # Available workers
# busy: 2, # Busy workers
# requests: 1534, # Total requests
# queued: 0, # Currently queued
# errors: 12, # Total errors
# queue_timeouts: 3, # Queue timeout count
# pool_saturated: 0 # Saturation rejections
# }
```
## 🏗️ Architecture Deep Dive
### Component Overview
```
┌─────────────────────────────────────────────────────────┐
│ Snakepit Application │
├───────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌──────────────┐ ┌───────────────┐ │
│ │ Pool │ │ SessionStore │ │ Registries │ │
│ │ Manager │ │ (ETS) │ │ (Worker/Proc)│ │
│ └──────┬──────┘ └──────────────┘ └───────────────┘ │
│ │ │
│ ┌──────▼────────────────────────────────────────────┐│
│ │ WorkerSupervisor (Dynamic) ││
│ └──────┬────────────────────────────────────────────┘│
│ │ │
│ ┌──────▼──────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Worker │ │ Worker │ │ Worker │ │
│ │ Starter │ │ Starter │ │ Starter │ │
│ │(Supervisor) │ │(Supervisor) │ │(Supervisor) │ │
│ └──────┬──────┘ └───────┬──────┘ └───────┬──────┘ │
│ │ │ │ │
│ ┌──────▼──────┐ ┌───────▼──────┐ ┌───────▼──────┐ │
│ │ Worker │ │ Worker │ │ Worker │ │
│ │ (GenServer) │ │ (GenServer) │ │ (GenServer) │ │
│ └──────┬──────┘ └───────┬──────┘ └───────┬──────┘ │
│ │ │ │ │
└─────────┼─────────────────┼─────────────────┼─────────┘
│ │ │
┌─────▼──────┐ ┌─────▼──────┐ ┌─────▼──────┐
│ External │ │ External │ │ External │
│ Process │ │ Process │ │ Process │
│ (Python) │ │ (Node.js) │ │ (Ruby) │
└────────────┘ └────────────┘ └────────────┘
```
### Key Design Decisions
1. **Concurrent Initialization**: Workers start in parallel using `Task.async_stream`
2. **Permanent Wrapper Pattern**: Worker.Starter supervises Workers for auto-restart
3. **Centralized State**: All session data in ETS, workers are stateless
4. **Registry-Based**: O(1) worker lookups and reverse PID lookups
5. **Port Communication**: Binary protocol with 4-byte length headers
### Process Lifecycle
1. **Startup**:
- Pool manager starts
- Concurrently spawns N workers via WorkerSupervisor
- Each worker starts its external process
- Workers send init ping and register when ready
2. **Request Flow**:
- Client calls `Snakepit.execute/3`
- Pool finds available worker (with session affinity if applicable)
- Worker sends request to external process
- External process responds
- Worker returns result to client
3. **Crash Recovery**:
- Worker crashes → Worker.Starter restarts it automatically
- External process dies → Worker detects and crashes → restart
- Pool crashes → Supervisor restarts entire pool
4. **Shutdown**:
- Pool manager sends shutdown to all workers
- Workers close ports gracefully (SIGTERM)
- ApplicationCleanup ensures no orphaned processes (SIGKILL)
## ⚡ Performance
### Benchmarks
```
Configuration: 16 workers, Python adapter
Hardware: 8-core CPU, 32GB RAM
Startup Time:
- Sequential: 16 seconds (1s per worker)
- Concurrent: 1.2 seconds (13x faster)
Throughput:
- Simple computation: 50,000 req/s
- Complex ML inference: 1,000 req/s
- Session operations: 45,000 req/s
Latency (p99):
- Simple computation: < 2ms
- Complex ML inference: < 100ms
- Session operations: < 1ms
```
### Optimization Tips
1. **Pool Size**: Start with `System.schedulers_online() * 2`
2. **Queue Size**: Monitor `pool_saturated` errors and adjust
3. **Timeouts**: Set appropriate timeouts per command type
4. **Session TTL**: Balance memory usage vs cache hits
5. **Health Checks**: Increase interval for stable workloads
## 🔧 Troubleshooting
### Common Issues
#### Workers Not Starting
```elixir
# Check adapter configuration
adapter = Application.get_env(:snakepit, :adapter_module)
adapter.executable_path() # Should return valid path
File.exists?(adapter.script_path()) # Should return true
# Check logs for errors
Logger.configure(level: :debug)
```
#### Port Exits
```elixir
# Enable port tracing
:erlang.trace(Process.whereis(Snakepit.Pool.Worker), true, [:receive, :send])
# Check external process logs
# Python: Add logging to bridge script
# Node.js: Check stderr output
```
#### Memory Leaks
```elixir
# Monitor ETS usage
:ets.info(:snakepit_sessions, :memory)
# Check for orphaned processes
Snakepit.Pool.ProcessRegistry.get_stats()
# Force cleanup
Snakepit.Bridge.SessionStore.cleanup_expired_sessions()
```
### Debug Mode
```elixir
# Enable debug logging
Logger.configure(level: :debug)
# Trace specific worker
:sys.trace(Snakepit.Pool.Registry.via_tuple("worker_1"), true)
# Get internal state
:sys.get_state(Snakepit.Pool)
```
## 🤝 Contributing
We welcome contributions! Please see our [Contributing Guide](https://github.com/nshkrdotcom/snakepit/blob/main/CONTRIBUTING.md) for details.
### Development Setup
```bash
# Clone the repo
git clone https://github.com/nshkrdotcom/snakepit.git
cd snakepit
# Install dependencies
mix deps.get
# Run tests
mix test
# Run with example scripts
elixir examples/session_based_demo.exs
elixir examples/javascript_session_demo.exs
# Check code quality
mix format --check-formatted
mix dialyzer
```
### Running Tests
```bash
# All tests
mix test
# With coverage
mix test --cover
# Specific test
mix test test/snakepit_test.exs:42
```
## 📝 License
Snakepit is released under the MIT License. See the [LICENSE](https://github.com/nshkrdotcom/snakepit/blob/main/LICENSE) file for details.
## 🙏 Acknowledgments
- Inspired by the need for reliable ML/AI integrations in Elixir
- Built on battle-tested OTP principles
- Special thanks to the Elixir community
## 📚 Resources
- [Hex Package](https://hex.pm/packages/snakepit)
- [API Documentation](https://hexdocs.pm/snakepit)
- [GitHub Repository](https://github.com/nshkrdotcom/snakepit)
- [Example Projects](https://github.com/nshkrdotcom/snakepit/tree/main/examples)
---
Made with ❤️ by [NSHkr](https://github.com/nshkrdotcom)