Executor executes the tasks in FIFO order. (#1304)

* Executor executes the tasks in FIFO order.

Signed-off-by: Tomoya Fujita <[email protected]>

* comment fix to use warning doc section.

Signed-off-by: Tomoya Fujita <[email protected]>

---------

Signed-off-by: Tomoya Fujita <[email protected]>

rclpy/rclpy/executors.py

@@ -210,6 +210,9 @@ def create_task(self, callback: Union[Callable, Coroutine], *args, **kwargs) ->
 
         Arguments to this function are passed to the callback.
 
+        .. warning:: Created task is queued in the executor in FIFO order,
+           but users should not rely on the task execution order.
+
         :param callback: A callback to be run in the executor.
         """
         task = Task(callback, args, kwargs, executor=self)
@@ -569,7 +572,7 @@ def _wait_for_ready_callbacks(
             with self._tasks_lock:
                 tasks = list(self._tasks)
             if tasks:
-                for task, entity, node in reversed(tasks):
+                for task, entity, node in tasks:
                     if (not task.executing() and not task.done() and
                             (node is None or node in nodes_to_use)):
                         yielded_work = True

rclpy/test/test_executor.py

@@ -270,7 +270,7 @@ def func():
         self.assertTrue(future.done())
         self.assertEqual('Sentinel Result', future.result())
 
-    def test_create_task_dependent_coroutines(self):
+    def test_create_task_fifo_order(self):
         self.assertIsNotNone(self.node.handle)
         executor = SingleThreadedExecutor(context=self.context)
         executor.add_node(self.node)
@@ -281,25 +281,51 @@ async def coro1():
         future1 = executor.create_task(coro1)
 
         async def coro2():
-            nonlocal future1
-            await future1
             return 'Sentinel Result 2'
 
         future2 = executor.create_task(coro2)
 
-        # Coro2 is newest task, so it gets to await future1 in this spin
-        executor.spin_once(timeout_sec=0)
-        # Coro1 execs in this spin
+        # Coro1 is the 1st task, so it gets executed in this spin
         executor.spin_once(timeout_sec=0)
         self.assertTrue(future1.done())
         self.assertEqual('Sentinel Result 1', future1.result())
         self.assertFalse(future2.done())
 
-        # Coro2 passes the await step here (timeout change forces new generator)
-        executor.spin_once(timeout_sec=1)
+        # Coro2 is the next in the queue, so it gets executed in this spin
+        executor.spin_once(timeout_sec=0)
+        self.assertTrue(future2.done())
+        self.assertEqual('Sentinel Result 2', future2.result())
+
+    def test_create_task_dependent_coroutines(self):
+        self.assertIsNotNone(self.node.handle)
+        executor = SingleThreadedExecutor(context=self.context)
+        executor.add_node(self.node)
+
+        async def coro1():
+            nonlocal future2
+            await future2
+            return 'Sentinel Result 1'
+
+        future1 = executor.create_task(coro1)
+
+        async def coro2():
+            return 'Sentinel Result 2'
+
+        future2 = executor.create_task(coro2)
+
+        # Coro1 is the 1st task, so it gets to await future2 in this spin
+        executor.spin_once(timeout_sec=0)
+        # Coro2 execs in this spin
+        executor.spin_once(timeout_sec=0)
+        self.assertFalse(future1.done())
         self.assertTrue(future2.done())
         self.assertEqual('Sentinel Result 2', future2.result())
 
+        # Coro1 passes the await step here (timeout change forces new generator)
+        executor.spin_once(timeout_sec=1)
+        self.assertTrue(future1.done())
+        self.assertEqual('Sentinel Result 1', future1.result())
+
     def test_create_task_during_spin(self):
         self.assertIsNotNone(self.node.handle)
         executor = SingleThreadedExecutor(context=self.context)