Merge pull request #2654 from locustio/fix-issue-with-concurrent-custom-messages-and-docs

Fix issue with concurrent custom messages and docs

Author: cyberw

Diff for: docs/running-distributed.rst

@@ -137,15 +137,13 @@ the messages will simply be handled by the runner that sends them.
     Using the default options while registering a message handler will run the listener function
     in a **blocking** way, resulting in the heartbeat and other messages being delayed for the amount
     of the execution.
-    If it is known that the listener function will handle time-intensive tasks, it is possible to register the
-    function as **concurrent** (as a separate greenlet).
+    If you think that your message handler will need to run for more than a second then you can register it
+    as **concurrent**. Locust will then make it run in its own greenlet. Note that these greenlets will never 
+    be join():ed.
 
     .. code-block::
         environment.runner.register_message('test_users', setup_test_users, concurrent=True)
 
-    Please use this feature with care, as otherwise it could result in greenlets running and influencing
-    the running loadtest.
-
 For more details, see the `complete example <https://github.com/locustio/locust/tree/master/examples/custom_messages.py>`_.
 
 

Diff for: examples/custom_messages.py

@@ -9,11 +9,8 @@
 def setup_test_users(environment, msg, **kwargs):
     # Fired when the worker receives a message of type 'test_users'
     usernames.extend(map(lambda u: u["name"], msg.data))
-    # Even though "acknowledge_concurrent_users" was sent first, "acknowledge_users"
-    # will print its statement first, as "acknowledge_concurrent_users" was registered
-    # running concurrently, and therefore not blocking other messages.
-    environment.runner.send_message("concurrent_message", "This is a non blocking message")
     environment.runner.send_message("acknowledge_users", f"Thanks for the {len(msg.data)} users!")
+    environment.runner.send_message("concurrent_message", "Message to concurrent handler")
 
 
 def on_acknowledge(msg, **kwargs):
@@ -22,8 +19,9 @@ def on_acknowledge(msg, **kwargs):
 
 
 def on_concurrent_message(msg, **kwargs):
-    gevent.sleep(10)
-    print(msg.data)
+    print(f"concurrent_message received with data: '{msg.data}'")
+    gevent.sleep(10)  # if this handler was run with concurrent=False it would halt the message handling loop in locust
+    print("finished processing concurrent_message")
 
 
 @events.init.add_listener

Diff for: locust/runners.py

@@ -1143,7 +1143,7 @@ def client_listener(self) -> NoReturn:
                     if not concurrent:
                         listener(environment=self.environment, msg=msg)
                     else:
-                        gevent.spawn(listener, self.environment, msg)
+                        gevent.spawn(listener, environment=self.environment, msg=msg)
                 except Exception:
                     logging.error(f"Uncaught exception in handler for {msg.type}\n{traceback.format_exc()}")