Distributed Workers

Distributed Workers Architecture

Load this skill when working on:

• Worker implementation or debugging
• ServiceOrchestrator or WorkerAPIBase patterns
• Operation dispatch, progress tracking, or cancellation
• Backend-to-worker communication
• Adding new worker types

Architecture Overview

KTRDR uses a distributed workers architecture where the backend orchestrates operations across worker nodes:

┌─────────────────────────────────────────────────────────────────┐
│ Backend (Docker Container, Port 8000)                          │
│  ├─ API Layer (FastAPI)                                        │
│  ├─ Service Orchestrators (NEVER execute operations)           │
│  ├─ WorkerRegistry (tracks all workers)                        │
│  └─ OperationsService (tracks all operations)                  │
└─────────────────────────────────────────────────────────────────┘
         │
         ├─ HTTP (Worker Registration & Operation Dispatch)
         │
    ┌────┴────┬──────────┬──────────┬─────────────┐
    │         │          │          │             │
    ▼         ▼          ▼          ▼             ▼
┌───────┐ ┌───────┐ ┌───────┐ ┌───────┐   ┌──────────────┐
│Backtest││Backtest││Training││Training│   │IB Host Service│
│Worker 1││Worker 2││Worker 1││Worker 2│   │(Port 5001)   │
│:5003   ││:5003   ││:5004   ││:5004   │   │Direct IB TCP │
└────────┘└────────┘└────────┘└────────┘   └──────────────┘
CPU-only  CPU-only  CPU-only  CPU-only    Direct IB Gateway

         ┌─────────────────────┐
         │Training Host Service│
         │(Port 5002)          │
         │GPU Access (CUDA/MPS)│
         └─────────────────────┘
         10x-100x faster training

Key Principles

1. Backend as Orchestrator Only: Backend does not execute operations locally — it only selects workers and dispatches operations
2. Distributed-Only Execution: All backtesting and training operations execute on workers (no local fallback)
3. Self-Registering Workers: Workers push-register with backend on startup (infrastructure-agnostic)
4. GPU-First Routing: Training operations prefer GPU workers (10x-100x faster) with CPU worker fallback
5. Horizontal Scalability: Add more workers = more concurrent operations

Worker Types

ServiceOrchestrator Pattern

Location: ktrdr/async_infrastructure/service_orchestrator.py

All service managers inherit from ServiceOrchestrator:

class DataAcquisitionService(ServiceOrchestrator):
    def __init__(self):
        # Reads USE_IB_HOST_SERVICE env var
        self.provider = self._initialize_provider()

    async def download_data(self, ...):
        # Unified async pattern with progress tracking
        return await self._execute_with_progress(...)

Features provided:

• Environment-based configuration
• Adapter initialization (local vs. host service routing)
• Unified async operations with progress tracking
• Cancellation token support
• Operations service integration

WorkerAPIBase Pattern

Location: ktrdr/workers/base.py

All workers inherit from WorkerAPIBase and get these features for free:

1. OperationsService singleton — Worker-local operation tracking
2. Operations proxy endpoints:
   • GET /api/v1/operations/{id} — Get operation status
   • GET /api/v1/operations/{id}/metrics — Get operation metrics
   • GET /api/v1/operations — List operations
   • DELETE /api/v1/operations/{id}/cancel — Cancel operation
3. Health endpoint — Reports busy/idle status (GET /health)
4. FastAPI app with CORS — Ready for Docker communication
5. Self-registration — Automatic registration with backend on startup

Key Pattern Elements

• Operation ID Synchronization: Accepts optional task_id from backend, returns same operation_id
• Progress Tracking: Workers register progress bridges in their OperationsService
• Remote Queryability: Backend can query worker's operations endpoints directly (1s cache TTL)
• Push-Based Registration: Workers call POST /workers/register on startup

Example Implementation

class BacktestWorker(WorkerAPIBase):
    def __init__(self, worker_port=5003, backend_url="http://backend:8000"):
        super().__init__(
            worker_type=WorkerType.BACKTESTING,
            operation_type=OperationType.BACKTESTING,
            worker_port=worker_port,
            backend_url=backend_url,
        )

        # Register domain-specific endpoint
        @self.app.post("/backtests/start")
        async def start_backtest(request: BacktestStartRequest):
            operation_id = request.task_id or f"worker_backtest_{uuid.uuid4().hex[:12]}"
            result = await self._execute_backtest_work(operation_id, request)
            return {"success": True, "operation_id": operation_id, **result}

Worker Implementations

• BacktestWorker (ktrdr/backtesting/backtest_worker.py):

  • Adds /backtests/start endpoint
  • Calls BacktestingEngine directly via asyncio.to_thread
  • Registers BacktestProgressBridge

• TrainingWorker (ktrdr/training/training_worker.py):

  • Adds /training/start endpoint
  • Calls TrainingManager directly (async)
  • Simplified progress tracking

Host Service Integration

IB Host Service (uses environment variables)

USE_IB_HOST_SERVICE=true
IB_HOST_SERVICE_URL=http://localhost:5001  # default

Why separate: IB Gateway requires direct TCP connection (Docker networking limitation)

Training & Backtesting (uses WorkerRegistry)

Environment flags REMOVED in Phase 5.3:

• ❌ USE_TRAINING_HOST_SERVICE
• ❌ REMOTE_BACKTEST_SERVICE_URL

Now uses WorkerRegistry:

• Workers self-register with backend on startup
• Backend selects available workers automatically
• GPU workers register with gpu: true capability (prioritized)
• CPU workers register as fallback

Starting Workers

# Docker Compose (development)
docker-compose up -d --scale backtest-worker=5 --scale training-worker=3

# Training Host Service (GPU, runs natively)
cd training-host-service && ./start.sh

# Workers self-register at:
# - Backtest: http://localhost:5003
# - Training (CPU): http://localhost:5004
# - Training (GPU): http://localhost:5002

Verification

# Check registered workers
curl http://localhost:8000/api/v1/workers | jq

# Expected: All workers show as AVAILABLE with proper capabilities

Cancellation Tokens

Location: ktrdr/async_infrastructure/cancellation.py

All long-running operations support cancellation:

from ktrdr.async_infrastructure.cancellation import create_cancellation_token

token = create_cancellation_token()

# In operation loop
if token.is_cancelled():
    raise asyncio.CancelledError()

• Create tokens with create_cancellation_token()
• Check with token.is_cancelled()
• Operations service manages tokens globally
• CLI displays cancellation status

Async Operations Pattern (CLI)

All CLI commands use AsyncCLIClient for API communication:

from ktrdr.cli.helpers.async_cli_client import AsyncCLIClient

async def some_command(symbol: str):
    async with AsyncCLIClient() as client:
        result = await client.post("/endpoint", json=data)

Progress display: Use GenericProgressManager with ProgressRenderer for live updates

Documentation

• Architecture: docs/architecture-overviews/distributed-workers.md
• Developer Guide: docs/developer/distributed-workers-guide.md
• Deployment: docs/user-guides/deployment.md