Ramanuzan/multiprocessing issue (#33)

* update cumtom_dict can be pickled * update qsize -> empty, full * update ray 1.7.0 -> 1.8.0 to supports Windows * remove _nomp.py * update docs and readme Co-authored-by: kakao_ent <kakao_ent@kakao-entui-MacBookPro.local>
kakaoenterprise · Nov 6, 2021 · 1d85e72 · 1d85e72
1 parent fa9564c
commit 1d85e72
Show file tree

Hide file tree

Showing 8 changed files with 16 additions and 141 deletions.
diff --git a/README.md b/README.md
@@ -13,15 +13,11 @@ Hello Wo**RL**d!!:hand:  **Join Our Reinforcement Learning framework for Develop
 - Distributed RL algorithms are provided using [ray](https://github.com/ray-project/ray)
 - Benchmark of the algorithms is conducted in many RL environment
 
-## :exclamation:Notification
+## :heavy_check_mark: Tested
 
-Currently, JORLDY is pre-release version. It supports Linux only, but all the scripts can be run on Windows and Mac in the following ways.
-- Windows: Docker or WSL
-- Mac: Docker 
-
-However, you can use only (single, sync_distributed)_train_nomp.py and eval.py on a local environment in Windows and Mac. We will address these issues as soon as possible.
-
-**\* (single, sync_distributed)_train_nomp.py: these scripts don't use multiprocessing library. In detail, the manage process is included in the main process. So it can be a bit slow.**
+| Python |   Windows   |   Mac   |   Linux  |
+| :----: | :---------: | :-----: | :------: |
+|  3.8  | :heavy_check_mark: | :heavy_check_mark: | WSL, Ubuntu 18.04 |
 
 ## :arrow_down: Installation
 

diff --git a/docs/How_to_use.md b/docs/How_to_use.md
@@ -5,7 +5,6 @@
 - sync_distributed_train.py: train with sychronous distributed setting.
 - async_distributed_train.py: train with asychronous distributed setting.
 - eval.py: evaluate with trained agent.
-- (single, sync_distributed)_train_nomp.py: this scripts don't use multiprocessing library. In detail, the manage process is included in the main process. So it can be a bit slow.
 if you want to know the specific process of each script, please refer to [Distributed Architecture](./Distributed_Architecture.md)
 
 ## How to Check Implemented List 

diff --git a/jorldy/async_distributed_train.py b/jorldy/async_distributed_train.py
@@ -26,7 +26,7 @@
     if config.train.distributed_batch_size:
         agent_config["batch_size"] = config.train.distributed_batch_size
 
-    trans_queue = mp.Queue()
+    trans_queue = mp.Queue(10)
     interact_sync_queue = mp.Queue(1)
     result_queue = mp.Queue()
     manage_sync_queue = mp.Queue(1)
@@ -58,7 +58,7 @@
         step, _step, print_stamp, save_stamp = 0, 0, 0, 0
         while step < config.train.run_step:
             transitions = []
-            while (_step == 0 or trans_queue.qsize() > 0) and\
+            while (_step == 0 or not trans_queue.empty()) and\
                   (_step - step < config.train.update_period):
                 _step, _transitions = trans_queue.get()
                 transitions += _transitions

diff --git a/jorldy/manager/config_manager.py b/jorldy/manager/config_manager.py
@@ -60,6 +60,12 @@ class CustomDict(dict):
     def __init__(self, init_dict={}):
         self.update(init_dict)
 
+    def __getstate__(self):
+        return self.__dict__
+
+    def __setstate__(self, d):
+        self.__dict__.update(d)
+
 def type_cast(var):
     try:
         return int(var)

diff --git a/jorldy/process.py b/jorldy/process.py
@@ -13,9 +13,9 @@ def interact_process(DistributedManager, distributed_manager_config,
             delta_t = len(transitions) / num_workers
             step += delta_t
             trans_queue.put((int(step), transitions))
-            if sync_queue.qsize() > 0:
+            if sync_queue.full():
                 distributed_manager.sync(sync_queue.get())
-            while trans_queue.qsize() == 10:
+            while trans_queue.full():
                 time.sleep(0.1)
     except Exception as e:
         traceback.print_exc()
@@ -39,7 +39,7 @@ def manage_process(Agent, agent_config,
     try:
         while step < run_step:
             wait = True
-            while wait or result_queue.qsize() > 0:
+            while wait or not result_queue.empty():
                 _step, result = result_queue.get()
                 metric_manager.append(result)
                 wait = False

diff --git a/jorldy/single_train_nomp.py b/jorldy/single_train_nomp.py
diff --git a/jorldy/sync_distributed_train_nomp.py b/jorldy/sync_distributed_train_nomp.py
diff --git a/requirements.txt b/requirements.txt
@@ -1,6 +1,6 @@
 torch==1.8.1
 tensorboard==2.5.0
-ray==1.7.0
+ray==1.8.0
 opencv-python==4.5.2.52
 pygifsicle==1.0.4
 gym==0.21.0