int8 test

Author: stephenoffer

.idea/workspace.xml

@@ -7,6 +7,7 @@
 
 import os
 import time
+import numpy as np
 
 import ray
 
@@ -138,7 +139,7 @@ def make_model():
                 else:
                     time.sleep(1e-3)
 
-    def predict(self, data, distributed=True, close=False):
+    def predict(self, data, distributed=True, close=False, int8_cvrt=False):
         """
         Run inference on a data batch, returns predictions
 
@@ -149,6 +150,9 @@ def predict(self, data, distributed=True, close=False):
         return: Predictions
         """
         if distributed:
+            if int8_cvrt:
+                data = np.asarray(data)
+                data = np.uint8(data*255)
             n_data = len(data)
             if n_data % self.n_workers > 0:
                 self.data_server.set_batch_size.remote(
@@ -159,12 +163,12 @@ def predict(self, data, distributed=True, close=False):
             self.data_server.push_data.remote(data)
             while not ray.get(self.data_server.is_complete.remote()):
                 time.sleep(1e-4)
+            if close:
+                self.close()
             return ray.get(self.data_server.pull_results.remote())
-            print("Completed!")
         else:
             return self.model.predict(data)
-        if close:
-            self.close()
+
 
     def close(self, stop_ray=False):
         """

dkeras/dkeras.py

@@ -30,8 +30,14 @@ def worker_task(weights, ds, make_model):
         if packet_id == 'STOP':
             break
         if len(data) > 0:
-            data = np.asarray(data)
-            results = worker_model.predict(data)
-            ds.push.remote(results, packet_id)
+            if packet_id == 'infer_float':
+                data = np.asarray(data)
+                results = worker_model.predict(data)
+                ds.push.remote(results, packet_id)
+            elif packet_id == 'infer_int8':
+                data = np.asarray(data)
+                data = np.float16(data/255)
+                results = worker_model.predict(data)
+                ds.push.remote(results, packet_id)
         else:
             time.sleep(config.WORKER_WAIT_TIME)

dkeras/worker.py

@@ -29,8 +29,14 @@ def worker_task(worker_id, weights, ds, make_model):
         if packet_id == 'STOP':
             break
         if len(data) > 0:
-            data = np.asarray(data)
-            results = worker_model.predict(data, batch_size=batch_size)
-            ds.push.remote(results, packet_id)
+            if packet_id == 'infer_float':
+                data = np.asarray(data)
+                results = worker_model.predict(data, batch_size=batch_size)
+                ds.push.remote(results, packet_id)
+            elif packet_id == 'infer_uint8':
+                data = np.asarray(data)
+                data = np.float16(data/255)
+                results = worker_model.predict(data, batch_size=batch_size)
+                ds.push.remote(results, packet_id)
         else:
             time.sleep(wait_time)

dkeras/workers/worker.py

@@ -0,0 +1,17 @@
+from tensorflow.keras.applications import ResNet50
+from dkeras import dKeras
+import numpy as np
+import time
+import ray
+
+ray.init()
+
+n_data = 100
+data = np.random.uniform(-1, 1, (n_data, 224, 224, 3))
+
+model = dKeras(ResNet50, init_ray=False, wait_for_workers=True, n_workers=4)
+
+start_time = time.time()
+preds = model.predict(data, int8_cvrt=True)
+elapsed = start_time - time.time()
+print(elapsed, n_data/elapsed)