Respect specified order in composite task

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):