faster serialize

tensorpack/RL/simulator.py

@@ -43,7 +43,7 @@ class SimulatorProcess(multiprocessing.Process):
         context = zmq.Context()
         c2s_socket = context.socket(zmq.DEALER)
         c2s_socket.identity = 'simulator-{}'.format(self.idx)
-        #c2s_socket.set_hwm(2)
+        c2s_socket.set_hwm(2)
         c2s_socket.connect(self.c2s)
 
         s2c_socket = context.socket(zmq.DEALER)
@@ -54,8 +54,7 @@ class SimulatorProcess(multiprocessing.Process):
         #cnt = 0
         while True:
             state = player.current_state()
-            #c2s_socket.send(dumps(state), copy=False)
-            c2s_socket.send('h')
+            c2s_socket.send(dumps(state), copy=False)
             #with total_timer('client recv_action'):
             data = s2c_socket.recv(copy=False)
             action = loads(data)
@@ -126,8 +125,7 @@ class SimulatorMaster(threading.Thread):
             client = self.clients[ident]
             client.protocol_state = 1 - client.protocol_state   # first flip the state
             if not client.protocol_state == 0:   # state-action
-                #state = loads(msg)
-                state = np.zeros((84, 84, 4), dtype='float32')
+                state = loads(msg)
                 self._on_state(state, ident)
             else:       # reward-response
                 reward, isOver = loads(msg)

tensorpack/predict/concurrency.py

@@ -112,6 +112,7 @@ class PredictorWorkerThread(threading.Thread):
             batched, futures = fetch()
             #print "batched size: ", len(batched), "queuesize: ", self.queue.qsize()
             outputs = self.func([batched])
+            # debug, for speed testing
             #if self.xxx is None:
                 #outputs = self.func([batched])
                 #self.xxx = outputs

tensorpack/utils/serialize.py

@@ -3,17 +3,17 @@
 # File: serialize.py
 # Author: Yuxin Wu <ppwwyyxxc@gmail.com>
 
-#import msgpack
-#import msgpack_numpy
-#msgpack_numpy.patch()
-import dill
+import msgpack
+import msgpack_numpy
+msgpack_numpy.patch()
+#import dill
 
 __all__ = ['loads', 'dumps']
 
 def dumps(obj):
-    return dill.dumps(obj)
-    #return msgpack.dumps(obj, use_bin_type=True)
+    #return dill.dumps(obj)
+    return msgpack.dumps(obj, use_bin_type=True)
 
 def loads(buf):
-    return dill.loads(buf)
-    #return msgpack.loads(buf)
+    #return dill.loads(buf)
+    return msgpack.loads(buf)