Thread-unsafe issue if we set ack=True in SubscriptionListener about stompest

[nikipore]

Thanks for this heads-up. But I don't see where anything would run in a thread different from the main thread without explicitly telling the reactor to do so. Where do we leave the main thread in the first place? And if so, how do you suggest to fix this?

[LiyiW]

You are right, only by telling the reactor to do run in another thread can that happen.
I provide an example.
twisted_test_consumer.txt
twisted_test_producer.txt

Just change the suffix of the files to be .py and run them.
As you will see, the consumer logs "Sending ACK frame..." in a thread different from main thread.

In my message handler "consume", I tell the reactor to run "callback_func" in another thread because somehow it runs for a while and I hope multiple messages be processed in parallel. As a result, the rest of the "onMessage" function(stompest/src/async/stompest/async/listener.py line 207) will also run in that thread due to the mechanism of Deferred and reactor.

As we can't let ack run in main thread again, I suggest a lock be acquired at before telling twisted transport to write, just use lock in the StompProtocol object

in init:
self._lock = threading.RLock()

and do the following
with self._lock:
self.transport.write(binaryType(frame))
(stompest/src/async/stompest/async/protocol.py line 48)

I think this resolves the problem

[nikipore]

Synchronizing the access to the transport level is exactly what you shouldn't do in an asynchronous (sic!) setting, so your suggested solution is invalid.

Looking at your message handler

@defer.inlineCallbacks
def consume(self, client, frame):
    d = defer.Deferred()
    d.addCallback(longRunningCalculation)
    reactor.callInThread(d.callback, "result 1")
    yield d

You are doing something very wicked, namely calling a callback from another thread, so you yield from one (the main) thread and send from another, which messes up the coroutine and imho breaks the reactor (which must remain single-threaded) altogether. If you do it this way (untested), it should work:

from twisted.internet import threads
[...]
@defer.inlineCallbacks
def consume(self, client, frame):
    result = yield threads.deferToThread(longRunningCalculation)
    logging.debug("In the main thread: %s", result)