Merge pull request #22165 from lidizheng/aio-benchmark

Integrate existing gRPC Python stack benchmarks with Bazel

src/python/grpcio_tests/tests/qps/BUILD.bazel

@@ -25,3 +25,66 @@ py_library(
         "//src/proto/grpc/testing:stats_py_pb2",
     ],
 )
+
+py_library(
+    name = "benchmark_client",
+    srcs = ["benchmark_client.py"],
+    srcs_version = "PY2AND3",
+    deps = [
+        "//src/proto/grpc/testing:benchmark_service_py_pb2_grpc",
+        "//src/proto/grpc/testing:py_messages_proto",
+        "//src/python/grpcio/grpc:grpcio",
+        "//src/python/grpcio_tests/tests/unit:resources",
+        "//src/python/grpcio_tests/tests/unit:test_common",
+    ],
+)
+
+py_library(
+    name = "benchmark_server",
+    srcs = ["benchmark_server.py"],
+    srcs_version = "PY2AND3",
+    deps = [
+        "//src/proto/grpc/testing:benchmark_service_py_pb2_grpc",
+        "//src/proto/grpc/testing:py_messages_proto",
+    ],
+)
+
+py_library(
+    name = "client_runner",
+    srcs = ["client_runner.py"],
+    srcs_version = "PY2AND3",
+)
+
+py_library(
+    name = "worker_server",
+    srcs = ["worker_server.py"],
+    srcs_version = "PY2AND3",
+    deps = [
+        ":benchmark_client",
+        ":benchmark_server",
+        ":client_runner",
+        ":histogram",
+        "//src/proto/grpc/core:stats_py_pb2",
+        "//src/proto/grpc/testing:benchmark_service_py_pb2_grpc",
+        "//src/proto/grpc/testing:control_py_pb2",
+        "//src/proto/grpc/testing:payloads_py_pb2",
+        "//src/proto/grpc/testing:stats_py_pb2",
+        "//src/proto/grpc/testing:worker_service_py_pb2_grpc",
+        "//src/python/grpcio/grpc:grpcio",
+        "//src/python/grpcio_tests/tests/unit:resources",
+        "//src/python/grpcio_tests/tests/unit:test_common",
+    ],
+)
+
+py_binary(
+    name = "qps_worker",
+    srcs = ["qps_worker.py"],
+    imports = ["../.."],
+    srcs_version = "PY2AND3",
+    deps = [
+        ":worker_server",
+        "//src/proto/grpc/testing:worker_service_py_pb2_grpc",
+        "//src/python/grpcio/grpc:grpcio",
+        "//src/python/grpcio_tests/tests/unit:test_common",
+    ],
+)

src/python/grpcio_tests/tests/qps/benchmark_client.py

@@ -61,14 +61,16 @@ class BenchmarkClient:
             self._stub = benchmark_service_pb2_grpc.BenchmarkServiceStub(
                 channel)
             payload = messages_pb2.Payload(
-                body='\0' * config.payload_config.simple_params.req_size)
+                body=bytes(b'\0' *
+                           config.payload_config.simple_params.req_size))
             self._request = messages_pb2.SimpleRequest(
                 payload=payload,
                 response_size=config.payload_config.simple_params.resp_size)
         else:
             self._generic = True
             self._stub = GenericStub(channel)
-            self._request = '\0' * config.payload_config.bytebuf_params.req_size
+            self._request = bytes(b'\0' *
+                                  config.payload_config.bytebuf_params.req_size)
 
         self._hist = hist
         self._response_callbacks = []

src/python/grpcio_tests/tests/qps/benchmark_server.py

@@ -20,12 +20,12 @@ class BenchmarkServer(benchmark_service_pb2_grpc.BenchmarkServiceServicer):
     """Synchronous Server implementation for the Benchmark service."""
 
     def UnaryCall(self, request, context):
-        payload = messages_pb2.Payload(body='\0' * request.response_size)
+        payload = messages_pb2.Payload(body=b'\0' * request.response_size)
         return messages_pb2.SimpleResponse(payload=payload)
 
     def StreamingCall(self, request_iterator, context):
         for request in request_iterator:
-            payload = messages_pb2.Payload(body='\0' * request.response_size)
+            payload = messages_pb2.Payload(body=b'\0' * request.response_size)
             yield messages_pb2.SimpleResponse(payload=payload)
 
 
@@ -34,7 +34,7 @@ class GenericBenchmarkServer(benchmark_service_pb2_grpc.BenchmarkServiceServicer
     """Generic Server implementation for the Benchmark service."""
 
     def __init__(self, resp_size):
-        self._response = '\0' * resp_size
+        self._response = b'\0' * resp_size
 
     def UnaryCall(self, request, context):
         return self._response

src/python/grpcio_tests/tests_aio/benchmark/worker.py

@@ -52,6 +52,8 @@ if __name__ == '__main__':
 
     if args.uvloop:
         import uvloop
-        uvloop.install()
+        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
+        loop = uvloop.new_event_loop()
+        asyncio.set_event_loop(loop)
 
     asyncio.get_event_loop().run_until_complete(run_worker_server(args.port))

tools/run_tests/performance/build_performance.sh

@@ -70,8 +70,7 @@ do
     tools/run_tests/performance/build_performance_node.sh
     ;;
   "python")
-    # python workers are only run with python2.7 and building with multiple python versions is costly
-    python tools/run_tests/run_tests.py -l "$language" -c "$CONFIG" --compiler python2.7 --build_only -j 8
+    $bazel build -c opt //src/python/grpcio_tests/tests/qps:qps_worker
     ;;
   "python_asyncio")
     $bazel build -c opt //src/python/grpcio_tests/tests_aio/benchmark:worker

tools/run_tests/performance/run_worker_python.sh

@@ -17,4 +17,4 @@ set -ex
 
 cd "$(dirname "$0")/../../.."
 
-PYTHONPATH=src/python/grpcio_tests:src/python/gens py27_native/bin/python src/python/grpcio_tests/tests/qps/qps_worker.py "$@"
+bazel-bin/src/python/grpcio_tests/tests/qps/qps_worker "$@"