Threads, Spinning, and Async in ROS 2

1. Create a service client (call it client) using Node.create_client
2. Wait for the service to become available using client.wait_for_service(timeout_sec…)=
3. Issue a request with client.call_async
   • Synchronous calls are generally not well supported in ROS 2, although there is a synchronous calling method in rclpy
4. If you don't want to block
   • Do a little bit of work
   • Check for a result
   • Repeat until a result is ready
5. If you do want to block, wait for the result until it is ready
   • The Service Tutorial uses rclpy.spin_until_future_complete to accomplish this task
   • But the code in the tutorial WILL NOT WORK if used from within another callback function!
   • What is a future anyway? And why is it spinning?

The rest of these notes address how to easily wait for a service call to complete in ROS 2, while also explaining in some detail the concepts and machinery required to make it happen. For quick instruction see Patterns.

• No more than one callback in the same MutuallyExclusiveCallbackGroup can ever be running concurrently.
• By default, all callbacks are placed in the same MutuallyExclusiveCallbackGroup.
• The mutually exclusive property makes it easy to reason about callbacks that may touch the same variables (e.g., global variables or member variables) because callbacks must always run to completion before another one can be called.
  • For example, when a timer is part of a MutuallyExclusiveCallbackGroup you know that the timer callback will always run to completion before it is called again, even if the timer expires sooner or if multiple threads are used
• If you explicitly spin_once in a callback within a MutuallyExclusiveCallbackGroup none of the callbacks in that group will be called, even if there is a pending event.
  • Callbacks in other callback groups can, however, be called

• All callbacks in a ReentrantCallbackGroup can run concurrently, this means that it is possible for a callback to be only partially complete before it is called again
• If you explicitly spin_once from within a callback that is part of a ReentrantCallbackGroup then any callback in that same group can be called
• For example, if a timer is in a ReentrantCallbackGroup, and you spin_once from the timer callback after the timer has expired (e.g., your callback took longer than the timer's period), the timer callback will be called again.

• The python asyncio library, along with the async and await keywords, enables single-threaded cooperative concurrency in python.
• The heart of any asynchronous (async) program is an event loop. In ROS, the event loop is provided by rclpy.spin.
• The event loop is responsible for scheduling the execution of Tasks.
• Tasks are able to run coroutines which are functions that can suspend themselves mid-execution and then resume
• When a coroutine suspends itself mid-execution the event loop can schedule another Task
• A coroutine is created by declaring a function with the async keyword: async def my_coroutine():
• A coroutine can suspend itself using the await keyword: this allows it to either
  1. Call another coroutine and wait for it to finish (e.g., await my_couroutine1())
  2. Wait for the result of a Future object to become available.

A function that may take a long time to compute a result (call it long_function) can instead

1. Create a Future() object (call it future)
2. Use a task to schedule a co-routine (call it data_provider) that has access to future and can provide it with the data, when that data is available
3. Return future to the caller
4. When the co-routine that calls long_function (call it callback) first gets future, future.done() is False, meaning that data is not available
5. callback needs to somehow wait for future.done() to become True and it can then find the data in future.result()
6. One way of waiting for future.done() to become true is to await the future
7. When the co-routine calls await on future it suspends its execution, allowing other co-routines to run, including data_provider
8. When data_provider gets the data, it provides that data to future
   • Now future.done() will be True and future.result() will contain the data
   • The next time callback is scheduled, it will resume from where it suspended execution via await, and the future will now have data.

• In ROS 2, the event loop is provided by rclpy.spin, and callbacks are scheduled by this event loop
• Callbacks in ROS 2 can be functions or co-routines.
• If a callback is a co-routine, it can await on a future
• For a ROS 2 service client, client.call_async() returns a future
• So, if a callback is async it can await the result of client.call_async().
  • It will suspend its operation and return control to the event loop (rclpy.spin)
  • If the callback groups are set up correctly, rclpy.spin() will be able to call the server client callback when the response arrives
  • The future will then be done, and the callback that issued the service request will have the result and be able to continue
  • To get the response from the service use: response = await client.call_async()

• In python asyncio the event loop is created by calling asyncio.run. In ROS 2, an event loop iteration us implemented using an rclpy.executor
• In python asyncio a task is created with asyncio.create_task. In ROS 2, a task is created with rclpy.executor.create_task and is associated with a given executor.
• In python asyncio.Future futures are used. In ROS 2 the futures are rclpy.task.Future objects and are associated with an executors event loop.
• Overall, the ROS 2 task objects behave in the same way as the corresponding python asyncio counterparts. However, ROS 2 facilitates the creation of different types of event loops, and these objects are associated with these loops.
  • For example, in ROS 2 there is a MultiThreadedExecutor that uses multiple threads to process events.

1. Use rclpy.spin in only one place in your program (likely the main entry point)
   • Now you know exactly where the spinning occurs
   • By following the other guidelines you will never need to spin again
   • An exception to this rule is if you need to do some one-time setup when your node starts
     • Then you may want to call services and spin until they are complete
     • Most setup, however, is not one-time. You may want the ability to, for example, reset and re-run the setup.
2. Structure the main execution of your code around a timer callback
   • Usually you want to be issuing commands at a single fixed frequency
   • The timer should advance the progress of your algorithm by one time-step
   • Subscribers do very little work and instead store data that is then used by the timer
   • This setup allows you to easily keep track of the node's state in a single location and allows. some calculations or decisions to take place over multiple timer ticks without needing to pause.
   • If you have a timer running at X hz, it's easy to subdivide that into X/N hz. Multiple timers could also be used, but doing a subdivision can be a good choice because it gives more deterministic performance and control in terms of when the various tasks occur relative to each other.
3. It is also possible to publish directly from a subscriber:
   • This pattern may be appropriate when you want published output to happen as close to the frequency/timing of the incoming messages as possible
   • This pattern provides the least amount of flexibility for, example, handling a flood of messages in an organized manner, and couples the timing of the node to the timing of the publishers it subscribes to

Here is information on two types of service calls: blocking/synchronous and non-blocking/asynchronous. Examples can be found at https://github.com/m-elwin/async_experiments and a walkthrough of the code is provided at here

1. ros2 launch async_experiments experiment.launch.xml experiment:=deadlock
2. The deadlock node starts in the async_client.py:deadlock_entry function.
3. The DeadlockClient constructor (__init__) is called:
   • It creates a service client
   • It creates a timer

4. rclpy.spin is called, which implements an event loop as follows in pseudocode:

   while Node_Has_Not_Exited:
      Check for an event
      If an event has occured, call the appropriate callback
   

5. After some time a timer event occurs and the timer callback (DeadlockClient.timer_callback) is called from the event loop.
6. The timer callback calls the "delay" service using call_async and retrieves a Future object called future
   • This call sends a request to the node running the "delay" service, so that it may generate a response
   • The future keeps track of the status of the response
   • Initially, the response has not been received, so future.done() returns False
7. The timer enters a loop that waits for future.done() to return true
8. Meanwhile, the node implementing the "delay" service does some processing, and sends a response
9. The DeadlockClient node is still looping, waiting for future.done() to be true
   • It will loop forever, because this loop does not call anything that checks for responses that can change the status of the Future() object so future.done() will always be False.

1. ros2 launch async_experiments experiment.launch.xml experiment:=await
2. The await node starts in the async_client.py:await_entry function.
3. The AwaitClient constructor (__init__) is called:
   • A callback group called cbgroup is created
   • A service client called _client is created and assigned to cbgroup
   • A timer is created, it's callback group is the default MutuallyExclusiveCallbackGroup, which is different than cbgroup.
   • Based on this setup, the _client callback can run when the timer_callback has called await and not complted.

4. rclpy.spin is called, which implements an event loop as follows in pseudocode:

   while Node_Has_Not_Exited:
      Check for an event
      If an event has occured, call the appropriate callback
   

5. After some time a timer event occurs and the timer callback (AwaitClient.timer_callback) is called from the event loop.
6. The timer_callback calls the "delay" service (running on the delay_server node) using call_async and retrieves a Future object
   • We refer to this object future in these notes. In the code we don't name this object because we directly await on the return value
   • _client also has a reference to future (since it created future it maintains its own reference to it)
   • The call_async sends a request to the delay_server node so that the "delay" service may generate a response
   • The future keeps track of the status of the response
   • The timer_callback calls await on the future object
   • Initially, the response has not been received, so future.done() returns False, thus the timer_callback function suspends itself, allowing the rclpy.spin event loop to continue processing events.
7. Meanwhile, after receiving the request, the delay_server node begins creating the request

8. Control of the await node is in rclpy.spin:

   while Node_Has_Not_Exited:
      Check for an event
      If an event has occured, call the appropriate callback
   

9. Meanwhile, delay_server node finishes its computation and sends it's response back to the _client
   • The await node event loop receives this event and calls the default callback associated with _client
10. The _client.callback does the following:
    • Store the response in the future
    • Signal that future is done
11. After the service client callback completes the event loop is re-entered.
    • Python detects that future.done() is true and resumes executing the co-routine where future was awaited (the timer_callback)
12. The timer_callback resumes execution from after the await statement
    • Because the future is done() it knows the service call has completed
    • It finishes and the event loop rclpy.spin() is re-entered

• In sequential computing, only one computation happens at a given time. The computations happen in sequence.
  • When using a sequential model, callback groups don't matter (unless you explicitly spin_once from within a callback).
    • Unless spin_once is called, nothing attempts to call the callbacks
    • Each callback will run until completion.
• In parallel computing, multiple computations can happen simultaneously.
  • Callback groups matter, and there are other synchronization issues related to multi-threading
• In concurrent computing, multiple computations can be pending at the same time, but are not executing sequentially.
• For example, consider the following chess-playing scenarios
  • A person playing one chess game, followed by another chess game: this is like sequential computation.
  • Two pairs of people playing two chess games simultaneously: this is like parallel computation.
  • One person playing two chess games against two opponents. First the person makes a move in game 1, then makes a move in game 2. Both games are in progress at the same time, but the person is not making a move in each game simultaneously.

1. A process is an abstraction used by the Linux Kernel (and other operating systems) to segregate memory address space.
   • Code running in each process thinks it has access to the full memory-address space: in reality the kernel maps a processes memory into the physical ram
   • Code running in each process cannot access memory used by other processes without operating-system intervention, increasing stability and security.
   • Programmers do not need to worry about the intricacies of physical memory: to them memory is just a flat contiguous address space
2. Every process has a "main" thread, which executes the machine code within the context of the process's memory address space

1. A thread is subordinate unit of execution within a process, and it is what actually executes the machine code
2. By default each process has a single "main" thread. But it can also create additional threads to run code simultaneously.
   • All threads within a process share the same memory address space
   • Each thread has its own stack (for calling functions and storing local variables)
   • The heap and global variables are shared by threads
   • Unlike processes, threads can directly read and write the memory used by another thread
3. Each thread is scheduled for execution by the kernel
   • If the machine has multiple CPUs then threads can be executed in parallel
   • If there are more threads than CPUs (as is usually the case), the kernel rapidly switches between threads, executing little bits of code each time
     • In this case the code is running concurrently, but it is fast enough to simulate simultaneity
4. If two threads access the same memory and are not coordinated properly, a bug called a race condition can occur.
   • A race condition can result in data corruption
   • It can also result in a deadlock, where no thread can continue executing
5. The exact interleaving of the instructions across multiple threads is non-deterministic because it depends on how the kernel schedules execution, which in turn depends on what else is happening on the computer
   • Non-deterministic bugs in multi-threaded code may be from a race condition
   • Be cautious and have a plan before introducing extra threads into your program
6. - As the operating system switches between threads, it is possible for instructions to be interrupted before completing:
   • Generally, a single python statement maps to multiple machine instructions and thus a statement can be interrupted in the middle of executing
   • Consider i = i + 1. The thread can be interrupted in the middle of reading the value of i, adding 1 to it, or storing the result in i
   • In the meantime, another thread can modify i, leading to incorrect results.
7. Atomic operations finish executing without interruption. Atomic operations guarantee that a complete result will be computed and seen by all threads.
8. Synchronization primitives such as a mutex or a semaphore can be used to coordinate execution between threads

1. The most common python interpreter, CPython, implements a Global Interpreter Lock (GIL)
   • The GIL simplifies the creation of the interpreter while preserving good single-threaded performance
   • It also severely limits multi-threaded performance in python.
2. The GIL prevents multiple python instructions from executing simultaneously on a multi-core system.
   • Thus, multi-threading with CPython is like multi-threading on a single-core CPU: the python code is not executed simultaneously
   • Operations on each thread are interleaved (that is, the python interpreter executes some commands on one thread, then switches to another).
   • The order of this interleaving is generally non-deterministic
3. The GIL does not prevent all race-conditions
   • You still need to synchronize threads that read/write to the same shared variables.
   • Bugs that occur due to specific orderings of operations on multiple threads can still occur
   • Non-atomic python statements can be interrupted before completion
4. The GIL only applies to python bytecode instructions
   • Python code can call C code, and that C code can bypass the GIL
   • Python waits for a system function (e.g., to reading from a file), it is not executing bytecode. Thus, another thread can run while the other thread waits for the system.
   • Thus, multiple-threads in python can improve performance of input/output (I/O) bound (but not CPU bound) python programs but not
     • CPU bound means performance is limited by the available CPU resources
     • I/O bound means performance is limited by input/output operations (such as reading a file)
5. Atomic Operations: The following operations in python are guaranteed to complete once started, prior to another thread being run
   • Reading or writing a single variable of a basic type (int, float, string, etc)
   • Assigning an object to a variable (e.g., x = y)
   • Reading an item from a list
   • Modifying an item in a list
   • Getting an item from a dictionary
   • A complete list of python atomic operations