Threadsafe library

[silasalves]

Is your feature request related to a problem? Please describe.
My robot's sensors use a library that runs on a different thread than the main thread, causing race conditions with the method void Send(), which throws the following exception:

TCPConnector Exception: System.Net.Sockets.SocketException (0x80004005): Only one usage of each socket address (protocol/network address/port) is normally permitted.

Describe the solution you'd like
It would be great if the ROSConnection class was threadsafe, or if it was possible to have multiple endpoints (one per thread).

Describe alternatives you've considered
I tried using lock() and a Semaphore but that did not help. I am not sure how to proceed.

Update 1: To ensure that only one thread could call the methods I am using (i.e. Send and RegisterPublisher), I created a wrapper that allowed all scripts to store their messages to a queue, and then only one endpoint was calling ProcessPublishers() but that did not work.

I also noticed the errors stats consistently at 75-80 seconds after the beginning of the simulation.

using Unity.Robotics.ROSTCPConnector;
using Unity.Robotics.ROSTCPConnector.MessageGeneration;
using System.Collections.Concurrent;

public class RosWrapper
{
    private static ConcurrentQueue<(string topicName, Message message)> msgBuffer = new ConcurrentQueue<(string, Message)>();

    private static ConcurrentQueue<(string topicName, string messageName)> pubRegBuffer = new ConcurrentQueue<(string, string)>();

    private static readonly object msgBufferLock = new object();

    public void Send(string rosTopicName, Message message)
    {
        msgBuffer.Enqueue((rosTopicName, message));
    }

    public void RegisterPublisher(string rosTopicName, string rosMessageName)
    {
        pubRegBuffer.Enqueue((rosTopicName, rosMessageName));
    }

    public void ProcessPublishers()
    {
        (string topicName, string messageName) regData;
        while(pubRegBuffer.TryDequeue(out regData))
            ROSConnection.instance.RegisterPublisher(regData.topicName, regData.messageName);

        (string topicName, Message message) pubData;
        while(msgBuffer.TryDequeue(out pubData))
            ROSConnection.instance.Send(pubData.topicName, pubData.message);
    }
}

Update 2: I decided to remove all the calls to the instance of ROSConnection and created a very simple "communication class" that just eases access to the message queues. Only one script is now allowed to use ROSConnection.instance. This implementation still does not work. The error continues to show up after 75-80 seconds.

using System.Collections.Concurrent;
using Message = Unity.Robotics.ROSTCPConnector.MessageGeneration.Message;

public class RosWrapper
{
    private static ConcurrentQueue<(string topicName, Message message)> msgBuffer = new ConcurrentQueue<(string, Message)>();

    private static ConcurrentQueue<(string topicName, string messageName)> pubRegBuffer = new ConcurrentQueue<(string, string)>();

    private static readonly object msgBufferLock = new object();

    public static void Send(string rosTopicName, Message message)
    {
        msgBuffer.Enqueue((rosTopicName, message));
    }

    public static void RegisterPublisher(string rosTopicName, string rosMessageName)
    {
        pubRegBuffer.Enqueue((rosTopicName, rosMessageName));
    }

    public static bool TryDequeueSend(out (string topicName, Message message) data)
    {
        return msgBuffer.TryDequeue(out data);
    }

    public static bool TryDequeuePubReg(out (string topicName, string messageName) data)
    {
        return pubRegBuffer.TryDequeue(out data);
    }
}

Update 3: I tried to create a thread dedicated to processing the information, but ROSConnector requires being executed on the main thread:

UnityException: Load can only be called from the main thread.

[mrpropellers]

Hi @silasalves, so to respond to your feature request first: ensuring thread-safety for our API would be a monumental task, as anything that calls UnityEngine/Unity API's is implicitly not thread-safe. In general, anything that operates on GameObjects within a scene is going to have difficulty behaving in a thread-safe manner, as there are specific guarantees a game engine needs to maintain with respect to the timing of modifications to the rendered scene's structure.

That said, you may be able to structure your implementations such that threaded operations don't call/reference unsafe functions/variables. Can you provide a little more detail on what your sensor implementation looks like and why they need to run on a parallel thread?

Using Queues to manage passing of data between threads and the main thread seems like a good approach, so I think you are headed in the right direction, but it looks like you may be calling RegisterPublisher more than once per topic. A given ROS node can only have one registered publisher per topic, so if you are registering duplicate publishers, you'll likely encounter undefined behavior that is unrelated to your threading issues.

[silasalves]

@mrpropellers Thanks for the explanation! I am rather new to Unity, so I didn't know that Unity's API was not thread-safe.

Can you provide a little more detail on what your sensor implementation looks like and why they need to run on a parallel thread?

I am actually using Unity's Sensor SDK to get the sensor (a Lidar, by the way), so I had to implement the ROS interface as a "Node Script" in System Graph, and so far I believe that System Graph runs in a separate thread.

I don't think there is another way of exposing the sensor data without using a Node Script in System Graph. I will investigate that too.

Using Queues to manage passing of data between threads and the main thread seems like a good approach, so I think you are headed in the right direction (...).

I will give Queue another try, but I find it strange that I got that error even after having a single object calling the Send(...) and RegisterPublisher(...) from a single object, as in the code in Updates 1 and 2. All scripts could call RosWrapper.Send(...) and RosWrapper.RegisterPublisher(...), but only one object called RosWrapper.ProcessPublishers(...). Even this setup did not solve the problem.

I also had one subscriber, which I thought might be the cause, but removing it did not solve the error either

(...) but it looks like you may be calling RegisterPublisher more than once per topic.

I just reviewed the code and RegisterPublisher is being called only once per topic (there are 5 different publishers in total). Whenever I disable the sensor, the error stop showing, so I don't think this is related to my problem. Nonetheless, if I continue using the RosWrapper class, I will add a test to ensure a topic is not registered more than once.

[silasalves]

I think I figured it out. I added a call to Thread.Sleep(1) after calling ROSConnection.instance.Send(...) and the problem stopped. The documentation for the TcpClient.Close() has the following remarks:

(...) Based on the LingerState property, the TCP connection may stay open for some time after the Close method is called when data remains to be sent. There is no notification provided when the underlying connection has completed closing.

This explains why the changes in Update 1 and 2 also had problems, even though I was making sure that I was in the main thread. Actually, it was never a race condition; my assumption that the Sensor SDK ran in a separate thread was false. I verified that using Debug.Log("Thread ID: " + System.Threading.Thread.CurrentThread.ManagedThreadId).

Using Thread.Sleep(1) is bad, so I will look for other solutions. So far this seems promising.

Is there any reason for ROSTCPConnector to open a new TcpClient for every interaction with the server, rather than a single permanent connection?

Was there any consideration on using a library like ZeroMQ for dealing with the communication?

---

Edit 1: I have made a small change to RosConnection.Send() that so far has solved the issue for me. I added a test for the TcpClient client state to ensure it is closed before the function ends. That solved the issue for me.

[mrpropellers]

Is there any reason for ROSTCPConnector to open a new TcpClient for every interaction with the server, rather than a single permanent connection?

"historical reasons." PR #94 actually changes this logic to constrain communication to a single connection, although it's still being vetted on dev. If you plan on sending messages frequently, or want to send relatively large messages, it might be worth attempting to switch over to the dev branch to take advantage of this change.

We've actually been doing some testing internally for other communication middlewares like ZeroMQ, but that's ongoing and I'm not sure we've drawn any definitive conclusions yet.

You might want to look into C# threading Events, if you can inject the a call to Set the event after the TCPClient is closed, that would allow you to wait for the Event handle to set instead of sleeping for an arbitrary amount of time. Anyway, it looks like the issue is resolved so I'm going to go ahead and close this ticket. Feel free to re-open it if further discussion would be helpful!

[silasalves]

@mrpropellers Thanks. The issue has been solved, so it's okay to close. =)

If you plan on sending messages frequently, or want to send relatively large messages, it might be worth attempting to switch over to the dev branch to take advantage of this change.

I will try changing to the dev branch. Should I change the ROS-TCP-Endpoint to dev as well or to another branch?

I don't see anyone using Unity/ROS without relying on frequent and/or large messages, so I believe that enabling this change would be beneficial to everyone.

We've actually been doing some testing internally for other communication middlewares like ZeroMQ, but that's ongoing and I'm not sure we've drawn any definitive conclusions yet.

On that note, why not use Siemens' RosBridgeClient as a base instead of writing a whole new stack from scratch? I have used the rosbridge_suite before and never had problems with it.

You might want to look into C# threading Events, if you can inject the a call to Set the event after the TCPClient is closed, that would allow you to wait for the Event handle to set instead of sleeping for an arbitrary amount of time.

I thought about that, but according to the .Net documentation's remarks on the close() method, it states that "there is no notification provided when the underlying connection has completed closing", so I don't think it is possible to set an event for this. Am I wrong?

[mrpropellers]

I will try changing to the dev branch. Should I change the ROS-TCP-Endpoint to dev as well or to another branch?

Yes. The connector and endpoint will both need to be switched to dev. You may want to check your project's package manifest.json to see if there are any Unity Robotics packages locked to specific versions, as well.

On that note, why not use Siemens' RosBridgeClient as a base instead of writing a whole new stack from scratch? I have used the rosbridge_suite before and never had problems with it.

The connector/endpoint are derived from the ROS# tools, but with a key difference that we skip the generic json serialization and serialize the message objects in Unity to match byte-for-byte how they'd be serialized by ROS, so that when they make it to the Endpoint (our "rosbridge") they don't need to be deserialized from json and reserialized, they can just be transparently passed.

I thought about that, but according to the .Net documentation's remarks on the close() method, it states that "there is no notification provided when the underlying connection has completed closing", so I don't think it is possible to set an event for this. Am I wrong?

If you've already investigated this then you have more information than I have!

[silasalves]

Yes. The connector and endpoint will both need to be switched to dev. You may want to check your project's package manifest.json to see if there are any Unity Robotics packages locked to specific versions, as well.

That worked well so far, thanks a lot for the tip! Is this branch eventually going to become the new version?

The connector/endpoint are derived from the ROS# tools, but with a key difference that we skip the generic json serialization and serialize the message objects in Unity to match byte-for-byte how they'd be serialized by ROS, so that when they make it to the Endpoint (our "rosbridge") they don't need to be deserialized from json and reserialized, they can just be transparently passed.

I see. If the messages match ROS messages byte-for-byte, are you going to eventually ditch the end-point and directly interact with ROS nodes? That would be amazing!

[mrpropellers]

That worked well so far, thanks a lot for the tip! Is this branch eventually going to become the new version?

Yep. dev is our "in-development" branch, whereas main is for validated releases, sort of like "nightly" and "stable." We don't have a strong release schedule yet, but the idea is that every once in a while, when we feel confident that dev is stable, we'll merge it into main and update documentation and dependencies to the new version.

are you going to eventually ditch the end-point and directly interact with ROS nodes?

Although ideal, it's unlikely. There's a huge amount of network coordination that roscore is doing that would need to be reimplemented, and our concern is in making more features that are generally useful in many different contexts, many of which may not even depend on ROS1.