Fixed memory leak in async/thread executors

Author: syrusakbary

graphql/execution/executor.py

@@ -105,6 +105,10 @@ def on_resolve(data):
     if not return_promise:
         exe_context.executor.wait_until_finished()
         return promise.get()
+    else:
+        clean = getattr(exe_context.executor, 'clean', None)
+        if clean:
+            clean()
 
     return promise
 

graphql/execution/executors/asyncio.py

@@ -52,6 +52,9 @@ def wait_until_finished(self):
             self.futures = []
             self.loop.run_until_complete(wait(futures))
 
+    def clean(self):
+        self.futures = []
+
     def execute(self, fn, *args, **kwargs):
         result = fn(*args, **kwargs)
         if isinstance(result, Future) or iscoroutine(result):

graphql/execution/executors/gevent.py

@@ -12,8 +12,13 @@ def __init__(self):
         self.jobs = []
 
     def wait_until_finished(self):
-        [j.join() for j in self.jobs]
         # gevent.joinall(self.jobs)
+        while self.jobs:
+            jobs = self.jobs
+            self.jobs = []
+            [j.join() for j in jobs]
+
+    def clean(self):
         self.jobs = []
 
     def execute(self, fn, *args, **kwargs):

graphql/execution/executors/process.py

@@ -17,11 +17,16 @@ def __init__(self):
         self.q = Queue()
 
     def wait_until_finished(self):
-        for _process in self.processes:
-            _process.join()
+        while self.processes:
+            processes = self.processes
+            self.processes = []
+            [_process.join() for _process in processes]
         self.q.close()
         self.q.join_thread()
 
+    def clean(self):
+        self.processes = []
+
     def execute(self, fn, *args, **kwargs):
         promise = Promise()
 

graphql/execution/executors/sync.py

@@ -3,5 +3,8 @@ class SyncExecutor(object):
     def wait_until_finished(self):
         pass
 
+    def clean(self):
+        pass
+
     def execute(self, fn, *args, **kwargs):
         return fn(*args, **kwargs)

graphql/execution/executors/thread.py

@@ -19,8 +19,13 @@ def __init__(self, pool=False):
             self.execute = self.execute_in_thread
 
     def wait_until_finished(self):
-        for thread in self.threads:
-            thread.join()
+        while self.threads:
+            threads = self.threads
+            self.threads = []
+            [thread.join() for thread in threads]
+
+    def clean(self):
+        self.threads = []
 
     def execute_in_thread(self, fn, *args, **kwargs):
         promise = Promise()