[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:

protobuf	run	p50 latency	p95 latency	mean
4.25.8	1	53.341 ms	53.830 ms	53.153 ms
4.25.8	2	53.559 ms	54.288 ms	53.622 ms
6.33.6	1	63.547 ms	64.799 ms	63.614 ms
6.33.6	2	63.694 ms	66.017 ms	63.860 ms

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()