Session class is not thread-safe by design

[fhriley]

In Session.onConnect is the following code:

runAsync(() -> {
    for (OnConnectListener listener: mOnConnectListeners) {
        listener.onConnect(this);
    }
}, getExecutor());

Since Session is not currently thread safe, this is implies that the OnConnectListener is responsible for synchronizing on the Session object it is passed before making any calls on it. Was this intended or should the Session class be made thread safe?

[om26er]

Was this intended or should the Session class be made thread safe?

That was a FIXME that I fixed. The inspiration came from autobahn-python implementation, which does the same. I don't think all onConnect listeners need to wait on each other. Think of a case where one listener does a blocking call causing others to wait as well.

but @oberstet will have a better answer than that.

[fhriley]

Sorry, I wasn't clear. I'm not asking if calling the onConnects should be serialized, I'm asking if Session itself needs to be thread safe. Since onConnect is asynchronous, if I wanted to, for example, subscribe to a message when onConnect fired, I would have to do this:

onConnect (Session session) {
    synchronized (session) {
        session.subscribe(...)
    }
}

because Session is not thread safe. If Session were thread safe, I would not have to worry about synchronizing the session. So my question is was it intended to require the synchronized blocks like that or should Session itself be made thread safe? Personally, I lean towards the latter.

[fhriley]

I've become more familiar with the code, and I think this is definitely a problem. Another example is the demo code given in the README. It has the following code:

session.addOnJoinListener(this::demonstrateSubscribe);
session.addOnJoinListener(this::demonstratePublish);
session.addOnJoinListener(this::demonstrateCall);
session.addOnJoinListener(this::demonstrateRegister);

If we look in Session, we can see that these listeners likely execute in parallel:

for (OnJoinListener listener: mOnJoinListeners) {
    futures.add(runAsync(() -> listener.onJoin(this, details), getExecutor()));
}

Each of the this::demonstrate* methods makes calls on the Session object. The Session object methods are not thread safe so we have a thread safety issue. Unless there is an objection, I will make the Session class thread safe and submit a pull request to fix this issue.

[oberstet]

@om26er pls revert this patch. the session class is not supposed to be thread safe (not necessary given the intended style of usage).

[fhriley]

the session class is not supposed to be thread safe (not necessary given the intended style of usage).

Can you explain what you mean by this?

Let me give an example of why it should be thread safe by design. This is a real use case because it is what I am currently working on. My app will be providing real time options quote data. Option chains can have 100s of symbols, and each symbol can have 10s of fields where each field can be updating individually at a high rate. My app will get an "updated field" message from the quote stream and stick it on a thread to be parsed and published for clients. If Session is not thread safe, I will be forced to do the following blocking synchronization:

synchronized(session) {
    session.put(...)
}

And the reason is this in Session.put:

mPublishRequests.put(requestID, new PublishRequest(requestID, future));

which is not thread safe as you've noted.

Obviously, doing the synchronized like this is a huge bottleneck and will introduce unacceptable latencies for my app. Why would the API force this synchronization on me?

[om26er]

@om26er pls revert this patch. the session class is not supposed to be thread safe (not necessary given the intended style of usage).

you might want to reconsider. The original comment of this bug report talked about not running the user facing callbacks async but the solution that was merged only changed the data structures to their thread-safe alternatives. So that's a win-win situation to me.

[oberstet]

@fhriley from what you describe, I'd probably approach it like this: you have a background thread pool of worker threads doing computationally expensive stuff. hence you want the thread-pool to scale on multi-core - and that is a valid use case of threads. The threads can push their results to a thread-safe FIFO. the workers sit at one end of the FIFO and push results into. the main thread sits at the other end. and only that latter thread holds a Session, which it uses to publish WAMP events.

The intended style of using AutobahnJava is: single threaded and asynchronous (CompletableFuture + executor). Making a single Session instance be usable from multiple threads comes at a price: it introduces complexity and overhead. It is unneeded and unwanted.

[oberstet]

@om26er so it is an explicit design choice: ABJ is not thread-safe. it is designed for async, single-threaded style.

[fhriley]

@oberstet:

The threads can push their results to a thread-safe FIFO. the workers sit at one end of the FIFO and push results into. the main thread sits at the other end. and only that latter thread holds a Session, which it uses to publish WAMP events.

Netty, which the API is using, has already implemented this. Why would I implement another queue in front of Netty when Netty is explicitly designed for this very use case??? Not only that, for high ops/low latency, you can't just throw any old FIFO in there. It's hard to do it correctly.

Making a single Session instance be usable from multiple threads comes at a price: it introduces complexity and overhead.

I take it you didn't even look at my commit. It didn't change the complexity at all, and I would argue it actually reduced complexity in a few places. I'll give you the overhead, though. However, what I implemented should have minimal additional overhead in real world use.

it is designed for async, single-threaded style.

This can't be since your API implementation passes the session to multiple threads (this example is from Session.onPreSessionMessage):

for (OnJoinListener listener: mOnJoinListeners) {
    futures.add(runAsync(() -> listener.onJoin(this, details), getExecutor()));
}

Are you really passing the session to multiple threads and then expecting those threads to not use the session object? Of course not, because the README gives example code which uses the Session object on multiple threads. Omer wrote that example code (sorry Omer, not picking on you). If an author of the API code uses the API in unsafe ways, what do you think users are going to do?

Furthermore, the following single threaded code is not safe with your existing Session code:

CompleteableFuture.allOf(session.publish(...), session.publish(...));

I can explain why that is not safe if you need me to.

I'm going to fork this API. I wish I didn't have to, but as is, this API will result in unsafe code, and a lot of extra work for users.

[ampeixoto]

Hi @fhriley . Do you know if you got any issue using your fork that makes Session thread-safe? I have implemented a similar solution, and so far I didn't notice any problem. But I am afraid I could get some subtle bugs in the future due to this change...

[fhriley]

@ampeixoto I abandoned crossbar/autobahn shortly after this. There were other bugs I filed (even really super obvious ones) that they would argue didn't exist. It's been long enough that I have no idea if my fork is good.