pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherMultipleConsumers.java - pulsar - 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.pulsar.broker.service;

 import com.carrotsearch.hppc.ObjectHashSet;
 import com.carrotsearch.hppc.ObjectSet;
 import java.util.Random;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import org.apache.pulsar.broker.service.persistent.PersistentStickyKeyDispatcherMultipleConsumers;
 import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 /**
  *
  */
 public abstract class AbstractDispatcherMultipleConsumers extends AbstractBaseDispatcher {

     protected final CopyOnWriteArrayList<Consumer> consumerList = new CopyOnWriteArrayList<>();
     protected final ObjectSet<Consumer> consumerSet = new ObjectHashSet<>();
     protected volatile int currentConsumerRoundRobinIndex = 0;

     protected static final int FALSE = 0;
     protected static final int TRUE = 1;
     protected static final AtomicIntegerFieldUpdater<AbstractDispatcherMultipleConsumers> IS_CLOSED_UPDATER =
             AtomicIntegerFieldUpdater
                     .newUpdater(AbstractDispatcherMultipleConsumers.class, "isClosed");
     private volatile int isClosed = FALSE;

     private Random random = new Random(42);

     protected AbstractDispatcherMultipleConsumers(Subscription subscription) {
         super(subscription);
     }

     public boolean isConsumerConnected() {
         return !consumerList.isEmpty();
     }

     public CopyOnWriteArrayList<Consumer> getConsumers() {
         return consumerList;
     }

     public synchronized boolean canUnsubscribe(Consumer consumer) {
         return consumerList.size() == 1 && consumerSet.contains(consumer);
     }

     public boolean isClosed() {
         return isClosed == TRUE;
     }

     public SubType getType() {
         return SubType.Shared;
     }

     public abstract boolean isConsumerAvailable(Consumer consumer);

     protected void cancelPendingRead() {}

     /**
      * <pre>
      * Broker gives more priority while dispatching messages. Here, broker follows descending priorities. (eg:
      * 0=max-priority, 1, 2,..)
      * <p>
      * Broker will first dispatch messages to max priority-level consumers if they
      * have permits, else broker will consider next priority level consumers.
      * Also on the same priority-level, it selects consumer in round-robin manner.
      * <p>
      * If subscription has consumer-A with  priorityLevel 1 and Consumer-B with priorityLevel 2
      * then broker will dispatch
      * messages to only consumer-A until it runs out permit and then broker starts dispatching messages to Consumer-B.
      * <p>
      * Consumer PriorityLevel Permits
      * C1       0             2
      * C2       0             1
      * C3       0             1
      * C4       1             2
      * C5       1             1
      * Result of getNextConsumer(): C1, C2, C3, C1, C4, C5, C4
      * </pre>
      *
      * <pre>
      * <b>Algorithm:</b>
      * 1. consumerList: it stores consumers in sorted-list: max-priority stored first
      * 2. currentConsumerRoundRobinIndex: it always stores last served consumer-index
      *
      * Each time getNextConsumer() is called:<p>
      * 1. It always starts to traverse from the max-priority consumer (first element) from sorted-list
      * 2. Consumers on same priority-level will be treated equally and it tries to pick one of them in
      *    round-robin manner
      * 3. If consumer is not available on given priority-level then only it will go to the next lower priority-level
      *    consumers
      * 4. Returns null in case it doesn't find any available consumer
      * </pre>
      *
      * @return nextAvailableConsumer
      */
     public Consumer getNextConsumer() {
         if (consumerList.isEmpty() || IS_CLOSED_UPDATER.get(this) == TRUE) {
             // abort read if no consumers are connected or if disconnect is initiated
             return null;
         }

         if (currentConsumerRoundRobinIndex >= consumerList.size()) {
             currentConsumerRoundRobinIndex = 0;
         }

         int currentRoundRobinConsumerPriority = consumerList.get(currentConsumerRoundRobinIndex).getPriorityLevel();

         // first find available-consumer on higher level unless currentIndex is not on highest level which is 0
         if (currentRoundRobinConsumerPriority != 0) {
             int higherPriorityConsumerIndex = getConsumerFromHigherPriority(currentRoundRobinConsumerPriority);
             if (higherPriorityConsumerIndex != -1) {
                 currentConsumerRoundRobinIndex = higherPriorityConsumerIndex + 1;
                 return consumerList.get(higherPriorityConsumerIndex);
             }
         }

         // currentIndex is already on highest level or couldn't find consumer on higher level so, find consumer on same
         // or lower level
         int availableConsumerIndex = getNextConsumerFromSameOrLowerLevel(currentConsumerRoundRobinIndex);
         if (availableConsumerIndex != -1) {
             currentConsumerRoundRobinIndex = availableConsumerIndex + 1;
             return consumerList.get(availableConsumerIndex);
         }

         // couldn't find available consumer
         return null;
     }

     /**
      * Get random consumer from consumerList.
      *
      * @return null if no consumer available, else return random consumer from consumerList
      */
     public Consumer getRandomConsumer() {
         if (consumerList.isEmpty() || IS_CLOSED_UPDATER.get(this) == TRUE) {
             // abort read if no consumers are connected of if disconnect is initiated
             return null;
         }

         return consumerList.get(random.nextInt(consumerList.size()));
     }


     /**
      * Finds index of first available consumer which has higher priority then given targetPriority.
      *
      * @param targetPriority
      * @return -1 if couldn't find any available consumer
      */
     private int getConsumerFromHigherPriority(int targetPriority) {
         for (int i = 0; i < currentConsumerRoundRobinIndex; i++) {
             Consumer consumer = consumerList.get(i);
             if (consumer.getPriorityLevel() < targetPriority) {
                 if (isConsumerAvailable(consumerList.get(i))) {
                     return i;
                 }
             } else {
                 break;
             }
         }
         return -1;
     }

     /**
      * Finds index of round-robin available consumer that present on same level as consumer on
      * currentRoundRobinIndex if doesn't find consumer on same level then it finds first available consumer on lower
      * priority level else returns
      * index=-1 if couldn't find any available consumer in the list.
      *
      * @param currentRoundRobinIndex
      * @return
      */
     private int getNextConsumerFromSameOrLowerLevel(int currentRoundRobinIndex) {

         int targetPriority = consumerList.get(currentRoundRobinIndex).getPriorityLevel();
         // use to do round-robin if can't find consumer from currentRR to last-consumer in list
         int scanIndex = currentRoundRobinIndex;
         int endPriorityLevelIndex = currentRoundRobinIndex;
         do {
             Consumer scanConsumer = scanIndex < consumerList.size() ? consumerList.get(scanIndex)
                     : null /* reached to last consumer of list */;

             // if reached to last consumer of list then check from beginning to currentRRIndex of the list
             if (scanConsumer == null || scanConsumer.getPriorityLevel() != targetPriority) {
                 endPriorityLevelIndex = scanIndex; // last consumer on this level
                 scanIndex = getFirstConsumerIndexOfPriority(targetPriority);
             } else {
                 if (isConsumerAvailable(scanConsumer)) {
                     return scanIndex;
                 }
                 scanIndex++;
             }
         } while (scanIndex != currentRoundRobinIndex);

         // it means: didn't find consumer in the same priority-level so, check available consumer lower than this level
         for (int i = endPriorityLevelIndex; i < consumerList.size(); i++) {
             if (isConsumerAvailable(consumerList.get(i))) {
                 return i;
             }
         }

         return -1;
     }

     /**
      * Finds index of first consumer in list which has same priority as given targetPriority.
      *
      * @param targetPriority
      * @return
      */
     private int getFirstConsumerIndexOfPriority(int targetPriority) {
         for (int i = 0; i < consumerList.size(); i++) {
             if (consumerList.get(i).getPriorityLevel() == targetPriority) {
                 return i;
             }
         }
         return -1;
     }

     private static final Logger log = LoggerFactory.getLogger(PersistentStickyKeyDispatcherMultipleConsumers.class);


 }
	/**
	* 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.pulsar.broker.service;

	import com.carrotsearch.hppc.ObjectHashSet;
	import com.carrotsearch.hppc.ObjectSet;
	import java.util.Random;
	import java.util.concurrent.CopyOnWriteArrayList;
	import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
	import org.apache.pulsar.broker.service.persistent.PersistentStickyKeyDispatcherMultipleConsumers;
	import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	/**
	*
	*/
	public abstract class AbstractDispatcherMultipleConsumers extends AbstractBaseDispatcher {

	protected final CopyOnWriteArrayList<Consumer> consumerList = new CopyOnWriteArrayList<>();
	protected final ObjectSet<Consumer> consumerSet = new ObjectHashSet<>();
	protected volatile int currentConsumerRoundRobinIndex = 0;

	protected static final int FALSE = 0;
	protected static final int TRUE = 1;
	protected static final AtomicIntegerFieldUpdater<AbstractDispatcherMultipleConsumers> IS_CLOSED_UPDATER =
	AtomicIntegerFieldUpdater
	.newUpdater(AbstractDispatcherMultipleConsumers.class, "isClosed");
	private volatile int isClosed = FALSE;

	private Random random = new Random(42);

	protected AbstractDispatcherMultipleConsumers(Subscription subscription) {
	super(subscription);
	}

	public boolean isConsumerConnected() {
	return !consumerList.isEmpty();
	}

	public CopyOnWriteArrayList<Consumer> getConsumers() {
	return consumerList;
	}

	public synchronized boolean canUnsubscribe(Consumer consumer) {
	return consumerList.size() == 1 && consumerSet.contains(consumer);
	}

	public boolean isClosed() {
	return isClosed == TRUE;
	}

	public SubType getType() {
	return SubType.Shared;
	}

	public abstract boolean isConsumerAvailable(Consumer consumer);

	protected void cancelPendingRead() {}

	/**
	* <pre>
	* Broker gives more priority while dispatching messages. Here, broker follows descending priorities. (eg:
	* 0=max-priority, 1, 2,..)
	* <p>
	* Broker will first dispatch messages to max priority-level consumers if they
	* have permits, else broker will consider next priority level consumers.
	* Also on the same priority-level, it selects consumer in round-robin manner.
	* <p>
	* If subscription has consumer-A with priorityLevel 1 and Consumer-B with priorityLevel 2
	* then broker will dispatch
	* messages to only consumer-A until it runs out permit and then broker starts dispatching messages to Consumer-B.
	* <p>
	* Consumer PriorityLevel Permits
	* C1 0 2
	* C2 0 1
	* C3 0 1
	* C4 1 2
	* C5 1 1
	* Result of getNextConsumer(): C1, C2, C3, C1, C4, C5, C4
	* </pre>
	*
	* <pre>
	* <b>Algorithm:</b>
	* 1. consumerList: it stores consumers in sorted-list: max-priority stored first
	* 2. currentConsumerRoundRobinIndex: it always stores last served consumer-index
	*
	* Each time getNextConsumer() is called:<p>
	* 1. It always starts to traverse from the max-priority consumer (first element) from sorted-list
	* 2. Consumers on same priority-level will be treated equally and it tries to pick one of them in
	* round-robin manner
	* 3. If consumer is not available on given priority-level then only it will go to the next lower priority-level
	* consumers
	* 4. Returns null in case it doesn't find any available consumer
	* </pre>
	*
	* @return nextAvailableConsumer
	*/
	public Consumer getNextConsumer() {
	if (consumerList.isEmpty() \|\| IS_CLOSED_UPDATER.get(this) == TRUE) {
	// abort read if no consumers are connected or if disconnect is initiated
	return null;
	}

	if (currentConsumerRoundRobinIndex >= consumerList.size()) {
	currentConsumerRoundRobinIndex = 0;
	}

	int currentRoundRobinConsumerPriority = consumerList.get(currentConsumerRoundRobinIndex).getPriorityLevel();

	// first find available-consumer on higher level unless currentIndex is not on highest level which is 0
	if (currentRoundRobinConsumerPriority != 0) {
	int higherPriorityConsumerIndex = getConsumerFromHigherPriority(currentRoundRobinConsumerPriority);
	if (higherPriorityConsumerIndex != -1) {
	currentConsumerRoundRobinIndex = higherPriorityConsumerIndex + 1;
	return consumerList.get(higherPriorityConsumerIndex);
	}
	}

	// currentIndex is already on highest level or couldn't find consumer on higher level so, find consumer on same
	// or lower level
	int availableConsumerIndex = getNextConsumerFromSameOrLowerLevel(currentConsumerRoundRobinIndex);
	if (availableConsumerIndex != -1) {
	currentConsumerRoundRobinIndex = availableConsumerIndex + 1;
	return consumerList.get(availableConsumerIndex);
	}

	// couldn't find available consumer
	return null;
	}

	/**
	* Get random consumer from consumerList.
	*
	* @return null if no consumer available, else return random consumer from consumerList
	*/
	public Consumer getRandomConsumer() {
	if (consumerList.isEmpty() \|\| IS_CLOSED_UPDATER.get(this) == TRUE) {
	// abort read if no consumers are connected of if disconnect is initiated
	return null;
	}

	return consumerList.get(random.nextInt(consumerList.size()));
	}


	/**
	* Finds index of first available consumer which has higher priority then given targetPriority.
	*
	* @param targetPriority
	* @return -1 if couldn't find any available consumer
	*/
	private int getConsumerFromHigherPriority(int targetPriority) {
	for (int i = 0; i < currentConsumerRoundRobinIndex; i++) {
	Consumer consumer = consumerList.get(i);
	if (consumer.getPriorityLevel() < targetPriority) {
	if (isConsumerAvailable(consumerList.get(i))) {
	return i;
	}
	} else {
	break;
	}
	}
	return -1;
	}

	/**
	* Finds index of round-robin available consumer that present on same level as consumer on
	* currentRoundRobinIndex if doesn't find consumer on same level then it finds first available consumer on lower
	* priority level else returns
	* index=-1 if couldn't find any available consumer in the list.
	*
	* @param currentRoundRobinIndex
	* @return
	*/
	private int getNextConsumerFromSameOrLowerLevel(int currentRoundRobinIndex) {

	int targetPriority = consumerList.get(currentRoundRobinIndex).getPriorityLevel();
	// use to do round-robin if can't find consumer from currentRR to last-consumer in list
	int scanIndex = currentRoundRobinIndex;
	int endPriorityLevelIndex = currentRoundRobinIndex;
	do {
	Consumer scanConsumer = scanIndex < consumerList.size() ? consumerList.get(scanIndex)
	: null /* reached to last consumer of list */;

	// if reached to last consumer of list then check from beginning to currentRRIndex of the list
	if (scanConsumer == null \|\| scanConsumer.getPriorityLevel() != targetPriority) {
	endPriorityLevelIndex = scanIndex; // last consumer on this level
	scanIndex = getFirstConsumerIndexOfPriority(targetPriority);
	} else {
	if (isConsumerAvailable(scanConsumer)) {
	return scanIndex;
	}
	scanIndex++;
	}
	} while (scanIndex != currentRoundRobinIndex);

	// it means: didn't find consumer in the same priority-level so, check available consumer lower than this level
	for (int i = endPriorityLevelIndex; i < consumerList.size(); i++) {
	if (isConsumerAvailable(consumerList.get(i))) {
	return i;
	}
	}

	return -1;
	}

	/**
	* Finds index of first consumer in list which has same priority as given targetPriority.
	*
	* @param targetPriority
	* @return
	*/
	private int getFirstConsumerIndexOfPriority(int targetPriority) {
	for (int i = 0; i < consumerList.size(); i++) {
	if (consumerList.get(i).getPriorityLevel() == targetPriority) {
	return i;
	}
	}
	return -1;
	}

	private static final Logger log = LoggerFactory.getLogger(PersistentStickyKeyDispatcherMultipleConsumers.class);


	}