Packages
snakepit
0.5.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
priv/python/snakepit_bridge/adapters/showcase/handlers/concurrent_ops.py
"""Concurrent operations handler for showcase adapter."""
import time
import os
import psutil
from typing import Dict, Any
from ..tool import Tool
class ConcurrentOpsHandler:
"""Handler for concurrent operations demonstrations."""
def get_tools(self) -> Dict[str, Tool]:
"""Return all tools provided by this handler."""
return {
"cpu_bound_task": Tool(self.cpu_bound_task),
"sleep_task": Tool(self.sleep_task),
"lightweight_task": Tool(self.lightweight_task),
"get_pool_stats": Tool(self.get_pool_stats),
}
def cpu_bound_task(self, ctx, task_id: str,
duration_ms: int) -> Dict[str, Any]:
"""Simulate CPU-bound work."""
start_time = time.time()
# Simulate work
result = 0
iterations = duration_ms * 1000 # Adjust for CPU speed
for i in range(iterations):
result += i ** 2
actual_duration = (time.time() - start_time) * 1000
return {
"task_id": task_id,
"result": result % 1000000, # Keep manageable
"actual_duration_ms": round(actual_duration, 2)
}
def sleep_task(self, ctx, duration_ms: int,
task_number: int) -> Dict[str, Any]:
"""Simple sleep task."""
time.sleep(duration_ms / 1000.0)
return {
"task_number": task_number,
"slept_ms": duration_ms
}
def lightweight_task(self, ctx, iteration: int) -> Dict[str, Any]:
"""Very lightweight task for benchmarking."""
return {
"iteration": iteration,
"result": iteration * 2
}
def get_pool_stats(self, ctx) -> Dict[str, Any]:
"""Get current pool statistics."""
process = psutil.Process(os.getpid())
memory_mb = process.memory_info().rss / 1024 / 1024
return {
"worker_pid": os.getpid(),
"memory_mb": round(memory_mb, 2),
"cpu_percent": process.cpu_percent()
}