Remove LZ4 compression

Author: Sharad24

actorQ/d4pg/actor_main.py

@@ -33,8 +33,6 @@
 import dm_env
 import gc
 import gym
-import lz4.block
-import lz4.frame
 import numpy as np
 import numpy as np
 import pickle
@@ -81,21 +79,15 @@ def get_variables(self, names):
       d = [x.tobytes() for x in sample.data]  
 
       try:
-        #decoded = [pickle.loads(lz4.frame.decompress(x.tobytes() +  b'\x00\x00\x00\x00')) for x in sample.data]    
-        #decoded = [pickle.loads(zlib.decompress(x.tobytes())) for x in sample.data]
         if self.args["compress"]:
           d = [zlib.decompress(x) for x in d]
-          #d = [lz4.frame.decompress(x +  b'\x00\x00\x00\x00') for x in d]
         tdecompress = time.time()
         decoded = [pickle.loads(x) for x in d]
-        #decoded = [pickle.loads(x) for x in d]    
-        #decoded = [pickle.loads(zlib.decompress(x.item())) for x in sample.data]
-        #decoded = [pickle.loads(x.tobytes()) for x in sample.data]
         tdecode = time.time()
         print("Pull time: %f, Decompress/tobytes time: %f, Deserialize time: %f" % (tend-tstart, tdecompress-tend, tdecode-tdecompress))
         return decoded
       except:
-        #  traceback.print_exc()
+         traceback.print_exc()
         pass
       return []
 

actorQ/d4pg/learner_main.py

@@ -33,9 +33,6 @@
 import pickle
 from multiprocessing import Pool
 from multiprocessing.pool import ThreadPool
-import lz4.frame
-import lz4.block
-#import pytorch_actors
 import torch
 import zstd
 import msgpack
@@ -151,15 +148,9 @@ def broadcast_shutdown(should_shutdown):
     steps = 0
 
     def broadcast_variables(weights):
-        #learner.client.insert([tf2_utils.to_numpy(v) for v in 
-        #                       agent_networks["policy"].variables], 
-        #{args.model_table_name : 1.0})
-        #learner.client.insert(weights, {args.model_table_name: 1.0})
         if weights is None:
-          #weights = [tf2_utils.to_numpy(v) for v in agent_networks["policy"].variables]
           weights = [tf2_utils.to_numpy(v) for v in learner.learner._policy_network.variables]  
         learner.client.insert(weights, {args.broadcaster_table_name: 1.0})
-        #learner.client.insert(weights, {args.model_table_name: 1.0})
 
     broadcast_shutdown(0)
     variable_broadcaster = PeriodicBroadcaster(broadcast_variables)
@@ -170,14 +161,6 @@ def broadcast_variables(weights):
         with tf.device(args.learner_device_placement):
             learner.learner.step()
 
-            #if i % 50 == 0:
-            #  #  #weights = [tf2_utils.to_numpy(v) for v in agent_networks["policy"].variables]
-            #  weights = [tf2_utils.to_numpy(v) for v in learner.learner._policy_network.variables]
-            #  #  #variable_broadcaster.update(weights)
-            #  broadcast_variables(weights)
-            #  #  #  #  pass
-            
-            #broadcast_variables(None)
             variable_broadcaster.update(None)
 
 

actorQ/d4pg/parameter_broadcaster.py

@@ -34,8 +34,6 @@
 import pickle
 from multiprocessing import Pool
 from multiprocessing.pool import ThreadPool
-import lz4.frame
-import lz4.block
 import pytorch_actors
 import torch
 import zstd
@@ -78,9 +76,6 @@ def get_weights_to_broadcast(client, broadcast_table_name):
     client = reverb.Client(address)
 
     def quantize_and_broadcast_weights(weights, id):
-        #client.insert(weights, 
-        #              {args.model_table_name : 1.0})
-        #return
 
         print("Broadcasting weights", time.time())
 
@@ -108,19 +103,11 @@ def quantize_and_broadcast_weights(weights, id):
         state_dict["id"] = id
 
         # Send over packed params to avoid overhead
-        #for name, child in quantized_actor._modules.items():
-        #    if type(child) == torch.nn.quantized.dynamic.modules.linear.Linear:
-        #        d_name = "%s._packed_params._packed_params.dumped" % name
-        #        state_dict[d_name] = pickle.dumps(child._packed_params._packed_params)
         tstart = time.time()
         weights = [pickle.dumps(state_dict)]
         if args.compress:
             weights = [zlib.compress(x) for x in weights]
-            #weights = [lz4.frame.compress(x) for x in weights]
-        #weights = [lz4.frame.compress(pickle.dumps(state_dict))]
         weights = [np.fromstring(x, dtype=np.uint8) for x in weights]
-        #weights = [zlib.compress(pickle.dumps(state_dict), level=1)]
-        #weights = [np.fromstring(pickle.dumps(state_dict), dtype=np.uint8)]
         print("Compress %f" % (time.time()-tstart))
 
 

actorQ/dqn/actor_main.py

@@ -47,8 +47,6 @@
 import torch
 import pickle
 import zlib
-import lz4.frame
-import lz4.block
 import time
 import zstd
 import gc
@@ -80,16 +78,10 @@ def get_variables(self, names):
       d = [x.tobytes() for x in sample.data]  
 
       try:
-        #decoded = [pickle.loads(lz4.frame.decompress(x.tobytes() +  b'\x00\x00\x00\x00')) for x in sample.data]    
-        #decoded = [pickle.loads(zlib.decompress(x.tobytes())) for x in sample.data]
         if self.args["compress"]:
           d = [zlib.decompress(x) for x in d]
-          #d = [lz4.frame.decompress(x +  b'\x00\x00\x00\x00') for x in d]
         tdecompress = time.time()
         decoded = [pickle.loads(x) for x in d]
-        #decoded = [pickle.loads(x) for x in d]    
-        #decoded = [pickle.loads(zlib.decompress(x.item())) for x in sample.data]
-        #decoded = [pickle.loads(x.tobytes()) for x in sample.data]
         tdecode = time.time()
         print("Pull time: %f, Decompress/tobytes time: %f, Deserialize time: %f" % (tend-tstart, tdecompress-tend, tdecode-tdecompress))
         return decoded

actorQ/dqn/learner_main.py

@@ -34,8 +34,6 @@
 import pickle
 from multiprocessing import Pool
 from multiprocessing.pool import ThreadPool
-import lz4.frame
-import lz4.block
 import pytorch_actors
 import torch
 import zstd

actorQ/dqn/parameter_broadcaster.py

@@ -34,8 +34,6 @@
 import pickle
 from multiprocessing import Pool
 from multiprocessing.pool import ThreadPool
-import lz4.frame
-import lz4.block
 import pytorch_actors
 import torch
 import zstd
@@ -76,10 +74,6 @@ def get_weights_to_broadcast(client, broadcast_table_name):
     client = reverb.Client(address)
 
     def quantize_and_broadcast_weights(weights, id):
-        #client.insert(weights, 
-        #              {args.model_table_name : 1.0})
-        #return
-
         print("Broadcasting weights", time.time())
 
         # Quantize weights artificially
@@ -106,19 +100,11 @@ def quantize_and_broadcast_weights(weights, id):
         state_dict["id"] = id
 
         # Send over packed params to avoid overhead
-        #for name, child in quantized_actor._modules.items():
-        #    if type(child) == torch.nn.quantized.dynamic.modules.linear.Linear:
-        #        d_name = "%s._packed_params._packed_params.dumped" % name
-        #        state_dict[d_name] = pickle.dumps(child._packed_params._packed_params)
         tstart = time.time()
         weights = [pickle.dumps(state_dict)]
         if args.compress:
             weights = [zlib.compress(x) for x in weights]
-            #weights = [lz4.frame.compress(x) for x in weights]
-        #weights = [lz4.frame.compress(pickle.dumps(state_dict))]
         weights = [np.fromstring(x, dtype=np.uint8) for x in weights]
-        #weights = [zlib.compress(pickle.dumps(state_dict), level=1)]
-        #weights = [np.fromstring(pickle.dumps(state_dict), dtype=np.uint8)]
         print("Compress %f" % (time.time()-tstart))