Fix race condition in eventletutils Event

The threading-compatible eventlet Event class has a race condition on
the wait method. If greenthread A is blocked on the wait, but another
greenthread B calls clear() and then set(), B calls self._event.send(),
but A is waiting on a different eventlet Event which is no longer used
by the oslo.service Event...

To resolve this, when clearing an Event trigger the underlying eventlet
Event immediately, then have the wait() method resume waiting on the new
eventlet Event.

Change-Id: I81579e2977bb965a5398a2cb4e3e24f5671e856a
Co-Authored-By: Victor Stinner <vstinner@redhat.com>
Co-Authored-By: Hervé Beraud <hberaud@redhat.com>
Closes-Bug: #1805706

lower-constraints.txt

@@ -4,6 +4,7 @@ bandit==1.4.0
 coverage==4.0
 ddt==1.0.1
 debtcollector==1.2.0
+eventlet==0.18.2
 extras==1.0.0
 fixtures==3.0.0
 flake8==2.5.5

oslo_utils/eventletutils.py

@@ -24,6 +24,8 @@ import threading
 import warnings
 
 from oslo_utils import importutils
+from oslo_utils import timeutils
+
 
 # These may or may not exist; so carefully import them if we can...
 _eventlet = importutils.try_import('eventlet')
@@ -151,8 +153,11 @@ class EventletEvent(object):
         self.clear()
 
     def clear(self):
+        old_event = getattr(self, "_event", None)
         self._set = False
         self._event = _eventlet.event.Event()
+        if old_event is not None:
+            old_event.send(True)
 
     def is_set(self):
         return self._set
@@ -167,9 +172,15 @@ class EventletEvent(object):
         self._event.send(True)
 
     def wait(self, timeout=None):
-        with _eventlet.timeout.Timeout(timeout, False):
-            self._event.wait()
-        return self.is_set()
+        with timeutils.StopWatch(timeout) as sw:
+            while True:
+                event = self._event
+                with _eventlet.timeout.Timeout(sw.leftover(return_none=True),
+                                               False):
+                    event.wait()
+                    if event is not self._event:
+                        continue
+                return self.is_set()
 
 
 def Event():

oslo_utils/tests/test_eventletutils.py

@@ -15,6 +15,8 @@
 import threading
 import warnings
 
+import eventlet
+from eventlet import greenthread
 import mock
 from oslotest import base as test_base
 import six
@@ -150,3 +152,52 @@ class EventletUtilsTest(test_base.BaseTestCase):
         self.assertEqual(0, mock_eventlet.event.Event().reset.call_count)
         e_event.set()
         self.assertEqual(1, mock_eventlet.event.Event().reset.call_count)
+
+    def test_event_no_timeout(self):
+        event = eventletutils.EventletEvent()
+
+        def thread_a():
+            self.assertTrue(event.wait())
+
+        a = greenthread.spawn(thread_a)
+
+        with eventlet.timeout.Timeout(0.5, False):
+            a.wait()
+            self.fail('wait() timed out')
+
+    def test_event_race(self):
+        event = eventletutils.EventletEvent()
+
+        def thread_a():
+            self.assertTrue(event.wait(2))
+
+        a = greenthread.spawn(thread_a)
+
+        def thread_b():
+            eventlet.sleep(0.1)
+            event.clear()
+            event.set()
+            a.wait()
+
+        b = greenthread.spawn(thread_b)
+        with eventlet.timeout.Timeout(0.5):
+            b.wait()
+
+    def test_event_clear_timeout(self):
+        event = eventletutils.EventletEvent()
+
+        def thread_a():
+            self.assertFalse(event.wait(0.5))
+
+        a = greenthread.spawn(thread_a)
+
+        def thread_b():
+            eventlet.sleep(0.1)
+            event.clear()
+            eventlet.sleep(0.1)
+            event.clear()
+            a.wait()
+
+        b = greenthread.spawn(thread_b)
+        with eventlet.timeout.Timeout(0.7):
+            b.wait()

test-requirements.txt

@@ -4,6 +4,7 @@
 
 hacking!=0.13.0,<0.14,>=0.12.0 # Apache-2.0
 
+eventlet>=0.18.2,!=0.18.3,!=0.20.1,!=0.21.0,!=0.23.0 # MIT
 fixtures>=3.0.0 # Apache-2.0/BSD
 testscenarios>=0.4 # Apache-2.0/BSD
 testtools>=2.2.0 # MIT