|
|
|
@ -165,10 +165,10 @@ def _cancel_all_rpcs(futures: Mapping[int, grpc.Future]) -> None: |
|
|
|
|
future.cancel() |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _run_single_channel(args: argparse.Namespace): |
|
|
|
|
def _run_single_channel(qps: int, server: str, rpc_timeout_sec: int, print_response: bool): |
|
|
|
|
global _global_rpc_id # pylint: disable=global-statement |
|
|
|
|
duration_per_query = 1.0 / float(args.qps) |
|
|
|
|
with grpc.insecure_channel(args.server) as channel: |
|
|
|
|
duration_per_query = 1.0 / float(qps) |
|
|
|
|
with grpc.insecure_channel(server) as channel: |
|
|
|
|
stub = test_pb2_grpc.TestServiceStub(channel) |
|
|
|
|
futures: Dict[int, grpc.Future] = {} |
|
|
|
|
while not _stop_event.is_set(): |
|
|
|
@ -178,8 +178,8 @@ def _run_single_channel(args: argparse.Namespace): |
|
|
|
|
_global_rpc_id += 1 |
|
|
|
|
start = time.time() |
|
|
|
|
end = start + duration_per_query |
|
|
|
|
_start_rpc(request_id, stub, float(args.rpc_timeout_sec), futures) |
|
|
|
|
_remove_completed_rpcs(futures, args.print_response) |
|
|
|
|
_start_rpc(request_id, stub, float(rpc_timeout_sec), futures) |
|
|
|
|
_remove_completed_rpcs(futures, print_response) |
|
|
|
|
logger.debug(f"Currently {len(futures)} in-flight RPCs") |
|
|
|
|
now = time.time() |
|
|
|
|
while now < end: |
|
|
|
@ -193,7 +193,7 @@ def _run(args: argparse.Namespace) -> None: |
|
|
|
|
global _global_server # pylint: disable=global-statement |
|
|
|
|
channel_threads: List[threading.Thread] = [] |
|
|
|
|
for i in range(args.num_channels): |
|
|
|
|
thread = threading.Thread(target=_run_single_channel, args=(args,)) |
|
|
|
|
thread = threading.Thread(target=_run_single_channel, args=(args.qps, args.server, args.rpc_timeout_sec, args.print_response,)) |
|
|
|
|
thread.start() |
|
|
|
|
channel_threads.append(thread) |
|
|
|
|
_global_server = grpc.server(futures.ThreadPoolExecutor()) |
|
|
|
|