import json import time from dataclasses import dataclass, asdict from typing import Any from grpc_client import GrpcClient from rest_client import RestClient from utils import NetworkingSize, generate_random_book_data from google.protobuf.json_format import MessageToDict from google._upb._message import RepeatedCompositeContainer import psutil import copy # SERVER_ADDRESS = "localhost" # REST_SERVER_PORT = 9500 # GRPC_SERVER_PORT = 50051 @dataclass class PerformanceResult: client_networking_ping: float = 0.0 client_request_time: float = 0.0 client_request_cpu: float = 0.0 server_deserialize_time: float = 0.0 server_deserialize_cpu: float = 0.0 server_serialize_time: float = 0.0 server_serialize_cpu: float = 0.0 server_protocol_total_time: float = 0.0 response_data: Any = None networking_size: NetworkingSize = None class ClientPerformanceTest: def __init__(self, server_address, rest_server_port, grpc_server_port): self.grpc = GrpcClient(f"{server_address}:{grpc_server_port}") self.rest = RestClient(f"http://{server_address}:{rest_server_port}") @staticmethod def ping_test(func, count=5): time_total = 0 for i in range(count): time_start = time.perf_counter() func() time_end = time.perf_counter() time_total += (time_end - time_start) time.sleep(0.1) return time_total / count @staticmethod def client_performance_test(networking_ping, func, *args, **kwargs): # networking_ping = ping_test(grpc.ping, count=5) process = psutil.Process() cpu_count = psutil.cpu_count() request_start_time = time.perf_counter() process.cpu_percent() # cpu_request_start = process.cpu_times() # result = grpc.get_list(pages=page, per_page=per_page) result, networking_size = func(*args, **kwargs) cpu_request_percent = process.cpu_percent() # cpu_request_end = process.cpu_times() request_end_time = time.perf_counter() client_request_time = request_end_time - request_start_time # CPU time seconds # client_request_cpu_time = (cpu_request_end.user - cpu_request_start.user) + \ # (cpu_request_end.system - cpu_request_start.system) # client_request_cpu_usage = client_request_cpu_time / client_request_time * 100 server_deserialize_time = result.server_deserialize.time server_deserialize_cpu = result.server_deserialize.cpu server_serialize_time = result.server_serialize.time server_serialize_cpu = result.server_serialize.cpu server_protocol_total_time = result.server_protocol_total_time return PerformanceResult(client_networking_ping=networking_ping, client_request_time=client_request_time, client_request_cpu=cpu_request_percent / cpu_count, server_deserialize_time=server_deserialize_time, server_deserialize_cpu=server_deserialize_cpu, server_serialize_time=server_serialize_time, server_serialize_cpu=server_serialize_cpu, server_protocol_total_time=server_protocol_total_time, response_data=result.response_data, networking_size=networking_size) def test_get_list_performance_grpc(self, per_page=10000, page=1, list_data_limit=30): networking_ping = self.ping_test(self.grpc.ping, count=3) return self.client_performance_test(networking_ping, self.grpc.get_list, pages=page, per_page=per_page , list_data_limit=list_data_limit) def test_get_list_performance_rest(self, per_page=10000, page=1, list_data_limit=30): networking_ping = self.ping_test(self.rest.ping, count=3) return self.client_performance_test(networking_ping, self.rest.get_list, pages=page, per_page=per_page, list_data_limit=list_data_limit) def _test_add_books_performance_grpc(self, books, test_only=False): networking_ping = self.ping_test(self.grpc.ping, count=3) return self.client_performance_test(networking_ping, self.grpc.add_books, books=books, test_only=test_only) def _test_add_books_performance_rest(self, books, test_only=False): networking_ping = self.ping_test(self.rest.ping, count=3,) return self.client_performance_test(networking_ping, self.rest.add_book, books=books, test_only=test_only) def test_add_books_performance(self, book_count=1000, test_only=True): books_data = generate_random_book_data(book_count) grpc_result = self._test_add_books_performance_grpc(books_data, test_only=test_only) if not test_only: self.grpc.delete_books([], book_count) rest_result = self._test_add_books_performance_rest(books_data, test_only=test_only) if not test_only: self.rest.delete_books([], book_count) return grpc_result, rest_result def test_get_main(per_page=1_0000, list_data_limit=100): client_performance_test = ClientPerformanceTest("127.0.0.1", 9500, 50051) grpc_result = client_performance_test.test_get_list_performance_grpc(per_page=per_page, page=1) if len(grpc_result.response_data) > list_data_limit: grpc_result.response_data = grpc_result.response_data[:list_data_limit] if isinstance(grpc_result.response_data, RepeatedCompositeContainer): grpc_result.response_data = list(grpc_result.response_data) for n, i in enumerate(grpc_result.response_data): grpc_result.response_data[n] = MessageToDict(i, preserving_proto_field_name=True) if len(grpc_result.response_data) <= 0: grpc_result.response_data = [] # result_dict = copy.deepcopy(grpc_result.__dict__) # result_dict.pop("response_data") # print("gRPC Performance Result:", result_dict) rest_result = client_performance_test.test_get_list_performance_rest(per_page=per_page, page=1) if len(rest_result.response_data) > list_data_limit: rest_result.response_data = rest_result.response_data[:list_data_limit] for n, i in enumerate(rest_result.response_data): rest_result.response_data[n] = dict(i) # result_dict = copy.deepcopy(rest_result.__dict__) # result_dict.pop("response_data") # print("REST Performance Result:", result_dict) return asdict(grpc_result), asdict(rest_result) def test_add_main(book_count=1_0000): client_performance_test = ClientPerformanceTest("127.0.0.1", 9500, 50051) grpc_result, rest_result = client_performance_test.test_add_books_performance(book_count=book_count) grpc_result.response_data = MessageToDict(grpc_result.response_data, preserving_proto_field_name=True) rest_result.response_data = dict(rest_result.response_data) # print("gRPC Add Books Performance Result:", grpc_result) # print("REST Add Books Performance Result:", rest_result) return asdict(grpc_result), asdict(rest_result) if __name__ == '__main__': print(test_get_main(10000, 2)) # test_add_main()