[Python] 10 MB `grpc.aio` request latency regression from `protobuf` 4.25.8 to 6.33.6

What version of protobuf and what language are you using?

• Language: Python

• Compared versions:

  • protobuf 4.25.8
  • protobuf 6.33.6

• Other dependencies:

  • grpcio 1.75.0
  • cloudpickle 3.1.1

What operating system version are you using?

Linux x86_64, kernel 7.0.0-1011-aws, glibc 2.39.

What runtime / compiler version are you using?

CPython 3.10.20.

What did you do?

The reproduction below sends a 10,000,000-character Python string as a cloudpickled protobuf bytes field over a local Python grpc.aio unary RPC to a separate server process.

python3.10 -m venv repro-env

repro-env/bin/python -m pip install \
  grpcio==1.75.0 cloudpickle==3.1.1 protobuf==4.25.8

repro-env/bin/python repro.py \
  --warmup 100 --requests 350

repro-env/bin/python -m pip install --force-reinstall protobuf==6.33.6

repro-env/bin/python repro.py \
  --warmup 100 --requests 350

Each run performs 100 untimed sequential warmups, followed by 350 timed sequential RPCs. The reported p50 is the median of those 350 samples.

What did you expect to see?

I expected no material latency regression for the same workload when upgrading the Python protobuf runtime.

What did you see instead?

The regression reproduced in two runs of each version:

In this environment, protobuf 6.33.6 adds about 10 ms, or about 19%, to the p50 of this end-to-end large-message request path compared with protobuf 4.25.8.

Could you help determine whether a Python/upb runtime behavior change affects this large-message gRPC workload?

Standalone reproduction

#!/usr/bin/env python3
"""repro.py: standalone large-message grpc.aio reproduction."""

import argparse
import asyncio
import pickle
import socket
import statistics
import subprocess
import sys
import time

import cloudpickle
import grpc
from google.protobuf import __version__ as protobuf_version
from google.protobuf import descriptor_pb2, descriptor_pool, message_factory

PAYLOAD_CHARS = 10_000_000
MAX_MESSAGE_BYTES = 10_500_000
METHOD = "/large_message.Echo/Handle"


def message_types():
    file_descriptor = descriptor_pb2.FileDescriptorProto(
        name="large_message_repro.proto",
        package="large_message",
        syntax="proto3",
    )

    request = file_descriptor.message_type.add()
    request.name = "Request"
    for number, name in enumerate(("metadata", "args", "kwargs"), start=1):
        field = request.field.add()
        field.name = name
        field.number = number
        field.label = descriptor_pb2.FieldDescriptorProto.LABEL_OPTIONAL
        field.type = descriptor_pb2.FieldDescriptorProto.TYPE_BYTES

    response = file_descriptor.message_type.add()
    response.name = "Response"

    field = response.field.add()
    field.name = "result"
    field.number = 1
    field.label = descriptor_pb2.FieldDescriptorProto.LABEL_OPTIONAL
    field.type = descriptor_pb2.FieldDescriptorProto.TYPE_BYTES

    field = response.field.add()
    field.name = "is_error"
    field.number = 2
    field.label = descriptor_pb2.FieldDescriptorProto.LABEL_OPTIONAL
    field.type = descriptor_pb2.FieldDescriptorProto.TYPE_BOOL

    pool = descriptor_pool.DescriptorPool()
    pool.Add(file_descriptor)

    request_type = message_factory.GetMessageClass(
        pool.FindMessageTypeByName("large_message.Request")
    )
    response_type = message_factory.GetMessageClass(
        pool.FindMessageTypeByName("large_message.Response")
    )
    return request_type, response_type


async def run_server(port: int) -> None:
    Request, Response = message_types()

    async def handle(request, context):
        metadata = pickle.loads(request.metadata)
        args = cloudpickle.loads(request.args)
        kwargs = cloudpickle.loads(request.kwargs)

        assert metadata["request_serialization"] == "cloudpickle"
        assert len(args) == 1 and len(args[0]) == PAYLOAD_CHARS and not kwargs

        return Response(result=cloudpickle.dumps(b""))

    server = grpc.aio.server(
        options=[("grpc.max_receive_message_length", MAX_MESSAGE_BYTES)]
    )
    handler = grpc.unary_unary_rpc_method_handler(
        handle,
        request_deserializer=Request.FromString,
        response_serializer=Response.SerializeToString,
    )
    server.add_generic_rpc_handlers(
        (
            grpc.method_handlers_generic_handler(
                "large_message.Echo",
                {"Handle": handler},
            ),
        )
    )
    assert server.add_insecure_port(f"127.0.0.1:{port}") == port
    await server.start()
    print("READY", flush=True)
    await server.wait_for_termination()


def unused_local_port() -> int:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.bind(("127.0.0.1", 0))
        return sock.getsockname()[1]


async def benchmark(warmup: int, requests: int) -> None:
    Request, Response = message_types()
    port = unused_local_port()

    server = subprocess.Popen(
        [sys.executable, __file__, "--server", str(port)],
        stdout=subprocess.PIPE,
        text=True,
    )
    assert server.stdout is not None

    try:
        assert server.stdout.readline().strip() == "READY"

        payload = "x" * PAYLOAD_CHARS
        metadata = pickle.dumps(
            {"request_serialization": "cloudpickle"},
            protocol=pickle.HIGHEST_PROTOCOL,
        )
        empty_kwargs = cloudpickle.dumps({})

        async with grpc.aio.insecure_channel(
            f"127.0.0.1:{port}",
            options=[("grpc.max_receive_message_length", MAX_MESSAGE_BYTES)],
        ) as channel:
            handle = channel.unary_unary(
                METHOD,
                request_serializer=Request.SerializeToString,
                response_deserializer=Response.FromString,
            )

            async def call_once():
                request = Request(
                    metadata=metadata,
                    args=cloudpickle.dumps((payload,)),
                    kwargs=empty_kwargs,
                )
                response = await handle(request)
                assert cloudpickle.loads(response.result) == b""

            for _ in range(warmup):
                await call_once()

            samples_ms = []
            for _ in range(requests):
                started = time.perf_counter_ns()
                await call_once()
                samples_ms.append(
                    (time.perf_counter_ns() - started) / 1_000_000
                )
    finally:
        server.terminate()
        server.wait(timeout=10)

    samples_ms.sort()
    print(
        "protobuf=%s grpcio=%s cloudpickle=%s "
        "payload_chars=%s n=%s p50_ms=%.3f p95_ms=%.3f mean_ms=%.3f"
        % (
            protobuf_version,
            grpc.__version__,
            cloudpickle.__version__,
            PAYLOAD_CHARS,
            len(samples_ms),
            statistics.median(samples_ms),
            samples_ms[round(0.95 * (len(samples_ms) - 1))],
            statistics.mean(samples_ms),
        )
    )


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--server", type=int)
    parser.add_argument("--warmup", type=int, default=100)
    parser.add_argument("--requests", type=int, default=500)
    args = parser.parse_args()

    if args.server:
        asyncio.run(run_server(args.server))
    else:
        asyncio.run(benchmark(args.warmup, args.requests))


if __name__ == "__main__":
    main()

Start by running the standalone repro.py with the stated CPython, grpcio, cloudpickle, and protobuf versions, comparing the reported p50 and p95 latency. Trace the large-message grpc.aio request path to determine whether the Python/upb runtime accounts for the difference. Done means a reproducible explanation of the approximately 10 ms regression and its affected runtime path.