Ensures that some assumptions are true.

It's documented, the application consumer must not use wait before stop.
but this is not enforced, so enforce it

Also the code assume start/stop/wait are called from the same thread,
but this is not enforced, so enforce it.

A common broken usage is:

    server = oslo.messaging.get_rpc_server(..., executor='eventlet')
    t = threading.Thread(target=server.start)
    t.daemon = True
    t.start()
    ...foobar code...
    server.stop()
    server.wait()

With monkey patching, start() will do a context switch and then stop()
is called but start is unfinished, that can cause unexpected behavior.

This patch fixes these issues by making all of this explicit.

Closes-bug: #1465850
Closes-bug: #1466001

Change-Id: I0fc1717e3118bc1cd7b9cd0ccc072251cfb2c038

1 files changed, 20 insertions, 0 deletions

diff --git a/oslo_messaging/_utils.py b/oslo_messaging/_utils.py
index ddec6d7..1c816de 100644
--- a/oslo_messaging/_utils.py
+++ b/oslo_messaging/_utils.py
@@ -14,6 +14,7 @@
 #    under the License.
 
 import logging
+import threading
 
 LOG = logging.getLogger(__name__)
 
@@ -94,3 +95,22 @@ class DispatcherExecutorContext(object):
         # else
         if self._post is not None:
             self._post(self._incoming, self._result)
+
+
+def fetch_current_thread_functor():
+    # Until https://github.com/eventlet/eventlet/issues/172 is resolved
+    # or addressed we have to use complicated workaround to get a object
+    # that will not be recycled; the usage of threading.current_thread()
+    # doesn't appear to currently be monkey patched and therefore isn't
+    # reliable to use (and breaks badly when used as all threads share
+    # the same current_thread() object)...
+    try:
+        import eventlet
+        from eventlet import patcher
+        green_threaded = patcher.is_monkey_patched('thread')
+    except ImportError:
+        green_threaded = False
+    if green_threaded:
+        return lambda: eventlet.getcurrent()
+    else:
+        return lambda: threading.current_thread()