Engine keeps running after stop() completes

[Oribow]

Describe the bug
Calling stopDispatcherThreads() seems to act weird. It looks like it's meant to wait for all threads to orderly stop before returning, but it doesn't.

Stop method:

public void stopDispatcherThreads() {
        // dispatchersToShutdown is backed by the map itself so changes in one are reflected in the other
        final Collection<MessageDispatchingThread> dispatchersToShutdown = dispatchers.values();
        for (final MessageDispatchingThread dispatcher : dispatchersToShutdown) {
            dispatcher.stopDispatcher();
        }
        // wait for threads to stop
        while (!dispatchersToShutdown.isEmpty()) {
            try {
                Thread.sleep(100);
            } catch (final InterruptedException e) {
                Thread.currentThread().interrupt();
            }

            dispatchersToShutdown.removeIf(MessageDispatchingThread::isStopped);
        }
    }

Looking at

dispatcher.stopDispatcher();

it is clear that the dispatcher immediately marks itself as stopped and returns.

public void stopDispatcher() {
            enqueue(END_OF_STREAM);
            stopping = true;
            stopped = true;
        }

So the whole 'wait for threads to stop' loop makes no sense, because 'MessageDispatchingThread::isStopped' returns true immediately as it just reads the 'stopped' property, even though the thread might actually still be running.

It seems to me like someone added 'stopped = true;' to the 'stopDispatcher()' function, even though that was never originally intended. It skips a lot of stopping logic still present elsewhere in the code. Even setting 'stopping = true;' is questionable since it should be a consequence of the 'run()' loop reading the END_OF_STREAM message and not be manually set?

void doRun() {
            while (!stopping) {
                try {
                    final Message message = getNextMessage(queueTracker);
                    if (message == null) {
                        // no message available in polling interval
                        continue;
                    }
                    quickfixSession.next(message);
                    **if (message == END_OF_STREAM) {
                        stopping = true;**
                    }

Also, from the way it looks the dummy END_OF_STREAM message is send to the 'quickfixSession.next()'. Is there a reason for that? Seems odd.

To Reproduce

• Have messages in the queue. Easiest way is to slow down processing of messages in the application with a delay.
• Call stop() on the acceptor/initiator.
• after stop() returns the engine will still be processing new messages, thus making stop() non-blocking

Expected behavior
After stop() completes, the engine should do nothing anymore. Or alternatively make it clear that stop() is non-blocking and clean up the unnecessary wait loop.

[chrjohn]

Hi @Oribow , thanks for the detailed report. The code in question (stopped=true) is very old, I'll have to check.

The END_OF_STREAM message is sent to the session to signal that there are no more messages. Formerly, the message processing stopped directly when the connection dropped, without processing in-flight messages.

[Oribow]

Hello!
Thanks for the continued maintenance of this project! It's very robustly implemented aside from this.

Just to elaborate on why that's important to us. We wanted to shutdown down the engine and spin up another engine on another node. Both engines shouldn't be active at the same time so that there is clean hand-off of the ownership of the shared message table. I thought stop() would only return once the engine is fully done processing messages, but that wasn't the case.