추론요청 시나리오 작성 모듈과 라운드로빈으로 요청을 엣지장비들에 전송하는 스케줄러 구현

bench/inception_v3/grpc_bench.py

@@ -5,21 +5,13 @@
 #tf log setting
 import os
 os.environ['TF_CPP_MIN_LOG_LEVEL'] = '3'
-import tensorflow as tf
-import numpy as np
-
-#preprocessing library
-from inception_v3 import preprocessing
 
 #병렬처리 library
 import concurrent.futures
 
-def run_bench(num_tasks, server_address, use_https):
+def run_bench(num_tasks, server_address, use_https, data):
     model_name = "inception_v3"
 
-    image_file_path = "../../dataset/imagenet/imagenet_1000_raw/n01843383_1.JPEG"
-    data = tf.make_tensor_proto(preprocessing.run_preprocessing(image_file_path))
-
     stub = module_grpc.create_grpc_stub(server_address, use_https)
 
     # gRPC 요청 생성

bench/inception_v3/preprocessing.py

@@ -1,15 +1,11 @@
 #image 전처리 library
-import tensorflow as tf
 import numpy as np
 from PIL import Image
 import os
 
-def get_file_path(filename):
-    return os.path.join(os.path.dirname(__file__), filename)
-
 # 이미지 로드 및 전처리 (for inception)
 def run_preprocessing(image_file_path):
-    img = Image.open(get_file_path(image_file_path))
+    img = Image.open(image_file_path)
     img = img.resize((299, 299))
     img_array = np.array(img)
     img_array = (img_array - np.mean(img_array)) / np.std(img_array) 

bench/inception_v3/rest_bench.py

@@ -1,19 +1,11 @@
-#preprocessing library
-from inception_v3 import preprocessing
-import numpy as np
-
 #REST 요청 관련 library
 from module import module_rest
-import json
 
 #병렬처리 library
 import concurrent.futures
 
-def run_bench(num_tasks, server_address):
+def run_bench(num_tasks, server_address, data):
     model_name = "inception_v3"
-    image_file_path = "../../../dataset/imagenet/imagenet_1000_raw/n01843383_1.JPEG"
-
-    data = json.dumps({"instances": preprocessing.run_preprocessing(image_file_path).tolist()})
 
     # REST 요청 병렬 처리
     with concurrent.futures.ThreadPoolExecutor(max_workers=num_tasks) as executor:

bench/inference_request_workload_manager.py

@@ -0,0 +1,66 @@
+from numpy import random
+import pickle as pk
+
+# (modle name, requests per second)
+inference_request_info = [
+    ('mobilenet_v1', 20),
+    ('mobilenet_v2', 2),
+    ('inception_v3', 2),
+    ('yolo_v5', 1)
+]
+
+file_name = 'workload4'
+req_time_num = 20
+
+def create_inference_request_workload_by_poisson(req_time):
+    requests = [[] for _ in range(req_time)]
+    total_req_num = 0
+    for (model_name, req_per_sec) in inference_request_info:
+        workloads = random.poisson(lam=req_per_sec, size=req_time)
+        for idx in range(req_time):
+            requests[idx].extend([model_name for _ in range(workloads[idx])])
+        total_req_num += sum(workloads)
+
+    for idx in range(req_time):
+        random.shuffle(requests[idx])
+
+    workload_info = {}
+    workload_info['total_request_num'] = total_req_num
+    workload_info['requests'] = requests
+
+    return workload_info
+
+
+def create_inference_request_workload_regularly(req_time):
+    requests = [[] for _ in range(req_time)]
+    total_req_num = 0
+    for (model_name, req_per_sec) in inference_request_info:
+        for idx in range(req_time):
+            requests[idx].extend([model_name for _ in range(req_per_sec)])
+            total_req_num += req_per_sec  
+
+    for idx in range(req_time):
+        random.shuffle(requests[idx])
+
+    workload_info = {}
+    workload_info['total_request_num'] = total_req_num
+    workload_info['requests'] = requests
+
+    return workload_info
+
+def save_workload_info_to_file(file_name, workloads):
+    with open(file_name, 'wb') as f:
+        pk.dump(workloads, f)
+
+
+def load_workload_info_from_file(file_name):
+    with open(file_name, 'rb') as f:
+        loaded_workloads = pk.load(f)
+    return loaded_workloads
+
+
+# workloads = create_inference_request_workload_by_poisson(req_time_num)
+# workloads = create_inference_request_workload_regularly(req_time_num)
+# save_workload_info_to_file(file_name, workloads)
+# loaded_workload_info = load_workload_info_from_file(file_name)
+# print(loaded_workload_info)

bench/mobilenet_v1/grpc_bench.py

@@ -5,21 +5,13 @@
 #tf log setting
 import os
 os.environ['TF_CPP_MIN_LOG_LEVEL'] = '3'
-import tensorflow as tf
-import numpy as np
-
-#preprocessing library
-from mobilenet_v1 import preprocessing
 
 #병렬처리 library
 import concurrent.futures
 
-def run_bench(num_tasks, server_address, use_https):
+def run_bench(num_tasks, server_address, use_https, data):
     model_name = "mobilenet_v1"
 
-    image_file_path = "../../dataset/imagenet/imagenet_1000_raw/n01843383_1.JPEG"
-    data = tf.make_tensor_proto(preprocessing.run_preprocessing(image_file_path))
-
     stub = module_grpc.create_grpc_stub(server_address, use_https)
 
     # gRPC 요청 생성

bench/mobilenet_v1/preprocessing.py

@@ -1,15 +1,11 @@
 #image 전처리 library
-import tensorflow as tf
 import numpy as np
 from PIL import Image
 import os
 
-def get_file_path(filename):
-    return os.path.join(os.path.dirname(__file__), filename)
-
 # 이미지 로드 및 전처리 (for mobilenet)
 def run_preprocessing(image_file_path):
-    img = Image.open(get_file_path(image_file_path))
+    img = Image.open(image_file_path)
     img = img.resize((224, 224))
     img_array = np.array(img)
     img_array = img_array.astype('float32') / 255.0

bench/mobilenet_v1/rest_bench.py

@@ -1,19 +1,11 @@
-#preprocessing library
-from mobilenet_v1 import preprocessing
-import numpy as np
-
 #REST 요청 관련 library
 from module import module_rest
-import json
 
 #병렬처리 library
 import concurrent.futures
 
-def run_bench(num_tasks, server_address):
+def run_bench(num_tasks, server_address, data):
     model_name = "mobilenet_v1"
-    image_file_path = "../../../dataset/imagenet/imagenet_1000_raw/n01843383_1.JPEG"
-
-    data = json.dumps({"instances": preprocessing.run_preprocessing(image_file_path).tolist()})
 
     # REST 요청 병렬 처리
     with concurrent.futures.ThreadPoolExecutor(max_workers=num_tasks) as executor:

bench/mobilenet_v2/grpc_bench.py

@@ -5,21 +5,13 @@
 #tf log setting
 import os
 os.environ['TF_CPP_MIN_LOG_LEVEL'] = '3'
-import tensorflow as tf
-import numpy as np
-
-#preprocessing library
-from mobilenet_v2 import preprocessing
 
 #병렬처리 library
 import concurrent.futures
 
-def run_bench(num_tasks, server_address, use_https):
+def run_bench(num_tasks, server_address, use_https, data):
     model_name = "mobilenet_v2"
 
-    image_file_path = "../../dataset/imagenet/imagenet_1000_raw/n01843383_1.JPEG"
-    data = tf.make_tensor_proto(preprocessing.run_preprocessing(image_file_path))
-
     stub = module_grpc.create_grpc_stub(server_address, use_https)
 
     # gRPC 요청 생성

bench/mobilenet_v2/preprocessing.py

@@ -1,15 +1,11 @@
 #image 전처리 library
-import tensorflow as tf
 import numpy as np
 from PIL import Image
 import os
 
-def get_file_path(filename):
-    return os.path.join(os.path.dirname(__file__), filename)
-
 # 이미지 로드 및 전처리 (for mobilenet)
 def run_preprocessing(image_file_path):
-    img = Image.open(get_file_path(image_file_path))
+    img = Image.open(image_file_path)
     img = img.resize((224, 224))
     img_array = np.array(img)
     img_array = img_array.astype('float32') / 255.0

bench/mobilenet_v2/rest_bench.py

@@ -1,20 +1,12 @@
-#preprocessing library
-from mobilenet_v2 import preprocessing
-import numpy as np
-
 #REST 요청 관련 library
 from module import module_rest
-import json
 
 #병렬처리 library
 import concurrent.futures
 
-def run_bench(num_tasks, server_address):
+def run_bench(num_tasks, server_address, data):
     model_name = "mobilenet_v2"
-    image_file_path = "../../../dataset/imagenet/imagenet_1000_raw/n01843383_1.JPEG"
 
-    data = json.dumps({"instances": preprocessing.run_preprocessing(image_file_path).tolist()})
-
     # REST 요청 병렬 처리
     with concurrent.futures.ThreadPoolExecutor(max_workers=num_tasks) as executor:
         futures = [executor.submit(lambda: module_rest.predict(server_address, model_name, data)) for _ in range(num_tasks)]

bench/preprocessing_data_manager.py

@@ -0,0 +1,32 @@
+import tensorflow as tf
+import json
+import importlib
+import os.path
+
+
+data_source_info = {'mobilenet_v1': '../dataset/imagenet/imagenet_1000_raw/n01843383_1.JPEG',
+                    'mobilenet_v2': '../dataset/imagenet/imagenet_1000_raw/n01843383_1.JPEG',
+                    'inception_v3': '../dataset/imagenet/imagenet_1000_raw/n01843383_1.JPEG',
+                    'yolo_v5': '../dataset/coco_2017/coco/images/val2017/000000089761.jpg',
+                }
+
+def get_file_path(filename):
+    return os.path.join(os.path.dirname(__file__), filename)
+
+def regist_preprocessed_datas(request_type):
+    preprocessed_datas = {}
+
+    for model in data_source_info.keys():
+        if request_type == 'rest':
+            preprocessing_module = importlib.import_module(f"{model}.preprocessing")
+            data_source = data_source_info.get(model)
+
+            data = json.dumps({"instances": preprocessing_module.run_preprocessing(get_file_path(data_source)).tolist()})
+            preprocessed_datas.update({model: data})
+        elif request_type == 'grpc':
+            preprocessing_module = importlib.import_module(f"{model}.preprocessing")
+            data_source = data_source_info.get(model)
+            data = tf.make_tensor_proto(preprocessing_module.run_preprocessing(get_file_path(data_source)))
+            preprocessed_datas.update({model: data})
+
+    return preprocessed_datas