Workaround for strange memory leak in Eventlet's Thread class (Fixes #328)

Author: miguelgrinberg

src/engineio/async_drivers/eventlet.py

@@ -1,11 +1,30 @@
 from __future__ import absolute_import
 
-from eventlet.green.threading import Thread, Event
-from eventlet import queue
-from eventlet import sleep
+from eventlet.green.threading import Event
+from eventlet import queue, sleep, spawn
 from eventlet.websocket import WebSocketWSGI as _WebSocketWSGI
 
 
+class EventletThread:  # pragma: no cover
+    """Thread class that uses eventlet green threads.
+
+    Eventlet's own Thread class has a strange bug that causes _DummyThread
+    objects to be created and leaked, since they are never garbage collected.
+    """
+    def __init__(self, target, args=None, kwargs=None):
+        self.target = target
+        self.args = args or ()
+        self.kwargs = kwargs or {}
+        self.g = None
+
+    def start(self):
+        self.g = spawn(self.target, *self.args, **self.kwargs)
+
+    def join(self):
+        if self.g:
+            return self.g.wait()
+
+
 class WebSocketWSGI(_WebSocketWSGI):
     def __init__(self, handler, server):
         try:
@@ -26,7 +45,7 @@ def __call__(self, environ, start_response):
 
 
 _async = {
-    'thread': Thread,
+    'thread': EventletThread,
     'queue': queue.Queue,
     'queue_empty': queue.Empty,
     'event': Event,

tests/common/test_server.py

@@ -126,11 +126,10 @@ def test_async_mode_eventlet(self):
         s = server.Server(async_mode='eventlet')
         assert s.async_mode == 'eventlet'
 
-        from eventlet.green import threading
         from eventlet import queue
         from engineio.async_drivers import eventlet as async_eventlet
 
-        assert s._async['thread'] == threading.Thread
+        assert s._async['thread'] == async_eventlet.EventletThread
         assert s._async['queue'] == queue.Queue
         assert s._async['websocket'] == async_eventlet.WebSocketWSGI
 
@@ -318,7 +317,7 @@ def test_on_event_invalid(self):
             s.on('invalid')
 
     def test_trigger_event(self):
-        s = server.Server()
+        s = server.Server(async_mode='threading')
         f = {}
 
         @s.on('connect')
@@ -1126,7 +1125,7 @@ def test_background_tasks(self):
         def bg_task():
             flag['task'] = True
 
-        s = server.Server()
+        s = server.Server(async_mode='threading')
         task = s.start_background_task(bg_task)
         task.join()
         assert 'task' in flag
@@ -1162,15 +1161,15 @@ def test_log_error_once(self):
         s.logger.info.assert_called_with('foo')
 
     def test_service_task_started(self):
-        s = server.Server(monitor_clients=True)
+        s = server.Server(async_mode='threading', monitor_clients=True)
         s._service_task = mock.MagicMock()
         environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=4'}
         start_response = mock.MagicMock()
         s.handle_request(environ, start_response)
         s._service_task.assert_called_once_with()
 
     def test_shutdown(self):
-        s = server.Server(monitor_clients=True)
+        s = server.Server(async_mode='threading', monitor_clients=True)
         environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=4'}
         start_response = mock.MagicMock()
         s.handle_request(environ, start_response)