common/util/src/main/java/org/opendaylight/yangtools/util/concurrent/TrackingLinkedBlockingQueue.java

   1 /*
   2  * Copyright (c) 2014 Brocade Communications Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.yangtools.util.concurrent;
  10
  11 import com.google.common.annotations.Beta;
  12 import java.util.Collection;
  13 import java.util.concurrent.LinkedBlockingQueue;
  14 import java.util.concurrent.TimeUnit;
  15 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
  16
  17 /**
  18  * A {@link LinkedBlockingQueue} that tracks the largest queue size for debugging.
  19  *
  20  * @author Thomas Pantelis
  21  *
  22  * @param <E> the element t.ype
  23  */
  24 public class TrackingLinkedBlockingQueue<E> extends LinkedBlockingQueue<E> {
  25     @SuppressWarnings("rawtypes")
  26     private static final AtomicIntegerFieldUpdater<TrackingLinkedBlockingQueue> LARGEST_QUEUE_SIZE_UPDATER = AtomicIntegerFieldUpdater.newUpdater(TrackingLinkedBlockingQueue.class, "largestQueueSize");
  27     private static final long serialVersionUID = 1L;
  28
  29     /**
  30      * Holds largestQueueSize, this long field should be only accessed
  31      * using {@link #LARGEST_QUEUE_SIZE_UPDATER}
  32      */
  33     private volatile int largestQueueSize = 0;
  34
  35     /**
  36      * @see LinkedBlockingQueue#LinkedBlockingQueue
  37      */
  38     public TrackingLinkedBlockingQueue() {
  39         super();
  40     }
  41
  42     /**
  43      * @see LinkedBlockingQueue#LinkedBlockingQueue(Collection)
  44      */
  45     public TrackingLinkedBlockingQueue( final Collection<? extends E> c ) {
  46         super(c);
  47     }
  48
  49     /**
  50      * @see LinkedBlockingQueue#LinkedBlockingQueue(int)
  51      */
  52     public TrackingLinkedBlockingQueue( final int capacity ) {
  53         super(capacity);
  54     }
  55
  56     /**
  57      * Returns the largest queue size.
  58      *
  59      * FIXME: the this return will be changed to int in a future release.
  60      */
  61     @Beta
  62     public long getLargestQueueSize() {
  63         return largestQueueSize;
  64     }
  65
  66     @Override
  67     public boolean offer( final E e, final long timeout, final TimeUnit unit ) throws InterruptedException {
  68         if (super.offer( e, timeout, unit ) ) {
  69             updateLargestQueueSize();
  70             return true;
  71         }
  72
  73         return false;
  74     }
  75
  76     @Override
  77     public boolean offer( final E e ) {
  78         if (super.offer( e ) ) {
  79             updateLargestQueueSize();
  80             return true;
  81         }
  82
  83         return false;
  84     }
  85
  86     @Override
  87     public void put( final E e ) throws InterruptedException {
  88         super.put( e );
  89         updateLargestQueueSize();
  90     }
  91
  92     @Override
  93     public boolean add( final E e ) {
  94         boolean result = super.add( e );
  95         updateLargestQueueSize();
  96         return result;
  97     }
  98
  99     @Override
 100     public boolean addAll( final Collection<? extends E> c ) {
 101         try {
 102             return super.addAll( c );
 103         } finally {
 104             updateLargestQueueSize();
 105         }
 106     }
 107
 108     private void updateLargestQueueSize() {
 109         final int size = size();
 110
 111         int largest;
 112         do {
 113             largest = largestQueueSize;
 114         } while (size > largest && !LARGEST_QUEUE_SIZE_UPDATER.weakCompareAndSet(this, largest, size));
 115     }
 116 }