Added explicit worker GC (throttled)

distributed/utils.py

@@ -948,7 +948,6 @@ def nbytes(frame, _bytes_like=(bytes, bytearray)):
         else:
             return frame.nbytes
 
-
 @contextmanager
 def time_warn(duration, text):
     start = time()

distributed/worker.py

@@ -153,6 +153,7 @@ def __init__(self, scheduler_ip=None, scheduler_port=None,
         self._last_disk_io = None
         self._last_net_io = None
         self._ipython_kernel = None
+        self._throttledGC = ThrottledGC()
 
         if self.local_dir not in sys.path:
             sys.path.insert(0, self.local_dir)
@@ -648,6 +649,8 @@ def gather(self, comm=None, who_has=None):
                           'keys': missing_keys})
         else:
             self.update_data(data=result, report=False)
+            del result
+            self._throttledGC.collect()
             raise Return({'status': 'OK'})
 
 
@@ -1905,6 +1908,8 @@ def update_who_has(self, who_has):
 
     def release_key(self, key, cause=None, reason=None, report=True):
         try:
+            nbytes_to_free = self.nbytes.get(key)
+
             if key not in self.task_state:
                 return
             state = self.task_state.pop(key)
@@ -1952,6 +1957,12 @@ def release_key(self, key, cause=None, reason=None, report=True):
                 self.batched_stream.send({'op': 'release',
                                           'key': key,
                                           'cause': cause})
+
+            force_gc = None
+            if nbytes_to_free is not None:
+                force_gc = nbytes_to_free > 10 * 2**20
+            self._throttledGC.collect(force_gc)
+
         except CommClosedError:
             pass
         except Exception as e:
@@ -2144,6 +2155,9 @@ def execute(self, key, report=False):
 
             self.ensure_computing()
             self.ensure_communicating()
+
+            del value
+            self._throttledGC.collect()
         except Exception as e:
             if executor_error is e:
                 logger.error("Thread Pool Executor error: %s", e)