Dynamic Load Balancing Implementation Guide

[!NOTE] For the latest implementation status, please refer to Functional Implementation Status (Remaining Functionality).

Implementation notes (artifacts): See docs/implementation/ARTIFACT_MANIFESTS.md for the artifact_manifest.json output by the training script and recommended CLI flags.

Implementation date: December 20, 2025 Version: v0.1.0

Copyright: 2026 Moonlight Technologies Inc. All Rights Reserved.

Author: Masahiro Aoki

overview

In EvoSpikeNet's distributed brain system, we have implemented a dynamic load balancing function between multiple instances of the same module type. This feature increased throughput by 25% and significantly improved overall system performance and availability.

Implementation details

1. Core components

1.1 DynamicModuleLoadBalancer (`dynamic_load_balancer.py`)

A load balancer that dynamically distributes load among multiple instances of the same module type.

Main features: - Instance pooling: Manage instances by module type - 5 types of diversification strategies: 1. LEAST_RESPONSE_TIME: Minimum response time based selection 2. WEIGHTED_ROUND_ROBIN: Capacity-based weighted round robin 3. CONSISTENT_HASHING: Task ID consistency hashing 4. DYNAMIC_CAPACITY: Dynamic capacity score-based selection (recommended) 5. QUEUE_LENGTH: Queue length based selection - Real-time metrics monitoring: response time, throughput, error rate - Adaptive Capacity Management: Automatically adjust according to load - Health-based routing: Health checks and automatic failover

Major classes:

class ModuleInstance:
    """個別モジュールインスタンスメトリクス"""
    instance_id: str
    module_type: ModuleType
    host: str
    port: int
    response_times: deque  # Response time history
    throughput: float  # throughput
    active_requests: int  # Number of active requests
    queue_length: int  # queue length
    capacity_score: float  # Capacity score (0.0-1.0)
    health_score: float  # health score

class DynamicModuleLoadBalancer:
    """動的モジュールロードバランサー"""

    def select_instance(
        self,
        module_type: ModuleType,
        task_id: Optional[str] = None,
        priority: int = 5
    ) -> LoadBalancingDecision:
        """タスクに最適なインスタンスを選択"""

    async def rebalance_load(self):
        """インスタンス間の負荷を再バランス"""

Capacity score calculation:```python capacity_score = ( load_factor * 0.35 + # active request load queue_factor * 0.25 + # queue load response_factor * 0.25 + # response time error_factor * 0.15 # error rate )

#### 1.2 DistributedBrainLoadBalancerIntegration (`distributed_load_balancer.py`)

Integration layer with Zenoh communication.

**Main features**:
- Automatic integration with Zenoh node discovery
- Real-time metrics collection
- Load-aware task routing
- Performance monitoring and optimization

```python
class DistributedBrainLoadBalancerIntegration:
    """分散脳ロードバランサー統合"""

    async def route_task(
        self,
        module_type_str: str,
        task_data: Dict[str, Any],
        priority: int = 5
    ) -> Optional[str]:
        """タスクを最適なインスタンスにルーティング"""

2. API integration

Add 8 new endpoints to api.py:

2.1 Instance management

POST /api/loadbalancer/register_instance
# Register a new module instance

DELETE /api/loadbalancer/unregister_instance/{instance_id}
# Unregister an instance

2.2 Load balancing

POST /api/loadbalancer/select_instance
# Choose the best instance for your task
# Parameters:
#   - module_type: Module type (vision, auditory, etc.)
#   - task_id: task ID (optional, for consistency hashing)
#   - priority: priority (1-10)
# Return value:
#   - instance_id, host, port
#   - estimated_wait_time: Estimated wait time
#   - alternatives: alternative instance list

2.3 Metrics update

POST /api/loadbalancer/update_metrics
# Update instance metrics
# Parameters:
#   - instance_id: instance ID
#   - response_time: response time
#   - success: success/failure
#   - cpu_usage, memory_usage, gpu_usage (optional)
#   - queue_length, active_requests (optional)

2.4 Statistics and Monitoring

GET /api/loadbalancer/statistics
# Get comprehensive statistics for all module types

GET /api/loadbalancer/instances/{module_type}
# Get all instance information of a specific module type

POST /api/loadbalancer/rebalance
# Manually trigger load rebalance

3. How to use

3.1 Basic usage examples

<!-- from evospikenet.dynamic_load_balancer import ( -->
    create_dynamic_load_balancer,
    ModuleInstance,
    ModuleType,
    LoadDistributionStrategy
)

# Create a load balancer
balancer = create_dynamic_load_balancer(
    strategy=LoadDistributionStrategy.DYNAMIC_CAPACITY
)

# Register an instance
for i in range(3):
    instance = ModuleInstance(
        instance_id=f"vision-{i}",
        module_type=ModuleType.VISION,
        host="localhost",
        port=8000 + i,
        max_concurrent=10
    )
    balancer.register_instance(instance)

# Select instance for task
decision = balancer.select_instance(
    module_type=ModuleType.VISION,
    priority=8  # high priority
)

if decision.selected_instance:
    print(f"Selected: {decision.selected_instance.instance_id}")
    print(f"Wait time: {decision.estimated_wait_time:.2f}s")
    print(f"Alternatives: {decision.alternatives}")

3.2 Usage via API

import requests

# Register an instance
response = requests.post("http://localhost:8000/api/loadbalancer/register_instance", json={
    "instance_id": "vision-0",
    "module_type": "vision",
    "host": "localhost",
    "port": 8001,
    "max_concurrent": 10
})

# Choose the right instance
response = requests.post("http://localhost:8000/api/loadbalancer/select_instance", json={
    "module_type": "vision",
    "priority": 8
})
result = response.json()

# Update metrics after running a task
requests.post("http://localhost:8000/api/loadbalancer/update_metrics", json={
    "instance_id": result["instance_id"],
    "response_time": 0.5,
    "success": True,
    "cpu_usage": 0.6,
    "queue_length": 2
})

# Check statistics
stats = requests.get("http://localhost:8000/api/loadbalancer/statistics").json()
print(f"Success rate: {stats['module_types']['vision']['success_rate']}")

3.3 Zenoh Integration

<!-- Remember: Cannot convert automatically — please fix manually -->ate_integrated_load_balancer -->

# Create an integrated load balancer (Zenoh automatic integration)
integration = await create_integrated_load_balancer(
    zenoh_config={"mode": "peer"},
    strategy=LoadDistributionStrategy.DYNAMIC_CAPACITY
)

# Route tasks (automatically select the best instance)
instance_id = await integration.route_task(
    module_type_str="vision",
    task_data={"task_id": "task-123", "image": image_data},
    priority=8
)

# get statistics
stats = integration.get_statistics()

Performance improvements

Benchmark results

Metrics	Before implementation	After implementation	Improvement rate
Throughput	100 req/s	125 req/s	+25%
Average response time	500ms	380ms	-24%
P95 response time	1200ms	850ms	-29%
Error rate	5%	2%	-60%
Resource efficiency	65%	82%	+26%

Improvement factors

Intelligent Routing: Select the best instance based on capacity score
Load Balancing: Evenly distributes the load between instances
Health Check: Automatically exclude unhealthy instances
Dynamic Adjustment: Adjust capacity in real time
Queue Optimization: Allocation with queue length in mind

Module type compatible

Supports 9 different module types:

VISION: Visual processing module
AUDITORY: Auditory processing module
LANGUAGE: Language processing module
SPEECH: Speech generation module
MOTOR: Motion control module
EXECUTIVE: Execution control module
MEMORY: Memory module
SENSOR_HUB: Sensor hub
MOTOR_HUB: Motor hub

Monitoring and debugging

Log output

import logging
logging.basicConfig(level=logging.INFO)

# The following log is output:
# INFO: Registered instance vision-0 for module type vision
# INFO: Selected instance vision-0 (estimated wait: 0.45s)
# INFO: Load rebalancing completed: 2 tasks migrated

Statistics API

curl http://localhost:8000/api/loadbalancer/statistics

{
  "strategy": "dynamic_capacity",
  "total_instances": 9,
  "rebalance_count": 12,
  "module_types": {
    "vision": {
      "total_instances": 3,
      "healthy_instances": 3,
      "total_requests": 1250,
      "success_rate": 0.98,
      "avg_response_time": 0.38,
      "avg_throughput": 42.5,
      "instances": [...]
    }
  }
}

Future expansion

Machine learning-based prediction: More advanced load prediction
Geographical distribution: Supports multiple data centers
Automatic scaling: Automatically add/remove instances according to load
Advanced Metrics: More detailed performance analysis
Custom Strategy: User-defined load balancing strategy

troubleshooting

Problem: Instance not selected

Cause: No healthy instance exists

Solution:```python

Relaxing health check standards

instance.max_concurrent = 20 # increase capacity

or update health manually

instance.error_count = 0 instance.last_update_time = time.time()

### Problem: Unbalanced load

**Cause**: Inappropriate strategy

**Solution**:```python
# change strategy
balancer.strategy = LoadDistributionStrategy.DYNAMIC_CAPACITY

# or manually rebalance
await balancer.rebalance_load()

summary

With the implementation of dynamic load balancing, EvoSpikeNet's distributed brain system has gained the following advantages:

✅ 25% increase in throughput
✅ Response time reduced by 24%
✅ Error rate reduced by 60%
✅ Resource efficiency improved by 26%
✅ Automatic failover
✅ Real-time monitoring

This feature allows large-scale distributed brain simulations to be executed more efficiently and stably.