fix benchmark

python/examples/pipeline/simple_web_service/benchmark.py

@@ -24,7 +24,9 @@ def run_http(idx, batch_size):
     print("start thread ({})".format(idx))
     url = "http://127.0.0.1:18082/uci/prediction"    
     start = time.time()
-    data = {"key": ["x"], "value": ["0.0137, -0.1136, 0.2553, -0.0692, 0.0582, -0.0727, -0.1583, -0.0584, 0.6283, 0.4919, 0.1856, 0.0795, -0.0332"]}
+    value = "0.0137, -0.1136, 0.2553, -0.0692, 0.0582, -0.0727, -0.1583, -0.0584, 0.6283, 0.4919, 0.1856, 0.0795, -0.0332"
+    all_value = ";".join([value for i in range(batch_size)])
+    data = {"key": ["x"], "value": [all_value]}
     for i in range(1000):
         r = requests.post(url=url, data=json.dumps(data))
     print(r.json())
@@ -33,21 +35,22 @@ def run_http(idx, batch_size):
 
 def multithread_http(thread, batch_size):
     multi_thread_runner = MultiThreadRunner()
-    result = multi_thread_runner.run(
-        run_http , thread, batch_size)
-
-
-def multithread_rpc(thraed, batch_size):
-    multi_thread_runner = MultiThreadRunner()
-
+    result = multi_thread_runner.run(run_http , thread, batch_size)
 
 def run_rpc(thread, batch_size):
     client = PipelineClient()
     client.connect(['127.0.0.1:9998'])
-    data = {"key": "x", "value": "0.0137, -0.1136, 0.2553, -0.0692, 0.0582, -0.0727, -0.1583, -0.0584, 0.6283, 0.4919, 0.1856, 0.0795, -0.0332"}
-    ret = client.predict(feed_dict={data["key"]: data["value"]}, fetch=["res"])
+    value = "0.0137, -0.1136, 0.2553, -0.0692, 0.0582, -0.0727, -0.1583, -0.0584, 0.6283, 0.4919, 0.1856, 0.0795, -0.0332"
+    all_value = ";".join([value for i in range(batch_size)])
+    data = {"key": "x", "value": all_value}
+    for i in range(1000):
+        ret = client.predict(feed_dict={data["key"]: data["value"]}, fetch=["res"])
     print(ret)
 
+def multithread_rpc(thraed, batch_size):
+    multi_thread_runner = MultiThreadRunner()
+    result = multi_thread_runner.run(run_rpc , thread, batch_size)
+
 if __name__ == "__main__":
     if sys.argv[1] == "yaml":
         mode = sys.argv[2] # brpc/  local predictor
@@ -60,6 +63,6 @@ if __name__ == "__main__":
         if mode == "http":
             multithread_http(thread, batch_size)
         elif mode == "rpc":
-            run_rpc(thread, batch_size)
+            multithread_rpc(thread, batch_size)
 
     

python/examples/pipeline/simple_web_service/benchmark.sh

+# HTTP
 ps -ef | grep web_service | awk '{print $2}' | xargs kill -9 
-rm -rf PipelineServingLogs
-rm -rf cpu_utilization.py
 sleep 3
 python3 benchmark.py yaml local_predictor 1 
 
-for thread_num in 1 8 32
+for thread_num in 1 
+do
+for batch_size in 1
 do
+rm -rf PipelineServingLogs
+rm -rf cpu_utilization.py
 python3 web_service.py >web.log 2>&1 &
 sleep 3
 echo "import psutil\ncpu_utilization=psutil.cpu_percent(1,False)\nprint('CPU_UTILIZATION:', cpu_utilization)\n" > cpu_utilization.py
-python3 benchmark.py run http $thread_num 1
+python3 benchmark.py run http $thread_num $batch_size
 python3 cpu_utilization.py
-echo "------------Fit a line pipeline benchmark (Thread: $thread_num) (Benchmark: 1)"
+echo "------------Fit a line pipeline benchmark (Thread: $thread_num) (BatchSize: $batch_size)"
 tail -n 25 PipelineServingLogs/pipeline.tracer 
 ps -ef | grep web_service | awk '{print $2}' | xargs kill -9
 done
+done
+
+# RPC
+ps -ef | grep web_service | awk '{print $2}' | xargs kill -9
+sleep 3
+python3 benchmark.py yaml local_predictor 1
+
+for thread_num in 1 
+do
+for batch_size in 1 
+do
+rm -rf PipelineServingLogs
+rm -rf cpu_utilization.py
+python3 web_service.py >web.log 2>&1 &
+sleep 3
+echo "import psutil\ncpu_utilization=psutil.cpu_percent(1,False)\nprint('CPU_UTILIZATION:', cpu_utilization)\n" > cpu_utilization.py
+python3 benchmark.py run rpc $thread_num $batch_size
+python3 cpu_utilization.py
+echo "------------Fit a line pipeline benchmark (Thread: $thread_num) (BatchSize: $batch_size)"
+tail -n 25 PipelineServingLogs/pipeline.tracer
+ps -ef | grep web_service | awk '{print $2}' | xargs kill -9
+done
+done

python/examples/pipeline/simple_web_service/web_service.py

@@ -25,32 +25,24 @@ _LOGGER = logging.getLogger()
 class UciOp(Op):
     def init_op(self):
         self.separator = ","
+        self.batch_separator = ";"
 
     def preprocess(self, input_dicts, data_id, log_id):
         (_, input_dict), = input_dicts.items()
         _LOGGER.error("UciOp::preprocess >>> log_id:{}, input:{}".format(
             log_id, input_dict))
-        x_value = input_dict["x"]
+        x_value = input_dict["x"].split(self.batch_separator)
+        x_lst = []
+        for x_val in x_value:
+            x_lst.append(np.array([float(x.strip()) for x in x_val.split(self.separator)]).reshape(1, 13))
+        input_dict["x"] = np.concatenate(x_lst, axis=0) 
         proc_dict = {}
-        if sys.version_info.major == 2:
-            if isinstance(x_value, (str, unicode)):
-                input_dict["x"] = np.array(
-                    [float(x.strip())
-                     for x in x_value.split(self.separator)]).reshape(1, 13)
-                _LOGGER.error("input_dict:{}".format(input_dict))
-        else:
-            if isinstance(x_value, str):
-                input_dict["x"] = np.array(
-                    [float(x.strip())
-                     for x in x_value.split(self.separator)]).reshape(1, 13)
-                _LOGGER.error("input_dict:{}".format(input_dict))
-
         return input_dict, False, None, ""
 
     def postprocess(self, input_dicts, fetch_dict, log_id):
         _LOGGER.info("UciOp::postprocess >>> log_id:{}, fetch_dict:{}".format(
             log_id, fetch_dict))
-        fetch_dict["price"] = str(fetch_dict["price"][0][0])
+        fetch_dict["price"] = str(fetch_dict["price"])
         return fetch_dict, None, ""