xai-org/x-algorithm · error · ValueError

Invalid {num_servers=} or {num_processors=} in CheckStateRes

Error message

Invalid {num_servers=} or {num_processors=} in CheckStateResponse

What it means

get_service_dimensions calls the KafkaDispatcher gRPC service's CheckState and expects both num_shards (servers) and num_processors to be set on the response. If either proto field is unset (proto3 default / None for optional fields), the response is considered invalid for computing the client-to-server sharding layout.

Source

Thrown at phoenix/xrex/data/streaming/kafkadispatcherloader.py:64

    ("grpc.keepalive_timeout_ms", 10000),
    ("grpc.http2.max_pings_without_data", 0),
]


def get_grpc_channel(grpc_host: str, grpc_port) -> grpc.aio.Channel:
    target = f"{grpc_host}:{grpc_port}"
    return grpc.aio.insecure_channel(target, options=OPTIONS)


async def get_service_dimensions(grpc_host_template: str, grpc_port: int) -> tuple[int, int]:
    master_node_host = grpc_host_template.format(shard_index="0")
    async with get_grpc_channel(master_node_host, grpc_port) as channel:
        stub = KafkaDispatcherStub(channel)
        response = await stub.CheckState(CheckStateRequest())
        num_servers = response.num_shards
        num_processors = response.num_processors
        if num_servers is None or num_processors is None:
            raise ValueError(f"Invalid {num_servers=} or {num_processors=} in CheckStateResponse")
        return num_servers, num_processors


async def get_server_assignments(
    grpc_host_template: str,
    grpc_port: int,
    num_clients: int,
    client_ix: int,
) -> tuple[int, int, list[int]]:
    num_servers, num_processors = await get_service_dimensions(grpc_host_template, grpc_port)
    if num_servers is None or num_processors is None:
        raise ValueError("Failed to get num_servers or num_processors from gRPC service")

    assert num_clients % (num_servers * num_processors) == 0, (
        f"Num dataloaders {num_clients} must be a multiple of number of servers {num_servers=} * {num_processors=} = {num_servers * num_processors}"
    )

    global_num_processors = num_servers * num_processors

View on GitHub (pinned to 24c60942c5)

Solutions

1. Check the dispatcher service version/proto definition and upgrade the server so CheckState populates both fields.
2. Ensure grpc_host_template/grpc_port point at the actual KafkaDispatcher service, not another gRPC endpoint.
3. If the service may still be booting, retry CheckState with backoff before failing.
4. Inspect the raw response (log response) to see which field is missing.

Example fix

# before
response = await stub.CheckState(CheckStateRequest())

# after
response = await stub.CheckState(CheckStateRequest())
rank_logger.info(f"CheckState response: {response}")
for attempt in range(5):
    if response.num_shards is not None and response.num_processors is not None:
        break
    await asyncio.sleep(2)
    response = await stub.CheckState(CheckStateRequest())

resp = await stub.CheckState(CheckStateRequest())
assert resp.num_shards is not None and resp.num_processors is not None

def check_state_is_valid(resp) -> bool:
    return resp.num_shards is not None and resp.num_processors is not None

try:
    num_servers, num_processors = await get_service_dimensions(host, port)
except ValueError as e:
    # retry with backoff — service may still be initializing
    await asyncio.sleep(5)
    num_servers, num_processors = await get_service_dimensions(host, port)

When it happens

Trigger: Calling get_server_assignments against a dispatcher service version that does not populate num_shards or num_processors in CheckStateResponse; service still initializing and state not yet computed; connecting to an unexpected/older gRPC endpoint that returns a differently-shaped response.

Common situations: Version skew between the Python client and the dispatcher server (new client expects fields the old server never sets); dispatcher pod restarted and reporting empty state; wrong grpc_port hitting a different gRPC service.

Related errors

• Failed to get num_servers or num_processors from gRPC servic
• Client registration failed: {register_response.error_message
• Unknown tag %d
• Failed to create serializer for ThriftStruct %s: %s
• enable_stale_post is unsupported on grpc_recsys, which does

AI-assisted analysis of xai-org/x-algorithm@24c60942c5 (2026-08-28). Data as JSON: /api/errors/3c4222fccb0b1c13. Report an issue: GitHub.