broker-core/src/main/java/org/apache/qpid/server/transfer/TransferQueueImpl.java - qpid-broker-j - Git at Google

 /*
  *
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  *
  */
 package org.apache.qpid.server.transfer;

 import java.nio.charset.StandardCharsets;
 import java.security.AccessControlException;
 import java.util.Collection;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.atomic.AtomicInteger;

 import org.apache.qpid.server.message.BaseMessageInstance;
 import org.apache.qpid.server.message.InstanceProperties;
 import org.apache.qpid.server.message.MessageReference;
 import org.apache.qpid.server.message.ServerMessage;
 import org.apache.qpid.server.model.NamedAddressSpace;
 import org.apache.qpid.server.model.State;
 import org.apache.qpid.server.model.VirtualHost;
 import org.apache.qpid.server.queue.QueueEntry;
 import org.apache.qpid.server.security.SecurityToken;
 import org.apache.qpid.server.store.MessageDurability;
 import org.apache.qpid.server.store.MessageEnqueueRecord;
 import org.apache.qpid.server.store.StorableMessageMetaData;
 import org.apache.qpid.server.txn.AutoCommitTransaction;
 import org.apache.qpid.server.txn.ServerTransaction;
 import org.apache.qpid.server.util.Action;
 import org.apache.qpid.server.virtualhost.VirtualHostUnavailableException;

 public class TransferQueueImpl implements TransferQueue
 {
     private static final UUID TRANSFER_QUEUE_ID = UUID.nameUUIDFromBytes("$transfer".getBytes(StandardCharsets.UTF_8));

     private static final int RECOVERING = 1;
     private static final int COMPLETING_RECOVERY = 2;
     private static final int RECOVERED = 3;

     private final AtomicInteger _recovering = new AtomicInteger(RECOVERING);
     private final AtomicInteger _enqueuingWhileRecovering = new AtomicInteger(0);

     private final ConcurrentLinkedQueue<EnqueueRequest> _postRecoveryQueue = new ConcurrentLinkedQueue<>();

     private final VirtualHost<?> _virtualHost;

     private final TransferQueueEntryList _queueEntryList;
     private Collection<TransferQueueConsumer> _consumers = new CopyOnWriteArrayList<>();


     public TransferQueueImpl(final VirtualHost<?> virtualHost)
     {
         _virtualHost = virtualHost;
         _queueEntryList = new TransferQueueEntryList(this);
     }

     @Override
     public void enqueue(final ServerMessage message,
                         final Action<? super BaseMessageInstance> action,
                         final MessageEnqueueRecord enqueueRecord)
     {
         if(_recovering.get() != RECOVERED)
         {
             _enqueuingWhileRecovering.incrementAndGet();

             boolean addedToRecoveryQueue;
             try
             {
                 if(addedToRecoveryQueue = (_recovering.get() == RECOVERING))
                 {
                     _postRecoveryQueue.add(new EnqueueRequest(message, action, enqueueRecord));
                 }
             }
             finally
             {
                 _enqueuingWhileRecovering.decrementAndGet();
             }

             if(!addedToRecoveryQueue)
             {
                 while(_recovering.get() != RECOVERED)
                 {
                     Thread.yield();
                 }
                 doEnqueue(message, action, enqueueRecord);
             }
         }
         else
         {
             doEnqueue(message, action, enqueueRecord);
         }


     }

     @Override
     public void recover(final ServerMessage<?> message, final MessageEnqueueRecord enqueueRecord)
     {
         doEnqueue(message, null, enqueueRecord);
     }

     @Override
     public final void completeRecovery()
     {
         if(_recovering.compareAndSet(RECOVERING, COMPLETING_RECOVERY))
         {
             while(_enqueuingWhileRecovering.get() != 0)
             {
                 Thread.yield();
             }

             // at this point we can assert that any new enqueue to the queue will not try to put into the post recovery
             // queue (because the state is no longer RECOVERING, but also no threads are currently trying to enqueue
             // because the _enqueuingWhileRecovering count is 0.

             enqueueFromPostRecoveryQueue();

             _recovering.set(RECOVERED);

         }
     }

     @Override
     public VirtualHost<?> getVirtualHost()
     {
         return _virtualHost;
     }

     private void enqueueFromPostRecoveryQueue()
     {
         while(!_postRecoveryQueue.isEmpty())
         {
             EnqueueRequest request = _postRecoveryQueue.poll();
             MessageReference<?> messageReference = request.getMessage();
             doEnqueue(messageReference.getMessage(), request.getAction(), request.getEnqueueRecord());
             messageReference.release();
         }
     }


     protected void doEnqueue(final ServerMessage message, final Action<? super BaseMessageInstance> action, MessageEnqueueRecord enqueueRecord)
     {
         final TransferQueueEntry entry = (TransferQueueEntry) _queueEntryList.add(message, enqueueRecord);
         for (TransferQueueConsumer consumer : getConsumers())
         {
             if (consumer.hasInterest(entry))
             {
                 consumer.notifyWork();
             }
         }
     }

     private Collection<TransferQueueConsumer> getConsumers()
     {
         return _consumers;
     }

     @Override
     public TransferQueueConsumer addConsumer(final TransferTarget target,
                                              final String consumerName)
     {


         TransferQueueConsumer consumer = new TransferQueueConsumer(this,
                                                                    target,
                                                                    consumerName);

         QueueContext queueContext = new QueueContext(_queueEntryList.getHead());
         consumer.setQueueContext(queueContext);
         _consumers.add(consumer);
         consumer.notifyWork();

         return consumer;
     }

     boolean hasAvailableMessages(final TransferQueueConsumer queueConsumer)
     {
         return getNextAvailableEntry(queueConsumer) != null;
     }


     void setLastSeenEntry(final TransferQueueConsumer sub, final TransferQueueEntry entry)
     {
         QueueContext subContext = sub.getQueueContext();
         if (subContext != null)
         {
             QueueEntry releasedEntry = subContext.getReleasedEntry();

             QueueContext._lastSeenUpdater.set(subContext, entry);
             if(releasedEntry == entry)
             {
                 QueueContext._releasedUpdater.compareAndSet(subContext, releasedEntry, null);
             }
         }
     }

     TransferQueueEntry getNextAvailableEntry(final TransferQueueConsumer sub)
     {
         QueueContext context = sub.getQueueContext();
         if(context != null)
         {
             QueueEntry lastSeen = context.getLastSeenEntry();
             QueueEntry releasedNode = context.getReleasedEntry();

             TransferQueueEntry node =
                     (TransferQueueEntry) ((releasedNode != null && lastSeen.compareTo(releasedNode) >= 0)
                                         ? releasedNode : _queueEntryList.next(lastSeen));

             boolean expired = false;
             while (node != null
                    && (!node.isAvailable()
                        || (expired = node.expired())
                        || !sub.hasInterest(node)))
             {
                 if (expired)
                 {
                     expired = false;
                     if (node.acquire())
                     {
                         dequeueEntry(node);
                     }
                 }

                 if(QueueContext._lastSeenUpdater.compareAndSet(context, lastSeen, node))
                 {
                     QueueContext._releasedUpdater.compareAndSet(context, releasedNode, null);
                 }

                 lastSeen = context.getLastSeenEntry();
                 releasedNode = context.getReleasedEntry();
                 node = (TransferQueueEntry) ((releasedNode != null && lastSeen.compareTo(releasedNode) >= 0)
                                         ? releasedNode
                                         : _queueEntryList.next(lastSeen));
             }
             return node;
         }
         else
         {
             return null;
         }
     }

     private void dequeueEntry(final QueueEntry node)
     {
         ServerTransaction txn = new AutoCommitTransaction(_virtualHost.getMessageStore());
         dequeueEntry(node, txn);
     }

     private void dequeueEntry(final QueueEntry node, ServerTransaction txn)
     {
         txn.dequeue(node.getEnqueueRecord(),
                     new ServerTransaction.Action()
                     {

                         public void postCommit()
                         {
                             node.delete();
                         }

                         public void onRollback()
                         {

                         }
                     });
     }

     @Override
     public boolean isDurable()
     {
         return true;
     }

     @Override
     public boolean isDeleted()
     {
         return false;
     }

     @Override
     public String getName()
     {
         return "$transfer";
     }

     @Override
     public NamedAddressSpace getAddressSpace()
     {
         return _virtualHost;
     }

     @Override
     public void authorisePublish(final SecurityToken token, final Map<String, Object> arguments)
             throws AccessControlException
     {
         // TODO
     }

     @Override
     public <M extends ServerMessage<? extends StorableMessageMetaData>> int send(final M message,
                                                                                  final String routingAddress,
                                                                                  final InstanceProperties instanceProperties,
                                                                                  final ServerTransaction txn,
                                                                                  final Action<? super BaseMessageInstance> postEnqueueAction)
     {
         if (_virtualHost.getState() != State.ACTIVE)
         {
             throw new VirtualHostUnavailableException(this._virtualHost);
         }

         if(!message.isReferenced(this))
         {
             txn.enqueue(this, message, new ServerTransaction.EnqueueAction()
             {
                 MessageReference _reference = message.newReference();

                 public void postCommit(MessageEnqueueRecord... records)
                 {
                     try
                     {
                         TransferQueueImpl.this.enqueue(message, postEnqueueAction, records[0]);
                     }
                     finally
                     {
                         _reference.release();
                     }
                 }

                 public void onRollback()
                 {
                     _reference.release();
                 }
             });
             return 1;
         }
         else
         {
             return 0;
         }

     }

     @Override
     public UUID getId()
     {
         return TRANSFER_QUEUE_ID;
     }

     @Override
     public MessageDurability getMessageDurability()
     {
         return MessageDurability.DEFAULT;
     }

     public void flushConsumer(final TransferQueueConsumer consumer, final int count)
     {

     }

     private static class EnqueueRequest
     {
         private final MessageReference<?> _message;
         private final Action<? super BaseMessageInstance> _action;
         private final MessageEnqueueRecord _enqueueRecord;

         public EnqueueRequest(final ServerMessage message,
                               final Action<? super BaseMessageInstance> action,
                               final MessageEnqueueRecord enqueueRecord)
         {
             _enqueueRecord = enqueueRecord;
             _message = message.newReference();
             _action = action;
         }

         public MessageReference<?> getMessage()
         {
             return _message;
         }

         public Action<? super BaseMessageInstance> getAction()
         {
             return _action;
         }

         public MessageEnqueueRecord getEnqueueRecord()
         {
             return _enqueueRecord;
         }
     }

 }
	/*
	*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*
	*/
	package org.apache.qpid.server.transfer;

	import java.nio.charset.StandardCharsets;
	import java.security.AccessControlException;
	import java.util.Collection;
	import java.util.Map;
	import java.util.UUID;
	import java.util.concurrent.ConcurrentLinkedQueue;
	import java.util.concurrent.CopyOnWriteArrayList;
	import java.util.concurrent.atomic.AtomicInteger;

	import org.apache.qpid.server.message.BaseMessageInstance;
	import org.apache.qpid.server.message.InstanceProperties;
	import org.apache.qpid.server.message.MessageReference;
	import org.apache.qpid.server.message.ServerMessage;
	import org.apache.qpid.server.model.NamedAddressSpace;
	import org.apache.qpid.server.model.State;
	import org.apache.qpid.server.model.VirtualHost;
	import org.apache.qpid.server.queue.QueueEntry;
	import org.apache.qpid.server.security.SecurityToken;
	import org.apache.qpid.server.store.MessageDurability;
	import org.apache.qpid.server.store.MessageEnqueueRecord;
	import org.apache.qpid.server.store.StorableMessageMetaData;
	import org.apache.qpid.server.txn.AutoCommitTransaction;
	import org.apache.qpid.server.txn.ServerTransaction;
	import org.apache.qpid.server.util.Action;
	import org.apache.qpid.server.virtualhost.VirtualHostUnavailableException;

	public class TransferQueueImpl implements TransferQueue
	{
	private static final UUID TRANSFER_QUEUE_ID = UUID.nameUUIDFromBytes("$transfer".getBytes(StandardCharsets.UTF_8));

	private static final int RECOVERING = 1;
	private static final int COMPLETING_RECOVERY = 2;
	private static final int RECOVERED = 3;

	private final AtomicInteger _recovering = new AtomicInteger(RECOVERING);
	private final AtomicInteger _enqueuingWhileRecovering = new AtomicInteger(0);

	private final ConcurrentLinkedQueue<EnqueueRequest> _postRecoveryQueue = new ConcurrentLinkedQueue<>();

	private final VirtualHost<?> _virtualHost;

	private final TransferQueueEntryList _queueEntryList;
	private Collection<TransferQueueConsumer> _consumers = new CopyOnWriteArrayList<>();


	public TransferQueueImpl(final VirtualHost<?> virtualHost)
	{
	_virtualHost = virtualHost;
	_queueEntryList = new TransferQueueEntryList(this);
	}

	@Override
	public void enqueue(final ServerMessage message,
	final Action<? super BaseMessageInstance> action,
	final MessageEnqueueRecord enqueueRecord)
	{
	if(_recovering.get() != RECOVERED)
	{
	_enqueuingWhileRecovering.incrementAndGet();

	boolean addedToRecoveryQueue;
	try
	{
	if(addedToRecoveryQueue = (_recovering.get() == RECOVERING))
	{
	_postRecoveryQueue.add(new EnqueueRequest(message, action, enqueueRecord));
	}
	}
	finally
	{
	_enqueuingWhileRecovering.decrementAndGet();
	}

	if(!addedToRecoveryQueue)
	{
	while(_recovering.get() != RECOVERED)
	{
	Thread.yield();
	}
	doEnqueue(message, action, enqueueRecord);
	}
	}
	else
	{
	doEnqueue(message, action, enqueueRecord);
	}


	}

	@Override
	public void recover(final ServerMessage<?> message, final MessageEnqueueRecord enqueueRecord)
	{
	doEnqueue(message, null, enqueueRecord);
	}

	@Override
	public final void completeRecovery()
	{
	if(_recovering.compareAndSet(RECOVERING, COMPLETING_RECOVERY))
	{
	while(_enqueuingWhileRecovering.get() != 0)
	{
	Thread.yield();
	}

	// at this point we can assert that any new enqueue to the queue will not try to put into the post recovery
	// queue (because the state is no longer RECOVERING, but also no threads are currently trying to enqueue
	// because the _enqueuingWhileRecovering count is 0.

	enqueueFromPostRecoveryQueue();

	_recovering.set(RECOVERED);

	}
	}

	@Override
	public VirtualHost<?> getVirtualHost()
	{
	return _virtualHost;
	}

	private void enqueueFromPostRecoveryQueue()
	{
	while(!_postRecoveryQueue.isEmpty())
	{
	EnqueueRequest request = _postRecoveryQueue.poll();
	MessageReference<?> messageReference = request.getMessage();
	doEnqueue(messageReference.getMessage(), request.getAction(), request.getEnqueueRecord());
	messageReference.release();
	}
	}



	protected void doEnqueue(final ServerMessage message, final Action<? super BaseMessageInstance> action, MessageEnqueueRecord enqueueRecord)
	{
	final TransferQueueEntry entry = (TransferQueueEntry) _queueEntryList.add(message, enqueueRecord);
	for (TransferQueueConsumer consumer : getConsumers())
	{
	if (consumer.hasInterest(entry))
	{
	consumer.notifyWork();
	}
	}
	}

	private Collection<TransferQueueConsumer> getConsumers()
	{
	return _consumers;
	}

	@Override
	public TransferQueueConsumer addConsumer(final TransferTarget target,
	final String consumerName)
	{



	TransferQueueConsumer consumer = new TransferQueueConsumer(this,
	target,
	consumerName);

	QueueContext queueContext = new QueueContext(_queueEntryList.getHead());
	consumer.setQueueContext(queueContext);
	_consumers.add(consumer);
	consumer.notifyWork();

	return consumer;
	}

	boolean hasAvailableMessages(final TransferQueueConsumer queueConsumer)
	{
	return getNextAvailableEntry(queueConsumer) != null;
	}


	void setLastSeenEntry(final TransferQueueConsumer sub, final TransferQueueEntry entry)
	{
	QueueContext subContext = sub.getQueueContext();
	if (subContext != null)
	{
	QueueEntry releasedEntry = subContext.getReleasedEntry();

	QueueContext._lastSeenUpdater.set(subContext, entry);
	if(releasedEntry == entry)
	{
	QueueContext._releasedUpdater.compareAndSet(subContext, releasedEntry, null);
	}
	}
	}

	TransferQueueEntry getNextAvailableEntry(final TransferQueueConsumer sub)
	{
	QueueContext context = sub.getQueueContext();
	if(context != null)
	{
	QueueEntry lastSeen = context.getLastSeenEntry();
	QueueEntry releasedNode = context.getReleasedEntry();

	TransferQueueEntry node =
	(TransferQueueEntry) ((releasedNode != null && lastSeen.compareTo(releasedNode) >= 0)
	? releasedNode : _queueEntryList.next(lastSeen));

	boolean expired = false;
	while (node != null
	&& (!node.isAvailable()
	\|\| (expired = node.expired())
	\|\| !sub.hasInterest(node)))
	{
	if (expired)
	{
	expired = false;
	if (node.acquire())
	{
	dequeueEntry(node);
	}
	}

	if(QueueContext._lastSeenUpdater.compareAndSet(context, lastSeen, node))
	{
	QueueContext._releasedUpdater.compareAndSet(context, releasedNode, null);
	}

	lastSeen = context.getLastSeenEntry();
	releasedNode = context.getReleasedEntry();
	node = (TransferQueueEntry) ((releasedNode != null && lastSeen.compareTo(releasedNode) >= 0)
	? releasedNode
	: _queueEntryList.next(lastSeen));
	}
	return node;
	}
	else
	{
	return null;
	}
	}

	private void dequeueEntry(final QueueEntry node)
	{
	ServerTransaction txn = new AutoCommitTransaction(_virtualHost.getMessageStore());
	dequeueEntry(node, txn);
	}

	private void dequeueEntry(final QueueEntry node, ServerTransaction txn)
	{
	txn.dequeue(node.getEnqueueRecord(),
	new ServerTransaction.Action()
	{

	public void postCommit()
	{
	node.delete();
	}

	public void onRollback()
	{

	}
	});
	}

	@Override
	public boolean isDurable()
	{
	return true;
	}

	@Override
	public boolean isDeleted()
	{
	return false;
	}

	@Override
	public String getName()
	{
	return "$transfer";
	}

	@Override
	public NamedAddressSpace getAddressSpace()
	{
	return _virtualHost;
	}

	@Override
	public void authorisePublish(final SecurityToken token, final Map<String, Object> arguments)
	throws AccessControlException
	{
	// TODO
	}

	@Override
	public <M extends ServerMessage<? extends StorableMessageMetaData>> int send(final M message,
	final String routingAddress,
	final InstanceProperties instanceProperties,
	final ServerTransaction txn,
	final Action<? super BaseMessageInstance> postEnqueueAction)
	{
	if (_virtualHost.getState() != State.ACTIVE)
	{
	throw new VirtualHostUnavailableException(this._virtualHost);
	}

	if(!message.isReferenced(this))
	{
	txn.enqueue(this, message, new ServerTransaction.EnqueueAction()
	{
	MessageReference _reference = message.newReference();

	public void postCommit(MessageEnqueueRecord... records)
	{
	try
	{
	TransferQueueImpl.this.enqueue(message, postEnqueueAction, records[0]);
	}
	finally
	{
	_reference.release();
	}
	}

	public void onRollback()
	{
	_reference.release();
	}
	});
	return 1;
	}
	else
	{
	return 0;
	}

	}

	@Override
	public UUID getId()
	{
	return TRANSFER_QUEUE_ID;
	}

	@Override
	public MessageDurability getMessageDurability()
	{
	return MessageDurability.DEFAULT;
	}

	public void flushConsumer(final TransferQueueConsumer consumer, final int count)
	{

	}

	private static class EnqueueRequest
	{
	private final MessageReference<?> _message;
	private final Action<? super BaseMessageInstance> _action;
	private final MessageEnqueueRecord _enqueueRecord;

	public EnqueueRequest(final ServerMessage message,
	final Action<? super BaseMessageInstance> action,
	final MessageEnqueueRecord enqueueRecord)
	{
	_enqueueRecord = enqueueRecord;
	_message = message.newReference();
	_action = action;
	}

	public MessageReference<?> getMessage()
	{
	return _message;
	}

	public Action<? super BaseMessageInstance> getAction()
	{
	return _action;
	}

	public MessageEnqueueRecord getEnqueueRecord()
	{
	return _enqueueRecord;
	}
	}

	}