Mutex not being released as expected

[rgleim99]

We tried setting up a scenario where uxr_run_session_time(&session_1, 1000) is run in one thread, but publishing happens on a separate thread.

  while (1)
  {
      printf("PUB hello\n");
      // Session 1 publication
      HelloWorld topic_1 = {
          count, "Publisher 1 says hello"
      };
      ucdrBuffer ub_1;
      uint32_t topic_size_1 = HelloWorld_size_of_topic(&topic_1, 0);
      uxr_prepare_output_stream(&session_1, reliable_out_1, datawriter_id_1, &ub_1, topic_size_1);
      HelloWorld_serialize_topic(&ub_1, &topic_1);

      sleep(1);
  }

This was not publishing to the agent as expected. We discovered that the Mutex acquired in uxr_prepare_output_stream(&session_1, reliable_out_1, datawriter_id_1, &ub_1, topic_size_1); was never being released. We found the issue in:

uint16_t uxr_prepare_output_stream(
        uxrSession* session,
        uxrStreamId stream_id,
        uxrObjectId entity_id,
        ucdrBuffer* ub,
        uint32_t len)
{
    uint16_t rv = UXR_INVALID_REQUEST_ID;

    UXR_LOCK_STREAM_ID(session, stream_id);

    size_t payload_size = WRITE_DATA_PAYLOAD_SIZE + len;

    ub->error = !uxr_prepare_stream_to_write_submessage(session, stream_id, payload_size, ub, SUBMESSAGE_ID_WRITE_DATA,
                    FORMAT_DATA);
    if (!ub->error)
    {
        WRITE_DATA_Payload_Data payload;
        rv = uxr_init_base_object_request(&session->info, entity_id, &payload.base);
        (void) uxr_serialize_WRITE_DATA_Payload_Data(ub, &payload);

        OnFullBuffer on_full_buffer = ub->on_full_buffer;
        void* args = ub->args;
        ucdr_init_buffer(ub, ub->iterator, (size_t)(ub->final - ub->iterator));
        ucdr_set_on_full_buffer_callback(ub, on_full_buffer, args);

        UXR_PREPARE_SHARED_MEMORY(session, entity_id, ub, (uint16_t) len, rv);
    }
    else
    {
        UXR_UNLOCK_STREAM_ID(session, stream_id);
    }

    return rv;
}

The Mutex is only released/unlocked if an error happens. Shouldn't this be released in any case at the end of the function?

Once we removed the else case and just called UXR_UNLOCK_STREAM_ID(session, stream_id); everytime, then the system operated as expected.

Is there some other reason that the UXR_UNLOCK_STREAM_ID(session, stream_id); is called only in the else case?

Proposed fix:

uint16_t uxr_prepare_output_stream(
        uxrSession* session,
        uxrStreamId stream_id,
        uxrObjectId entity_id,
        ucdrBuffer* ub,
        uint32_t len)
{
    uint16_t rv = UXR_INVALID_REQUEST_ID;

    UXR_LOCK_STREAM_ID(session, stream_id);

    size_t payload_size = WRITE_DATA_PAYLOAD_SIZE + len;

    ub->error = !uxr_prepare_stream_to_write_submessage(session, stream_id, payload_size, ub, SUBMESSAGE_ID_WRITE_DATA,
                    FORMAT_DATA);
    if (!ub->error)
    {
        WRITE_DATA_Payload_Data payload;
        rv = uxr_init_base_object_request(&session->info, entity_id, &payload.base);
        (void) uxr_serialize_WRITE_DATA_Payload_Data(ub, &payload);

        OnFullBuffer on_full_buffer = ub->on_full_buffer;
        void* args = ub->args;
        ucdr_init_buffer(ub, ub->iterator, (size_t)(ub->final - ub->iterator));
        ucdr_set_on_full_buffer_callback(ub, on_full_buffer, args);

        UXR_PREPARE_SHARED_MEMORY(session, entity_id, ub, (uint16_t) len, rv);
    }

    UXR_UNLOCK_STREAM_ID(session, stream_id);

    return rv;
}

[pablogs9]

Hello @rgleim99 , sorry if this is not well documented or has examples, but when preparing the buffer the session lock is not unlocked on purpose because after the "buffer preparation" is expected that the user writes the payload in the prepared buffer. So it is the user responsible for unlocking the session.

This is why the session is unlocked only when the buffer preparation fails.

You can see an example of the usage at the application level here: https://github.com/micro-ROS/rmw_microxrcedds/blob/bc4eb312ac4601a4137c35f4a56b9b83b4b18339/rmw_microxrcedds_c/src/rmw_publish.c#L76

[ls6777]

I see. Maybe this should turn into a feature request or improvement suggestion then. Relying on the user to know when to LOCK or UNLOCK is not ideal and is generally considered bad practice.

[pablogs9]

IMO this shall be documented because it is completely required that the used unlocks the "prepared buffer lock", or that the user notifies the library that the buffer was written.

[jpardeiro]

@pablogs9 I think the behavior should definitely be documented and shown in the examples. The current examples are functional because the mutexes are created as recursive, otherwise, you would be getting a deadlock when trying to call the run session function.

[pablogs9]

@jpardeiro AFAIK there is no multithreading example in this repo.

[jpardeiro]

@pablogs9 even in single thread, if the mutexes wouldn't be declared as recursive mutexes here, the examples would not work because they don't release the mutex. Instead, they just rely on the recursive configuration to ensure the mutex can be acquired if the current mutex holder is in the same thread.

[pablogs9]

@jpardeiro Yes, I know, but I meant that the examples of this repo were not written to be run in a multithread environment. Unfortunately, the multithread support was written for micro-ROS support and it is used there, but it was not properly documented for the usage within the bare XRCE-DDS Client library.

I will add this to the backlog, but we do not have much time right now.