Respect specified order in composite task (#1126)

With this commit we ensure that operations that are intertwined with
streams don't get executed before streams but rather respect the order
in which they have been specified in the track.

Relates #1112

esrally/driver/runner.py

@@ -2014,6 +2014,10 @@ async def run_stream(self, es, stream, connection_limit):
                 if "stream" in item:
                     streams.append(asyncio.create_task(self.run_stream(es, item["stream"], connection_limit)))
                 elif "operation-type" in item:
+                    # consume all prior streams first
+                    if streams:
+                        await asyncio.gather(*streams)
+                        streams = []
                     op_type = item["operation-type"]
                     if op_type not in self.supported_op_types:
                         raise exceptions.RallyAssertionError(
@@ -2031,6 +2035,7 @@ async def run_stream(self, es, stream, connection_limit):
                     s.cancel()
             raise
 
+        # complete any outstanding streams
         if streams:
             await asyncio.gather(*streams)
 

tests/driver/runner_test.py

@@ -4461,19 +4461,31 @@ async def __aenter__(self):
         async def __call__(self, es, params):
             self.max_value = max(self.max_value, self.current)
             # wait for a short moment to ensure overlap
-            await asyncio.sleep(0.2)
+            await asyncio.sleep(0.1)
 
         async def __aexit__(self, exc_type, exc_val, exc_tb):
             self.current -= 1
             return False
 
+    class CallRecorderRunner:
+        def __init__(self):
+            self.calls = []
+
+        async def __call__(self, es, params):
+            self.calls.append(params["name"])
+            # wait for a short moment to ensure overlap
+            await asyncio.sleep(0.1)
+
     def setUp(self):
         runner.register_default_runners()
         self.counter_runner = CompositeTests.CounterRunner()
+        self.call_recorder_runner = CompositeTests.CallRecorderRunner()
         runner.register_runner("counter", self.counter_runner, async_runner=True)
+        runner.register_runner("call-recorder", self.call_recorder_runner, async_runner=True)
 
     def tearDown(self):
         runner.remove_runner("counter")
+        runner.remove_runner("call-recorder")
 
     @mock.patch("elasticsearch.Elasticsearch")
     @run_async
@@ -4512,6 +4524,75 @@ async def test_execute_multiple_streams(self, es):
                                                              body={},
                                                              params={})
 
+    @mock.patch("elasticsearch.Elasticsearch")
+    @run_async
+    async def test_executes_tasks_in_specified_order(self, es):
+        es.transport.perform_request.return_value = as_future()
+
+        params = {
+            "requests": [
+                {
+                    "name": "initial-call",
+                    "operation-type": "call-recorder",
+                },
+                {
+                    "stream": [
+                        {
+                            "name": "stream-a",
+                            "operation-type": "call-recorder",
+                        }
+                    ]
+                },
+                {
+                    "stream": [
+                        {
+                            "name": "stream-b",
+                            "operation-type": "call-recorder",
+                        }
+                    ]
+                },
+                {
+                    "name": "call-after-stream-ab",
+                    "operation-type": "call-recorder",
+                },
+                {
+                    "stream": [
+                        {
+                            "name": "stream-c",
+                            "operation-type": "call-recorder",
+                        }
+                    ]
+                },
+                {
+                    "stream": [
+                        {
+                            "name": "stream-d",
+                            "operation-type": "call-recorder",
+                        }
+                    ]
+                },
+                {
+                    "name": "call-after-stream-cd",
+                    "operation-type": "call-recorder",
+                },
+
+            ]
+        }
+
+        r = runner.Composite()
+        r.supported_op_types = ["call-recorder"]
+        await r(es, params)
+
+        self.assertEqual([
+            "initial-call",
+            # concurrent
+            "stream-a", "stream-b",
+            "call-after-stream-ab",
+            # concurrent
+            "stream-c", "stream-d",
+            "call-after-stream-cd"
+        ], self.call_recorder_runner.calls)
+
     @mock.patch("elasticsearch.Elasticsearch")
     @run_async
     async def test_limits_connections(self, es):
@@ -4603,7 +4684,6 @@ async def test_rejects_unsupported_operations(self, es):
         self.assertEqual("Unsupported operation-type [bulk]. Use one of [raw-request, sleep].", ctx.exception.args[0])
 
 
-
 class RetryTests(TestCase):
     @run_async
     async def test_is_transparent_on_success_when_no_retries(self):