Page History

Change Details

The change summary page only covers the idea that extracting the Dispatcher class from the AMQSession class is a goal. The page does not adequately explain how this will address the ordering issue highlighted in QPID-1871. The chief concerns here are to minimise complexity and increase readability/understandability of the client code base.

The goal of this page is to identify the changes that will be required and to show that the change will address the current rollback issue whilst addressing the chief concerns above. In addition the page should detail how the change can be safely executed, given the know issues with locking in the client.

New Dispatcher Interface

The extraction of the Dispatcher from AMQSession will improve readability as we will not have such a large monolithic class file that is currently AMQSession.

/** 
 * Dispatcher is responsible for delivering messages from the IO layer to registered consumers.
 * The dispatcher provides asynchronous dispatch via Consumer MessageListeners and
 * synchronous delivery by placing incoming messages in a Consumers recieveQueue.
 */
public interface DispatcherInterface<C extends BasicMessageConsumer>
{

Java Client Dispatcher Changes.

Investigation of QPID-1871 has highlighted a race condition between the Dispatcher and the clients request to rollback.

Problem Summary

The problem here is that the Dispatcher has the ability to hold on to a message so when the rollback
process is believed to have completed the Dispatcher then rejects the final message AFTER the TxRollback
so that one message gets sent ahead of the other messages. The reject is dropped as the message has been
resent. This is specific to the Java Client causing the Java Broker to return messages out of order. This may be the reason that the RollbackOrderTest has been disabled. It is not clear currently if this will also affect the CPP broker. Further investigation is in required.

Operation Details

Due to the way that the AMQSession.Dispatcher is paused when a rollback operation is in progress it is possible that the Dispatcher thread is 'holding' a message for dispatch. The main loop of AMQSession.Dispatcher is shown here:

  while (!_closed.get() && ((disp = (Dispatchable) _queue.take()) != null))
  {
      disp.dispatch(AMQSession.this);
  }

The problem is highlighted in the dispatchMessage call below (which is the result of disp.dispatch() on an UnprocessedMessage). If the Dispatcher is in the process of dispatching messages when a second thread calls rollback then the connection will be stopped and the dispatcher can remove a message from _queue and then stop in the dispatchMessage

 private void dispatchMessage(UnprocessedMessage message)
        {
     /** Start this Dispatcher if required */
       long deliveryTagpublic =void message.getDeliveryTag(startDispatcherIfNecessary(boolean initiallyStopped);

        /** Close this Dispatcher synchronized (_lock)
       and release any resources held. Cannot be re-opened */
     {
   public void close();
         /** Register try
a consumer to receive messages from this Dispatcher. */
        {
public void registerConsumer(C consumer);
        /** Unregister a consumer from this Dispatcher @return C while (connectionStopped())removed consumer */
        public C unregisterConsumer(C consumer);
         {
         /** Rollback the received message state in this Dispatcher */
        public       _lock.waitvoid rollback();
        /** Process the received message from the IO     }Layer */
        public void       }
                catch (InterruptedException e)
       dispatchMessage(UnprocessedMessage message);
}

Message Dispatch

The extraction of this class will wrap the current _queue object that receives incoming messages from the the IO Layer. As a result a new interface method will be required for the AMQSession to add the incoming messages to _queue. This is the only interaction that AMQSession has with _queue so this is a simple refactor.

        /** {
Process the received message from the IO Layer */
        public void dispatchMessage(UnprocessedMessage message);

Ownership of Consumer list

The list of consumers in _consumers is used by both AMQSession and AMQSession.Dispatcher. Here there are two approaches presented for discussion as a clean refactoring is not possible. The introduction of _removedConsumers, which does not behave as the local comments describe, gives two approaches to the refactor. Whilst the need for _removedConsumers is unclear its removal is not being considered here, the focus is to address the current rollback issues.

Dispatcher maintains own copy

Additionally methods will be required to update the _consumers list used for message dispatching. The _consumer list is used for more than just dispatching. AMQSession also uses the list of consumers during failover for resubscription. At this time refactoring how failover operates is not prudent.
So it is proposed that the Dispatcher maintains its own list of consumers which is updated via this interface.

   // pass
                }

                if (!(message instanceof CloseConsumerMessage)
                    && tagLE(deliveryTag, _rollbackMark.get()))
                {
                    rejectMessage(message, true);
                }
                else
                {
               /** Register a consumer to synchronized (_messageDeliveryLock)
    receive messages from this Dispatcher. */
        public void registerConsumer(C consumer);
     {
   /** Unregister a consumer from this Dispatcher @return C removed consumer */
        public C notifyConsumerunregisterConsumer(message);
                    }
                }
            }

            long current = _rollbackMark.get();
            if (updateRollbackMark(current, deliveryTag))
    C consumer);

This interface can easily be called from AMQSession as there is a single put call made from the consumeFromQueue() method. Removal is performed during deregisterConsumers() and a call to keep the dispatcher in sync not unreasonable. This method does highlight an additional data structure that is updated by AMQSession but only used by the Dispatcher: _removedConsumers. The list of removed consumers can be maintained through the unregisterConsumer() method this simplifying AMQSession.
The remaining point of shared use of _consumers is in resubscribeConsumers() and is called as part of failover. The _consumer list is cleared and all consumers re-registered.

        {
/** Clear all consumers from this Dispatcher as used by failover @return List<C> removed   _rollbackMark.compareAndSet(current, deliveryTag);consumers */
        public    }
        }

When the connection is resumed the deliveryTag of the current message will be 'less than or equal' to the _rollbackMark as this has been set to the highest deliveryTag received prior to rollback.

List<C> clearAllConsumer();

This approach is allows the resubscribeConsumers() method to clear all the consumers in the Dispatcher ahead of it re-registering. We only need to perform this clear because the client does not reuse the consumerTag when it re-registers the client. New tags are used in the registration with the broker.

AMQSession maintains canonical list

An alternative would be to define an internal interface between the Dispatcher and AMQSession.

        /** Get the current consumer for the given _rollbackMark.set(_highestDeliveryTag.get());

There are no guards in the code to stop the IO layer adding a new message to _queue whilst rollback is in progress. However, both 0-8 and 0-10 ensure that message flow has stopped whilst recovery is processed. The 0-8 sets ChannelFlow=false and waits for the Ok, in 0-10 the consumers are stopped and a sync performed.

Code Problem

The investigation of this problem has highlighted a two areas which need to be addressed:

1. The ability to ensure the dispatcher is not holding a message.
2. The ability to confirm when the dispatcher will not process any more messages.

How the Dispatcher holds a message

The _queue.take() call is guaranteed never to return null and once we have entered the take() call there is no way to stop the Dispatcher.

  while (!_closed.get() && ((disp = (Dispatchable) _queue.take()) != null))

Hence we perform the stop as soon as possible after the take(), but this results in us holding on to a message.

Ideally we need to be able to stop the Dispatcher whilst it is in the take() method.

How the Dispatcher can keep processing.

The Dispatcher is currently uses the connecitonStopped() call to suspend its activities when the connection has been marked as stopped. However, we need to know that the Dispatcher has actually hit this section otherwise we need to guarantee that the _queue is empty.

consumerTag. */
        public C getConsumerForTag(int consumerTag);
        /** Get all the consumers for this Session. */ 
        public Collection<C> getAllConsumersRegistered();

This would allow AMQSession to maintain the list of consumers but the list of _removedConsumers would need to be made available to the Dispatcher class. I do not intend to change how this is used as the comments and the usage of this data structure do not align. In deregisterConsumer() it states:

            // Consumers that are closed in a transaction must be stored
            // so that messages they have received can be acknowledged on commit
            if (_transacted)
   

            synchronized (_lock){
            {
    _removedConsumers.add(consumer);
            }

However, the only usage of _removedConsumers is in the Dispatcher rollback(), which does not send acknowledgements, yet the comments in both sections of code state that is the use of this structure.

try
                {
for (int i = 0; i < _removedConsumers.size(); i++)
             while (connectionStopped())   {
                    {
// Sends acknowledgement to server
                    _lock.waitremovedConsumers.get(i).rollback();
                    }_removedConsumers.remove(i);
                }
                catch (InterruptedException e)
                {
                    // pass

Looking at it it appears that this is used to ensure that any prefetched messages are rolled back. However, there will only be messages to rollback if the consumer was closed by an error. I believe that the correct course of action is actually to ensure that when all consumers are closed/become invalid/are stopped that any prefetched messages are correctly released. However, this is an additional change out of the scope of this refactoring. If it is desired that AMQSession alone should maintain a list of consumers then the Dispatcher interface could simply be modified to maintain the functionality of _removedConsumers.

        /** Record a consumer that has been removed in Dispatcher  }

Having the Dispatcher signal that it has stopped processing will allow us to know that we have hit the stopped state. However, this will mean that we have the opportunity to process one extra message AFTER the rollback command has been requested.

Proposed Solution

Currently there is a lot of synchronisation to ensure that we can safely start the rollback process in the AMQSession before asking the Dispatcher to clean up its resources. To ensure signal that we have stopped the dispatcher and so can guarantee we are no longer holding a message will require more synchronisation, which is both error prone and will and additional complexity to the client.

The proposed alternative is to modify the FlowControllingBlockingQueue so that we can delegate all rollback processing to the Dispatcher. This removes the need to stop the Dispatcher and if the Dispatcher is performing the rollback then it can be sure it is not currently processing an UnprocessedMessage.

While delegating the rollback of consumed messages to the Dispatcher it makes sense to give the Dispatcher more formal control over the receipt and dispatching of incoming messages. By extracting the Dispatcher from the Session class we can simplify the both classes. Locking will be reduced and the responsibility of message processing will be more cleanly delegated to the Dispatcher.

/** 
 * Dispatcher is responsible for delivering messages from the IO layer to registered consumers.
 * The dispatcher provides asynchronous dispatch via Consumer MessageListeners and
 * synchronous delivery by placing incoming messages in a Consumers recieveQueue.
 */
public interface DispatcherInterface<C extends BasicMessageConsumer>
{
        /** Start this Dispatcher if required */
        public void startDispatcherIfNecessary(boolean initiallyStopped);
        /** Close this Dispatcher and release any resources held. Cannot be re-opened */
        public void close();
        /** Register a consumer to receive messages from this Dispatcher. */
        public void registerConsumer(C consumer);
        /** Unregister a consumer from this Dispatcher @return C removed consumer */
        public C unregisterConsumer(C consumer);
        /** Rollback the received message state in this Dispatcher */
        public void rollback();
        /** Process the received message from the IO Layer */
        public void dispatchMessage(UnprocessedMessage message);
}

The clean interface with the Dispatcher from the Session means that we can more clearly delegate the rollback() control to the Dispatcher. The FlowControllingBlockingQueue will need to be augmented so that when an asynchronous request for rollback is made the Dispatcher can then pick up on this 'ServiceRequest'.

Steps

1. Extract Dispatcher code to separate class and validate interface
2. Update AMQSession to use new Dispatcher interface.
3. Augment FlowControllingBlockingQueue to allow the injection of 'ServiceRequests'

Comment Responses

...

User

...

Comment

...

via

...

Response

...

rhs

...

AMQSession.syncDispatchQueue is used in 0-10 for this

...

email

...

This will not work if the dispatcher is performing the rollback (Deadlock).
Also we need to stop processing the messages immediately and not allow any further processing.

...

rhs

...

Agree the client is badly in need of some improvements in maintainability and readability, however in this particular case I don't think moving the rollback processing from one thread to another actually improves the situation significantly.

...

email

...

It is not so much moving from on thread to another but from moving from the AMQSession / Dispatcher objects to just the Dispatcher.

...

rhs

...

I suspect in order do this properly we really need to stop thinking in terms of code being associated with a given thread, and think instead about what locks we have, what data structures those locks protect, and which locks need to be held in order to execute a given piece of code.

...

email

...

The focus of this change was to consolodate the operations on the received messagse. I would like to see a clean interface where messages are passed in for for dispatching. The cleaning operations should then be full contained in that interface not in a couple of locations as it is currently.

for rollback purposes. */
        public void recordRemovedConsumer(C consumer);

Rollback Changes

The Dispatcher Interface calls for a rollback() method and this is where part of the proposed changes will occur.

        /** Rollback the received message state in this Dispatcher */
        public void rollback();

The current code in AMQSession rollback() will be moved to the new Dispatcher class. The existing locking, transactional checks, channel suspension and clearing of session state will be left for AMQSession to handle.

                releaseForRollback();

                sendRollback();

The extraction of releaseForRollback() is where the protocol dependent components will initially be presented. The initial refactoring will maintain the protocol dependent differences however, after the work is completed it is expected that the only difference will be that 0-10 allows for release of a range of deliveryTags while prior to that individual rejects must be sent.

Current Dispatcher Rollback pseudo code

Currently the act of calling session.rollback() performs the following tasks:

# In AMQSession.rollback
 Whilst under the _suspensionLock
  - Check that we are transacted
  - Suspend the session, stopping all new message delivery from broker
  - Perform releaseForRollback()
   # 0-8 
    - reject all Delivered Messages
    - Call dispatcher.rollback()
   # 0-10
    - startDispatcherIfNecessary
    - Place a message on the IO to Dispatcher _queue
    - wait for this message to be processed by the dispatcher thread
    - Call dispatcher.rollback()
    - release and then clear all recorded message tags in _txRangeSet

# In AMQSession.Dispatcher.rollback()
 Whilst under Dispatcher._lock
  - Set the connection Stopped(Stop Dispatcher Thread running)
  - For all Consumers, perform rollback
  - For all browsers, clear prefetch 
  - For all removed consumers perform rollback
  - Set the connection stop/start state back to the value before we started

During this time the Dispatcher thread will be dong the following

# If Disptacher is already running 0-8 & 0-10
   - block for next message on _queue 
   - remove message from _queue and start to dispatch
   # For Dispatchable = UnprocessedMessages
    Whilst under Dispatcher._lock
     - wait until the connection is not stopped
     - If the message is not a CloseConsumerMessage and the deliveryTag is <= current _rollbackMark 
       - reject message
     - Otherwise 
       Whilst under AMQSession._messageDeliveryLock
         - notifyConsumer of the message        
     
   # For Dispatchable = syncDispatchQueue
    - signal queue processed to this point
  
# if Dispatcher is not running 0-8
  Dispatcher is not started, so _queue is not processed. However, Dispatcher is always started if we
  have the possibility of receiving messages so there is no potential to have messages stuck on _queue.

# if Dispatcher is not running 0-10
  Not sure this is possible as the logic should be the same as for 0-8, however as there is an explicit
  startDispatcherIfNecessary call there is the assumption that it may not be running so lets track what 
  happens in that case.
  - Dispatcher thread starts and waits for next message on _queue
  - remove message (which will be a syncDispatchQueue Dispatchable) from _queue and start to dispatch
  - Signal that we have processed it.
  - block on _queue.take() as _queue will be empty.

The current problem is in the Dispatcher thread, when it is already running when the rollback is called and the _queue has messages for dispatch. The Dispatcher Thread can remove a message and then become blocked whilst the connection is stopped. Then after the other thread has performed the rollback the Dispatcher Thread then proceeds to reject the message it has.

Proposed Dispatcher Rollback pseudo code

# In AMQSession.rollback
 Whilst under the _suspensionLock
  - Check that we are transacted
  - Suspend the session
  - Call dispatcher.rollback()
   # 0-10
    - release and then clear all recorded message tags in _txRangeSet

# In AMQSession.Dispatcher.rollback()
 - If we are the Dispatcher Thread
   - Perform actual rollback()
 - otherwise
   - Request rollback be completed by placing service message in FlowControlBlockingQueue
   - wait for completion

# If Disptacher is already running 0-8 & 0-10
   - block for next message on _queue 
   - remove message from _queue and start to dispatch
   # For Dispatchable = UnprocessedMessages
    Whilst under Dispatcher._lock
     - wait until the connection is not stopped
     - If the message is not a CloseConsumerMessage and the deliveryTag is <= current _rollbackMark 
       - reject message
     - Otherwise 
       Whilst under AMQSession._messageDeliveryLock
         - notifyConsumer of the message        
     
   # For Dispatchable = syncDispatchQueue
    - signal queue processed to this point

   # For Dispatchable = RollbackService Message
    Whilst under the _lock
     - Set the connection Stopped(Stop Dispatcher Thread running)
     # For 0-8
       - Release/Reject all delivered messages  
     - For all Consumers, perform rollback
     - For all browsers, clear prefetch 
     - For all removed consumers perform rollback
     - Release/Reject all messages still in _queue
     - Set the connection stop/start state back to the value before we started

The above change to the Dispatcher is dependant on a change to the FlowControllingBlockingQueue to allow service requests to be injected. This is how this is proposed

    public Object take() throws InterruptedException
    {
        Object o = _queue.poll();
        if(o == null)
        {
            synchronized(this)
            {
                while((o = _queue.poll())==null)
                {
                    wait();
                }
            }
        }
     ...

    public void addServiceRequest(ServiceRequest o)
    {
        synchronized(this)
        {
            _serviceQueue.add(o);

            notifyAll();
        }
    }    

    public Object take() throws InterruptedException
    {
        Object o = _queue.poll();
        if(o == null)
        {
            synchronized(this)
            {
                while(((o = _serviceQueue.poll())==null) && 
                      ((o = _queue.poll())==null))
                {
                    wait();
                }
            }
        }
        // Return early so we do not upset the FlowControlCounts.
        if (o instanceof ServiceRequest)
        {
            return o;
        }
        
     ...

...

rhs

...

Really we need to be able to articulate exactly what locks the client has, what data structure(s) each lock protects, and what order should be used to acquire multiple locks when necessary.

...

email

...