fix the thread-safe problem of parl.client (#141)

* fix the thread-safe problem of  parl.client

* yapf

parl/remote/client.py

@@ -202,14 +202,18 @@ class Client(object):
                     logger.error(
                         'Job {} exceeds max memory usage, will stop this job.'.
                         format(job_address))
+                    self.lock.acquire()
                     self.actor_num -= 1
+                    self.lock.release()
                     job_is_alive = False
                 else:
                     time.sleep(remote_constants.HEARTBEAT_INTERVAL_S)
 
             except zmq.error.Again as e:
                 job_is_alive = False
+                self.lock.acquire()
                 self.actor_num -= 1
+                self.lock.release()
 
             except zmq.error.ZMQError as e:
                 break
@@ -248,7 +252,9 @@ class Client(object):
                     check_result = self._check_and_monitor_job(
                         job_heartbeat_address, ping_heartbeat_address)
                     if check_result:
+                        self.lock.acquire()
                         self.actor_num += 1
+                        self.lock.release()
                         return job_address
 
                 # no vacant CPU resources, cannot submit a new job

parl/remote/tests/cluster_monitor_3_test.py

@@ -79,15 +79,29 @@ class TestClusterMonitor(unittest.TestCase):
         time.sleep(1)
         self.assertEqual(20, len(cluster_monitor.data['workers']))
 
+        # check if the number of workers drops by 10
         for i in range(10):
             workers[i].exit()
-        time.sleep(60)
-        self.assertEqual(10, len(cluster_monitor.data['workers']))
+
+        check_flag = False
+        for _ in range(10):
+            if 10 == len(cluster_monitor.data['workers']):
+                check_flag = True
+                break
+            time.sleep(10)
+        self.assertTrue(check_flag)
 
         for i in range(10, 20):
             workers[i].exit()
-        time.sleep(60)
-        self.assertEqual(0, len(cluster_monitor.data['workers']))
+
+        # check if the number of workers drops to 0
+        check_flag = False
+        for _ in range(10):
+            if 0 == len(cluster_monitor.data['workers']):
+                check_flag = True
+                break
+            time.sleep(10)
+        self.assertTrue(check_flag)
 
         master.exit()