It may appear that asynchronous subscriptions are not concurrent

[Kimhlo]

I used the following code to test the efficiency of rclrs's async subscription. I found that consuming large-volume messages (such as image messages: 1843200 bytes) was time-consuming, and messages of other topics could not be consumed during this period.I tried the callback in Box mode, but the result was not improved.What is the reason?How can I solve this problem?

#[tokio::main]
async fn main() -> Result<(), Box> {
let mut executor = Context::default_from_env()?.create_basic_executor();
let node = executor.create_node(NODE_NAME)?;

let joint_state_topic = "/joint_state";
let _joint_state_subscription = node.create_async_subscription::<JointState, _>(
    joint_state_topic.keep_all(),
    move |msg: JointState| async move {
        logger_info!(
            "Received jonit state msg from topic={}, time_stamp={}.{}",
            joint_state_topic,
            msg.header.stamp.sec,
            msg.header.stamp.nanosec
        );
    },
)?;

let image_topic = "/rgbd_camera/head_front/depth";
let _image_subscription = node.create_async_subscription::<Image, _>(
    image_topic.keep_all(),
    move |msg: Image| async move {
    // move |msg: Box<Image>| async move {
        logger_info!(
            "Received image msg from topic={}, time_stamp={}.{}",
            image_topic,
            msg.header.stamp.sec,
            msg.header.stamp.nanosec
        );
    },
)?;

executor
    .spin(SpinOptions::default())
    .first_error()
    .map_err(|err| err.into())

}

here are the logs：
[INFO] [1763032042.307860435] Received jonit state msg from topic=/joint_state, time_stamp=1756784135.751376032
[INFO] [1763032042.307985049] Received jonit state msg from topic=/joint_state, time_stamp=1756784135.761748320

// here const 50 ms to consume Image message, and no JointState was consumed
[INFO] [1763032042.358362197] Received image msg from topic=/rgbd_camera/head_front/depth, time_stamp=1756784135.673184512

// blocked messages of another topic are continuously consumed here
[INFO] [1763032042.358461875] Received jonit state msg from topic=/joint_state, time_stamp=1756784135.771320896
[INFO] [1763032042.358473256] Received jonit state msg from topic=/joint_state, time_stamp=1756784135.781257760
[INFO] [1763032042.358528479] Received jonit state msg from topic=/joint_state, time_stamp=1756784135.791177792
[INFO] [1763032042.358737191] Received jonit state msg from topic=/joint_state, time_stamp=1756784135.801259392
[INFO] [1763032042.358903182] Received jonit state msg from topic=/joint_state, time_stamp=1756784135.812630112

// seems recovered here
[INFO] [1763032042.365064485] Received jonit state msg from topic=/joint_state, time_stamp=1756784135.820907936
[INFO] [1763032042.376233981] Received jonit state msg from topic=/joint_state, time_stamp=1756784135.831356288
[INFO] [1763032042.384994048] Received jonit state msg from topic=/joint_state, time_stamp=1756784135.841266144

// here const 50 ms to consume image message again
[INFO] [1763032042.434110913] Received image msg from topic=/rgbd_camera/head_front/depth, time_stamp=1756784135.740388608
[INFO] [1763032042.434509701] Received jonit state msg from topic=/joint_state, time_stamp=1756784135.851205408
[INFO] [1763032042.434720777] Received jonit state msg from topic=/joint_state, time_stamp=1756784135.861171616
[INFO] [1763032042.435392546] Received jonit state msg from topic=/joint_state, time_stamp=1756784135.871180160
[INFO] [1763032042.435612609] Received jonit state msg from topic=/joint_state, time_stamp=1756784135.881273312

[mxgrey]

First it's important to note the difference between concurrency and parallelism. Concurrency in async Rust means that whenever you use .await inside of an async function, that function will yield the thread it's running on so that another function can run on that thread until an update arrives for the .await. Concurrency in JavaScript and Python is typically single-threaded. In Rust your async executor determines whether your concurrency is single-threaded or multi-threaded.

In rclrs we have have an ExecutorRuntime trait which makes it possible to implement different Executors that could have different characteristics. For example you could have an Executor that uses a thread pool to run all the callbacks in parallel.

Right now we only offer the "basic executor" out of the box, which has the following characteristics:

• All Node callbacks (e.g. Node::create_subscription, Node::create_async_subscription, Node::create_service, Node::create_async_service) are run in a single-threaded concurrent executor, which allows these callbacks to be async (concurrent) but does not allow them to execute in parallel, meaning only one callback at a time is able to make progress when not "awaiting" something.
• All Worker callbacks (e.g. Worker::create_subscription, Worker::create_service) are run on a thread that is dedicated to that specific worker. The worker threads are separate from the concurrent executor thread.

In your case, if you have a high volume subscription and you want to make sure it can be processed without blocking other subscriptions, you should create a worker and then create a subscription with that worker for the high-volume subscription. Then your other subscriptions can be done directly on the node or on a separate worker so they can run in parallel.

Your subscription would look something like this:

let _joint_state_subscription = node
    .create_worker(())
    .create_subscription::<JointState, _>(
    joint_state_topic.keep_all(),
    move |msg: JointState| async move {
        logger_info!(
            "Received jonit state msg from topic={}, time_stamp={}.{}",
            joint_state_topic,
            msg.header.stamp.sec,
            msg.header.stamp.nanosec
        );
    },
)?;

I tried the callback in Box mode

This won't have a significant impact on the performance or change any other characteristics of the callback. I think we mostly provide this API to mimic the std::unique_ptr<Msg> API in rclcpp.

I'll also note that if you aren't using .await in your callback then there's no need to use node.create_async_subscription. The behavior of node.create_subscription is exactly the same as what you're doing in your example, but there will be less noise (e.g. no need for an async move { ... }).