当前位置:网站首页>2、 Multi concurrent interface pressure test
2、 Multi concurrent interface pressure test
2022-07-29 06:08:00 【My hair is messy】
One 、flsak Interface pressure test
import base64
import logging
import os, cv2,time
import urllib, glob
import numpy as np
import requests, time, json, threading, random
import traceback
def cv_imread(file_path):
# Reading Chinese path
cv_img = cv2.imdecode(np.fromfile(file_path, dtype=np.uint8), -1)
return cv_img
class LogTxt:
def __init__(self, txt_path):
self.current_path = txt_path
create_time = time.strftime("%Y-%m-%d %H:%M:%S")
self.txt_handle = open(self.current_path, 'a',encoding="utf-8")
self.info(f" ----------- start test in {create_time} --------\n")
def info(self, content):
try:
now_time = time.strftime("%Y-%m-%d %H:%M:%S")
write_content = now_time + f' - keyPoint - INFO - :' + content
self.txt_handle.write(write_content + ' \n')
self.txt_handle.flush()
print(write_content)
except:
print("=============== > LogTxt info have bug < =========== ", traceback.format_exc())
def close(self):
self.txt_handle.close()
class Presstest(object):
"""
Concurrent stress testing
"""
def __init__(self, press_url, file,logger):
self.press_url = press_url
self.file = file
self.logger=logger
def test_interface(self):
''' Pressure measuring interface '''
global INDEX
INDEX += 1
global ERROR_NUM
global TIME_LENS
try:
start = time.time()
logger.info(f" The start time of identifying this HD image is :{start}")
r = requests.post(self.press_url, files=self.file)
# print(r.text)
logger.info(f" The recognition result of this HD image is :{r.text}")
end = time.time()
total_time=end-start
logger.info(f" The end time of this HD image recognition is :{start}")
logger.info(f" The total time for identifying this HD image is :{total_time}")
TIME_LENS.append(end - start)
logger.info(" HD image recognition is over , Identify the next high-definition image .\n\n")
print('end')
except Exception as e:
ERROR_NUM += 1
logger.info(f" It is recognized that the abnormality of this HD image is :{e}")
print(e)
# headers = {
'Content-Type': 'application/json; charset=UTF-8',
# 'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64; rv:23.0) Gecko/20100101 Firefox/23.0'}
def test_onework(self):
''' Concurrent processing of a single task at a time '''
i = 0
while i < ONE_WORKER_NUM:
i += 1
self.test_interface()
# print('one worker num {}'.format(i))
logger.info('one worker num {}'.format(i))
time.sleep(LOOP_SLEEP)
##---------------------------------------------
# todo Universal http obtain webapi Request result method
## --------------------------------------------
# def do_request(self, press_url,register_data):
# ''' Universal http obtain webapi Request result method '''
# headers = {
# 'Content-Type': 'application/json; charset=UTF-8',
# }
# request = urllib.request.Request(url, json.dumps(payload).encode("utf-8"), headers=headers)
# retry_num = 0
# while retry_num < 3:
# response = urllib.request.urlopen(request, timeout=300)
# if not response or response.status == 421:
# time.sleep(1)
# retry_num = retry_num + 1
# continue
# else:
# break
# response_content = response.read()
# if hasattr(response_content, 'decode'):
# response_content = response_content.decode('utf-8')
#
# return response_content
def run(self):
''' Concurrent testing using multithreaded processes '''
t1 = time.time()
Threads = []
for i in range(THREAD_NUM):
# print('thread Num {}'.format(i))
logger.info('thread Num {}'.format(i))
t = threading.Thread(target=self.test_onework, name="T" + str(i))
t.setDaemon(True)
Threads.append(t)
for t in Threads:
t.start()
for t in Threads:
t.join()
t2 = time.time()
# print("=============== Pressure test results ===================")
# print("URL:", self.press_url)
# print(" Number of tasks :", THREAD_NUM, "*", ONE_WORKER_NUM, "=", THREAD_NUM * ONE_WORKER_NUM)
# print(" Total time ( second ):", t2 - t1)
# print(" Each request takes time ( second ):", (t2 - t1) / (THREAD_NUM * ONE_WORKER_NUM))
# print(" Number of requests per second :", 1 / ((t2 - t1) / (THREAD_NUM * ONE_WORKER_NUM)))
# print(" Number of errors :", ERROR_NUM)
# print(INDEX)
logger.info("=============== Pressure test results ===================")
logger.info("URL: {}".format(self.press_url))
logger.info(" Number of tasks :{} * {} = {}".format(THREAD_NUM,ONE_WORKER_NUM,THREAD_NUM * ONE_WORKER_NUM))
logger.info(" Total time ( second ):{}".format(t2-t1))
logger.info(" Each request takes time ( second ):{}".format((t2 - t1) / (THREAD_NUM * ONE_WORKER_NUM)))
logger.info(" Number of requests per second :{}".format(1 / ((t2 - t1) / (THREAD_NUM * ONE_WORKER_NUM))))
logger.info(" Number of errors :{}".format(ERROR_NUM))
logger.info("{}".format(INDEX))
logger.info("---------------- This test is over , The above is the test result -----------------")
if __name__ == '__main__':
press_url = 'http://192.168.2.93:8095/recog'
TIME_LENS = []
INDEX = 0
THREAD_NUM = 25 # Total concurrent threads
ONE_WORKER_NUM = 50 # Number of cycles per thread
LOOP_SLEEP = 0 # Time interval per request ( second )
ERROR_NUM = 0 # The number of errors
logger=LogTxt(rf"./log/thread-{THREAD_NUM}.log")
path1 = r"C:\val"
for path in glob.glob(os.path.join(path1, "*.jpg" or "*.png")):
# path=r"/data2/enducation/answer_card/answer-card-recognition/pic/2021-08-17_14_16_18.jpg"
with open(path, "rb") as f:
img_data = f.read() # <class 'bytes'>
# print(type(img_data))
files = {
'file': img_data}
obj = Presstest(press_url, files,logger)
obj.run()
print('onetime')
边栏推荐
- 一、网页端文件流的传输
- AttributeError: module ‘tensorflow‘ has no attribute ‘placeholder‘
- PyTorch的数据读取机制
- Improve quality with intelligence financial imaging platform solution
- 虚假新闻检测论文阅读(三):Semi-supervised Content-based Detection of Misinformation via Tensor Embeddings
- 性能优化之趣谈线程池:线程开的越多就越好吗?
- Nifi changed UTC time to CST time
- 【目标检测】Generalized Focal Loss V1
- 一、常见损失函数的用法
- 迁移学习——Low-Rank Transfer Subspace Learning
猜你喜欢
C connect to SharePoint online webservice
iSCSI vs iSER vs NVMe-TCP vs NVMe-RDMA
【Attention】Visual Attention Network
ASM piling: after learning ASM tree API, you don't have to be afraid of hook anymore
anaconda中移除旧环境、增加新环境、查看环境、安装库、清理缓存等操作命令
[target detection] generalized focal loss v1
Beijing Baode & taocloud jointly build the road of information innovation
【目标检测】KL-Loss:Bounding Box Regression with Uncertainty for Accurate Object Detection
【Transformer】ATS: Adaptive Token Sampling For Efficient Vision Transformers
神经网络相关知识回顾(PyTorch篇)
随机推荐
Ffmpeg creation GIF expression pack tutorial is coming! Say thank you, brother black fly?
Error in installing pyspider under Windows: Please specify --curl dir=/path/to/build/libcurl solution
【Attention】Visual Attention Network
anaconda中移除旧环境、增加新环境、查看环境、安装库、清理缓存等操作命令
[target detection] 6. SSD
迁移学习——Transfer Joint Matching for Unsupervised Domain Adaptation
ML8自学笔记-LDA原理公式推导
一、multiprocessing.pool.RemoteTraceback
clion+opencv+aruco+cmake配置
第三周周报 ResNet+ResNext
【TensorRT】将 PyTorch 转化为可部署的 TensorRT
Yum local source production
有价值的博客、面经收集(持续更新)
[semantic segmentation] full attention network for semantic segmentation
ML15 neural network (1)
1、 Usage of common loss function
PyTorch的数据读取机制
GA-RPN:引导锚点的建议区域网络
ML15-神经网络(1)
【语义分割】语义分割综述