31024361752

Committed 05 Aug 2026 04:14PM UTC coverage: 64.336% (-0.09%) from 64.422%

Build # 31024361752

Build Type

push

github

Committed by

ibacher

Commit Message

TRUNK-6687: Follow-up: The test here makes incorrect assumptions on the cap

With duplicate records (what we're testing here) it is possible to need
to page over slightly more results than are in the page. This is expected
and was causing this test to fail in some circumstances where index order
became relevant

Coverage Stats

24620 of 38268 relevant lines covered (64.34%)

0.64 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.05

/api/src/main/java/org/openmrs/util/ThreadSafeCircularFifoQueue.java

/**
 * This Source Code Form is subject to the terms of the Mozilla Public License,
 * v. 2.0. If a copy of the MPL was not distributed with this file, You can
 * obtain one at http://mozilla.org/MPL/2.0/. OpenMRS is also distributed under
 * the terms of the Healthcare Disclaimer located at http://openmrs.org/license.
 *
 * Copyright (C) OpenMRS Inc. OpenMRS is a registered trademark and the OpenMRS
 * graphic logo is a trademark of OpenMRS Inc.
 */
package org.openmrs.util;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.Serializable;
import java.lang.ref.WeakReference;
import java.lang.reflect.Array;
import java.util.AbstractQueue;
import java.util.Collection;
import java.util.NoSuchElementException;
import java.util.Objects;
import java.util.Queue;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Predicate;

/**
 * A thread-safe first-in, first-out queue with a fixed size that replaces the oldest element when full.
 * 
 * This class does not support null elements.
 *
 * @param <E> the type of elements in this collection
 * @since 2.4
 */
/*
 * There are already several existing implementations that are similar to this, so why create a new class?
 * Commons Collection's CircularFifoQueue and Guava's EvictingQueue are not thread-safe. The built-in ArrayBlockingQueue
 * is thread-safe, but implements a blocking queue. This involves safety guarantees that we don't need to make as this
 * circular queue is never "full"
 */
public class ThreadSafeCircularFifoQueue<E> extends AbstractQueue<E> implements Queue<E>, Serializable {

        private static final long serialVersionUID = -89162358098721L;

        // queue capacity
        private final int maxElements;

        // Underlying storage
        private final E[] elements;

        private transient ReentrantLock lock = new ReentrantLock();

        // index of the "start" of the queue, i.e., where data is read from
        private int read = 0;

        // index of the "end" of the queue, i.e., where data is written to
        private int write = 0;

        // number of elements in the queue
        private int size;

        // tracks the state of any iterators
        private transient Iterators iterators = null;

        @SuppressWarnings("unused")
        public ThreadSafeCircularFifoQueue() {
                this(32);
        }

        @SuppressWarnings("unchecked")
        public ThreadSafeCircularFifoQueue(int maxElements) {
                if (maxElements <= 0) {
                        throw new IllegalArgumentException("The size must be greater than 0");
                }

                // NB add one more element so that the queue size matches the expected size
                elements = (E[]) new Object[maxElements];
                this.maxElements = elements.length;
        }

        @SuppressWarnings("unused")
        public ThreadSafeCircularFifoQueue(Collection<E> collection) {
                this(collection.size());
                this.addAll(collection);
        }

        @Override
        public boolean add(E e) {
                Objects.requireNonNull(e);

                final ReentrantLock lock = this.lock;
                lock.lock();
                try {
                        internalAdd(e);
                }
                finally {
                        lock.unlock();
                }

                return true;
        }

        @Override
        public boolean addAll(Collection<? extends E> c) {
                Objects.requireNonNull(c);

                final ReentrantLock lock = this.lock;
                lock.lock();
                try {
                        for (E e : c) {
                                Objects.requireNonNull(e);
                                internalAdd(e);
                        }
                }
                finally {
                        lock.unlock();
                }

                return true;
        }

        /**
         * Returns the capacity (i.e., the maximum number of elements stored) of this
         * ThreadSafeCircularFifoQueue
         *
         * @return the capacity of the current queue
         */
        public int capacity() {
                return this.maxElements;
        }

        @Override
        public void clear() {
                final E[] elements = this.elements;
                final ReentrantLock lock = this.lock;
                lock.lock();
                try {
                        if (size > 0) {
                                int idx = read;
                                do {
                                        elements[idx] = null;

                                        idx = increment(idx);
                                } while (idx != write);

                                read = write = size = 0;
                        }
                }
                finally {
                        lock.unlock();
                }
        }

        @Override
        public boolean contains(Object o) {
                if (null == o) {
                        return false;
                }

                final ReentrantLock lock = this.lock;
                lock.lock();
                try {
                        return internalContains(o);
                }
                finally {
                        lock.unlock();
                }
        }

        @Override
        public boolean containsAll(Collection<?> c) {
                if (c == null || c.isEmpty()) {
                        return true;
                }

                final ReentrantLock lock = this.lock;
                lock.lock();
                try {
                        for (Object o : c) {
                                if (!internalContains(o)) {
                                        return false;
                                }
                        }
                        
                        return true;
                }
                finally {
                        lock.unlock();
                }
        }

        @Override
        public E element() {
                final ReentrantLock lock = this.lock;
                lock.lock();
                try {
                        if (size == 0) {
                                throw new NoSuchElementException("queue is empty");
                        }

                        return elements[read];
                }
                finally {
                        lock.unlock();
                }

        }

        @Override
        public java.util.Iterator<E> iterator() {
                return new Iterator();
        }

        @Override
        public boolean isEmpty() {
                final ReentrantLock lock = this.lock;
                lock.lock();
                try {
                        return size == 0;
                }
                finally {
                        lock.unlock();
                }
        }

        @Override
        public boolean offer(E e) {
                return add(e);
        }

        @Override
        public E peek() {
                final ReentrantLock lock = this.lock;
                lock.lock();
                try {
                        return size == 0 ? null : elements[read];
                }
                finally {
                        lock.unlock();
                }
        }

        @Override
        public E poll() {
                final ReentrantLock lock = this.lock;
                lock.lock();
                try {
                        return size == 0 ? null : internalRemove();
                }
                finally {
                        lock.unlock();
                }
        }

        @Override
        public E remove() {
                final ReentrantLock lock = this.lock;
                lock.lock();
                try {
                        if (size == 0) {
                                throw new NoSuchElementException("queue is empty");
                        }

                        return internalRemove();
                }
                finally {
                        lock.unlock();
                }
        }

        @Override
        public boolean remove(Object o) {
                if (null == o) {
                        return false;
                }

                final ReentrantLock lock = this.lock;
                lock.lock();
                try {
                        if (size == 0) {
                                return false;
                        }

                        int idx = this.read;
                        do {
                                if (o.equals(elements[idx])) {
                                        internalRemoveAtIndex(idx);
                                        return true;
                                }

                                idx = increment(idx);
                        } while (idx != write);

                        return false;
                }
                finally {
                        lock.unlock();
                }
        }

        @Override
        public boolean removeAll(Collection<?> c) {
                Objects.requireNonNull(c);

                final ReentrantLock lock = this.lock;
                lock.lock();
                try {
                        return removeIf(c::contains);
                }
                finally {
                        lock.unlock();
                }
        }

        @Override
        public boolean retainAll(Collection<?> c) {
                Objects.requireNonNull(c);

                final ReentrantLock lock = this.lock;
                lock.lock();
                try {
                        return removeIf(o -> !c.contains(o));
                }
                finally {
                        lock.unlock();
                }
        }

        @Override
        public int size() {
                final ReentrantLock lock = this.lock;
                lock.lock();
                try {
                        return size;
                }
                finally {
                        lock.unlock();
                }
        }

        @Override
        public Object[] toArray() {
                return toArray(new Object[0]);
        }

        @Override
        @SuppressWarnings("unchecked")
        public <T> T[] toArray(T[] a) {
                Objects.requireNonNull(a);

                final ReentrantLock lock = this.lock;
                lock.lock();
                try {
                        final int size = this.size;
                        final T[] result = a.length < size ? (T[]) Array.newInstance(a.getClass().getComponentType(), size) : a;

                        final int n = elements.length - read;
                        if (size <= n) {
                                System.arraycopy(elements, read, result, 0, size);
                        } else {
                                System.arraycopy(elements, read, result, 0, n);
                                System.arraycopy(elements, 0, result, n, size - n);
                        }

                        if (result.length > size) {
                                for (int i = Math.max(0, size - 1); i < result.length; i++) {
                                        result[i] = null;
                                }
                        }

                        return result;
                }
                finally {
                        lock.unlock();
                }
        }

        public String toString() {
                final ReentrantLock lock = this.lock;
                lock.lock();
                try {
                        if (size == 0) {
                                return "[]";
                        }

                        StringBuilder sb = new StringBuilder();
                        sb.append('[');

                        int idx = read;
                        while (true) {
                                E e = elements[idx];
                                sb.append(e == this ? "(this Collection)" : e);

                                idx = (idx + 1) % maxElements;

                                if (idx == write) {
                                        return sb.append(']').toString();
                                } else {
                                        sb.append(',').append(' ');
                                }
                        }
                }
                finally {
                        lock.unlock();
                }
        }

        /* Internal implementations: MUST BE USED INSIDE LOCKS  */
        
        private int increment(int i) {
                return (i + 1) % maxElements;
        }

        private int decrement(int i) {
                return ((i == 0) ? maxElements : i) - 1;
        }

        private void internalAdd(E e) {
                if (size == maxElements) {
                        internalRemove();
                }

                size++;
                elements[write] = e;
                write = increment(write);
        }
        
        private boolean internalContains(Object o) {
                if (size > 0) {
                        int idx = read;
                        do {
                                if (o.equals(elements[idx])) {
                                        return true;
                                }

                                idx = (idx + 1) % maxElements;
                        } while (idx != write);
                }

                return false;
        }

        private E internalRemove() {
                final E element = elements[read];

                if (null != element) {
                        internalRemoveAtIndex(read);
                }

                return element;
        }

        private void internalRemoveAtIndex(int idx) {
                if (idx == read) {
                        elements[read] = null;

                        read = increment(read);

                        size--;
                        if (iterators != null) {
                                iterators.elementRemoved();
                        }
                } else {
                        int i = idx;
                        while (true) {
                                int next = increment(i);

                                if (next != write) {
                                        elements[i] = elements[next];
                                        i = next;
                                } else {
                                        elements[i] = null;
                                        write = i;
                                        break;
                                }
                        }

                        size--;
                        if (iterators != null) {
                                iterators.removedAt(idx);
                        }
                }
        }

        private void readObject(final ObjectInputStream in) throws IOException, ClassNotFoundException {
                in.defaultReadObject();
                lock = new ReentrantLock();
        }

        /**
         * A linked list maintaining references between the queue and any iterators
         * 
         * This class exists to ensure that iterator objects are properly updated when items are removed from the queue and
         * are invalidated if the underlying queue becomes incompatible with the iterator's view of it.
         * 
         * This is based on the implementation of ArrayBlockingQueue.Itrs and involves the same garbage collection scheme
         * described there.
         */
        private class Iterators {

                private static final int SHORT_SWEEP_PROBES = 4;

                private static final int LONG_SWEEP_PROBES = 16;

                private Node head;

                private Node sweeper = null;

                int cycles = 0;

                Iterators(Iterator iterator) {
                        register(iterator);
                }

                void register(Iterator iterator) {
                        head = new Node(iterator, head);
                }

                void elementRemoved() {
                        if (size == 0) {
                                queueEmptied();
                        } else if (read == 0) {
                                readWrapped();
                        }
                }

                void queueEmptied() {
                        for (Node p = head; p != null; p = p.next) {
                                Iterator it = p.get();
                                if (it != null) {
                                        p.clear();
                                        it.shutdown();
                                }
                        }

                        head = null;
                        iterators = null;
                }

                void removedAt(int removedIndex) {
                        prune(it -> it.removedAt(removedIndex));
                }

                void readWrapped() {
                        cycles++;
                        prune(Iterator::readWrapped);
                }

                void sweep(boolean tryHarder) {
                        int probes = tryHarder ? LONG_SWEEP_PROBES : SHORT_SWEEP_PROBES;
                        Node o, p;
                        final Node sweeper = this.sweeper;
                        boolean completeCycle;   // to limit search to one full sweep

                        if (sweeper == null) {
                                o = null;
                                p = head;
                                completeCycle = true;
                        } else {
                                o = sweeper;
                                p = o.next;
                                completeCycle = false;
                        }

                        for (; probes > 0; probes--) {
                                if (p == null) {
                                        if (completeCycle) {
                                                break;
                                        }

                                        o = null;
                                        p = head;
                                        completeCycle = true;
                                }

                                final Iterator it = p.get();
                                final Node next = p.next;
                                if (it == null || it.isDetached()) {
                                        // found a discarded/exhausted iterator
                                        probes = LONG_SWEEP_PROBES; // "try harder"
                                        // unlink p
                                        p.clear();
                                        p.next = null;
                                        if (o == null) {
                                                head = next;
                                                if (next == null) {
                                                        // We've run out of iterators to track; retire
                                                        iterators = null;
                                                        return;
                                                }
                                        } else {
                                                o.next = next;
                                        }
                                } else {
                                        o = p;
                                }
                                p = next;
                        }

                        this.sweeper = (p == null) ? null : o;
                }

                private void prune(Predicate<Iterator> shouldRemove) {
                        for (Node o = null, p = head; p != null; ) {
                                final Iterator it = p.get();
                                final Node next = p.next;

                                if (it == null || shouldRemove.test(it)) {
                                        p.clear();
                                        p.next = null;

                                        if (o == null) {
                                                head = next;
                                        } else {
                                                o.next = next;
                                        }
                                } else {
                                        o = p;
                                }

                                p = next;
                        }

                        if (head == null) {
                                ThreadSafeCircularFifoQueue.this.iterators = null;
                        }
                }

                /**
                 * The actual list node implementation
                 */
                private class Node extends WeakReference<Iterator> {

                        Node next;

                        Node(Iterator iterator, Node next) {
                                super(iterator);
                                this.next = next;
                        }
                }
        }

        /**
         * An attempt to be a straight-forward iterator implementation for a ThreadSafeCircularFifoQueue.
         * 
         * It should iterate over each member in the queue only once, assuming that the queue is not modified while this
         * iterator remains in use. If the underlying queue is modified, the iterator will attempt to recover and keep going.
         * 
         * If it becomes too far out of synch with the underlying data, an iterator may fail before iterating over all elements
         * in a queue. However if {@link #hasNext()} returns true, {@link #next()} will always return a result.
         */
        private class Iterator implements java.util.Iterator<E> {

                /* Special index values */

                // indicates an index that is invalid or empty
                private static final int NONE = -1;

                // used as a value for prevRead when the iterator is no longer valid
                private static final int DETACHED = -2;

                private int nextIndex;

                private E nextItem;

                private int prevIndex = NONE;

                private E prevItem = null;

                private int prevRead;

                private int prevCycles;

                Iterator() {
                        final ReentrantLock lock = ThreadSafeCircularFifoQueue.this.lock;
                        lock.lock();
                        try {
                                if (size == 0) {
                                        nextIndex = NONE;
                                        prevRead = DETACHED;
                                } else {
                                        nextItem = elements[read];
                                        nextIndex = read;
                                        prevRead = read;

                                        if (iterators == null) {
                                                iterators = new Iterators(this);
                                        } else {
                                                iterators.register(this);
                                                iterators.sweep(false);
                                        }

                                        prevCycles = iterators.cycles;
                                }
                        }
                        finally {
                                lock.unlock();
                        }
                }

                @Override
                public boolean hasNext() {
                        // no locks for the simplest case
                        if (nextItem != null) {
                                return true;
                        }

                        final ReentrantLock lock = ThreadSafeCircularFifoQueue.this.lock;
                        lock.lock();
                        try {
                                if (!isDetached()) {
                                        updateIndices();
                                        if (prevIndex >= 0) {
                                                prevItem = elements[prevIndex];
                                                detach();
                                        }
                                }
                        }
                        finally {
                                lock.unlock();
                        }
                        
                        return false;
                }

                @Override
                public E next() {
                        final E it = nextItem;
                        if (it == null) {
                                throw new NoSuchElementException();
                        }

                        final ReentrantLock lock = ThreadSafeCircularFifoQueue.this.lock;
                        lock.lock();
                        try {
                                if (!isDetached()) {
                                        updateIndices();
                                }
                                
                                prevIndex = nextIndex;
                                prevItem = it;

                                if (nextIndex < 0) {
                                        nextIndex = NONE;
                                        nextItem = null;
                                } else {
                                        nextIndex = increment(nextIndex);
                                        if (nextIndex == write) {
                                                nextIndex = NONE;
                                                nextItem = null;
                                        } else {
                                                nextItem = elements[nextIndex];
                                        }
                                }

                                return it;
                        }
                        finally {
                                lock.unlock();
                        }
                }

                @Override
                public void remove() {
                        final ReentrantLock lock = ThreadSafeCircularFifoQueue.this.lock;
                        lock.lock();
                        try {
                                if (!isDetached()) {
                                        updateIndices();
                                }
                                
                                if (prevIndex == NONE) {
                                        throw new IllegalStateException();
                                } else if (prevIndex >= 0 && elements[prevIndex] == prevItem) {
                                        internalRemoveAtIndex(prevIndex);

                                        if (prevIndex != read) {
                                                nextIndex = Math.max(prevIndex, read);
                                        }
                                }

                                prevIndex = NONE;
                                prevItem = null;
                                
                                if (nextIndex < 0) {
                                        detach();
                                }
                        }
                        finally {
                                lock.unlock();
                        }
                }

                boolean readWrapped() {
                        if (isDetached()) {
                                return true;
                        }

                        if (iterators.cycles - prevCycles > 1) {
                                shutdown();
                                return true;
                        }

                        return false;
                }

                boolean removedAt(int removedIndex) {
                        if (isDetached()) {
                                return true;
                        }

                        final int cycles = iterators.cycles;
                        final int read = ThreadSafeCircularFifoQueue.this.read;
                        
                        int cycleDiff = cycles - prevCycles;

                        if (removedIndex < read) {
                                cycleDiff++;
                        }

                        final int removedDistance = (cycleDiff * maxElements) + (removedIndex - prevRead);

                        if (prevIndex >= 0) {
                                int x = distance(prevIndex);
                                if (x == removedDistance) {
                                        prevIndex = NONE;
                                } else if (x > removedDistance) {
                                        prevIndex = decrement(prevIndex);
                                }
                        }

                        if (nextIndex >= 0) {
                                int x = distance(nextIndex);
                                if (x == removedDistance) {
                                        nextIndex = NONE;
                                } else if (x > removedDistance) {
                                        nextIndex = decrement(nextIndex);
                                }
                        } else if (prevIndex < 0) {
                                // don't call detach() as returning true will trigger a full sweep anyways
                                prevRead = DETACHED;
                                return true;
                        }

                        return false;
                }

                void shutdown() {
                        // nextItem is not set to null as it has been cached and can be returned
                        nextIndex = nextIndex >= 0 ? NONE : nextIndex;
                        prevIndex = prevIndex >= 0 ? NONE : nextIndex;
                        prevItem = null;
                        prevRead = DETACHED;
                }

                /**
                 * Detach the iterator and trigger a sweep of the iterator queue
                 */
                private void detach() {
                        if (prevRead >= 0) {
                                prevRead = DETACHED;
                                iterators.sweep(true);
                        }
                }

                private boolean isDetached() {
                        return prevRead < 0;
                }

                private int distance(int index) {
                        int distance = index - prevRead;
                        if (distance < 0) {
                                distance += maxElements;
                        }
                        return distance;
                }
                
                private boolean indexInvalidated(int index, long dequeues) {
                        if (index < 0) {
                                return false;
                        }
                        
                        int distance = distance(index);
                        
                        return dequeues > distance;
                }
                
                private void updateIndices() {
                        final int cycles = ThreadSafeCircularFifoQueue.this.iterators.cycles;
                        if (cycles != prevCycles || read != prevRead) {
                                long dequeues = (cycles - prevCycles) * maxElements + (read - prevRead);
                                
                                if (indexInvalidated(prevIndex, dequeues)) {
                                        prevIndex = NONE;
                                }

                                if (indexInvalidated(nextIndex, dequeues)) {
                                        nextIndex = NONE;
                                }
                                
                                if (prevIndex < 0 && nextIndex < 0) {
                                        detach();
                                } else {
                                        prevCycles = cycles;
                                        prevRead = read;
                                }
                        }
                }
        }
}

1	/**
2	* This Source Code Form is subject to the terms of the Mozilla Public License,
3	* v. 2.0. If a copy of the MPL was not distributed with this file, You can
4	* obtain one at http://mozilla.org/MPL/2.0/. OpenMRS is also distributed under
5	* the terms of the Healthcare Disclaimer located at http://openmrs.org/license.
6	*
7	* Copyright (C) OpenMRS Inc. OpenMRS is a registered trademark and the OpenMRS
8	* graphic logo is a trademark of OpenMRS Inc.
9	*/
10	package org.openmrs.util;
11
12	import java.io.IOException;
13	import java.io.ObjectInputStream;
14	import java.io.Serializable;
15	import java.lang.ref.WeakReference;
16	import java.lang.reflect.Array;
17	import java.util.AbstractQueue;
18	import java.util.Collection;
19	import java.util.NoSuchElementException;
20	import java.util.Objects;
21	import java.util.Queue;
22	import java.util.concurrent.locks.ReentrantLock;
23	import java.util.function.Predicate;
24
25	/**
26	* A thread-safe first-in, first-out queue with a fixed size that replaces the oldest element when full.
27	*
28	* This class does not support null elements.
29	*
30	* @param <E> the type of elements in this collection
31	* @since 2.4
32	*/
33	/*
34	* There are already several existing implementations that are similar to this, so why create a new class?
35	* Commons Collection's CircularFifoQueue and Guava's EvictingQueue are not thread-safe. The built-in ArrayBlockingQueue
36	* is thread-safe, but implements a blocking queue. This involves safety guarantees that we don't need to make as this
37	* circular queue is never "full"
38	*/
39	public class ThreadSafeCircularFifoQueue<E> extends AbstractQueue<E> implements Queue<E>, Serializable {
40
41	private static final long serialVersionUID = -89162358098721L;
42
43	// queue capacity
44	private final int maxElements;
45
46	// Underlying storage
47	private final E[] elements;
48
49	private transient ReentrantLock lock = new ReentrantLock();	1✔
50
51	// index of the "start" of the queue, i.e., where data is read from
52	private int read = 0;	1✔
53
54	// index of the "end" of the queue, i.e., where data is written to
55	private int write = 0;	1✔
56
57	// number of elements in the queue
58	private int size;
59
60	// tracks the state of any iterators
61	private transient Iterators iterators = null;	1✔
62
63	@SuppressWarnings("unused")
64	public ThreadSafeCircularFifoQueue() {
65	this(32);	×
66	}	×
67
68	@SuppressWarnings("unchecked")
69	public ThreadSafeCircularFifoQueue(int maxElements) {	1✔
70	if (maxElements <= 0) {	1✔
71	throw new IllegalArgumentException("The size must be greater than 0");	×
72	}
73
74	// NB add one more element so that the queue size matches the expected size
75	elements = (E[]) new Object[maxElements];	1✔
76	this.maxElements = elements.length;	1✔
77	}	1✔
78
79	@SuppressWarnings("unused")
80	public ThreadSafeCircularFifoQueue(Collection<E> collection) {
81	this(collection.size());	×
82	this.addAll(collection);	×
83	}	×
84
85	@Override
86	public boolean add(E e) {
87	Objects.requireNonNull(e);	1✔
88
89	final ReentrantLock lock = this.lock;	1✔
90	lock.lock();	1✔
91	try {
92	internalAdd(e);	1✔
93	}
94	finally {
95	lock.unlock();	1✔
96	}
97
98	return true;	1✔
99	}
100
101	@Override
102	public boolean addAll(Collection<? extends E> c) {
103	Objects.requireNonNull(c);	1✔
104
105	final ReentrantLock lock = this.lock;	1✔
106	lock.lock();	1✔
107	try {
108	for (E e : c) {	1✔
109	Objects.requireNonNull(e);	1✔
110	internalAdd(e);	1✔
111	}	1✔
112	}
113	finally {
114	lock.unlock();	1✔
115	}
116
117	return true;	1✔
118	}
119
120	/**
121	* Returns the capacity (i.e., the maximum number of elements stored) of this
122	* ThreadSafeCircularFifoQueue
123	*
124	* @return the capacity of the current queue
125	*/
126	public int capacity() {
127	return this.maxElements;	1✔
128	}
129
130	@Override
131	public void clear() {
132	final E[] elements = this.elements;	1✔
133	final ReentrantLock lock = this.lock;	1✔
134	lock.lock();	1✔
135	try {
136	if (size > 0) {	1✔
137	int idx = read;	1✔
138	do {
139	elements[idx] = null;	1✔
140
141	idx = increment(idx);	1✔
142	} while (idx != write);	1✔
143
144	read = write = size = 0;	1✔
145	}
146	}
147	finally {
148	lock.unlock();	1✔
149	}
150	}	1✔
151
152	@Override
153	public boolean contains(Object o) {
154	if (null == o) {	1✔
155	return false;	×
156	}
157
158	final ReentrantLock lock = this.lock;	1✔
159	lock.lock();	1✔
160	try {
161	return internalContains(o);	1✔
162	}
163	finally {
164	lock.unlock();	1✔
165	}
166	}
167
168	@Override
169	public boolean containsAll(Collection<?> c) {
170	if (c == null \|\| c.isEmpty()) {	1✔
171	return true;	1✔
172	}
173
174	final ReentrantLock lock = this.lock;	1✔
175	lock.lock();	1✔
176	try {
177	for (Object o : c) {	1✔
178	if (!internalContains(o)) {	1✔
179	return false;	1✔
180	}
181	}	1✔
182
183	return true;	1✔
184	}
185	finally {
186	lock.unlock();	1✔
187	}
188	}
189
190	@Override
191	public E element() {
192	final ReentrantLock lock = this.lock;	1✔
193	lock.lock();	1✔
194	try {
195	if (size == 0) {	1✔
196	throw new NoSuchElementException("queue is empty");	1✔
197	}
198
199	return elements[read];	1✔
200	}
201	finally {
202	lock.unlock();	1✔
203	}
204
205	}
206
207	@Override
208	public java.util.Iterator<E> iterator() {
209	return new Iterator();	1✔
210	}
211
212	@Override
213	public boolean isEmpty() {
214	final ReentrantLock lock = this.lock;	1✔
215	lock.lock();	1✔
216	try {
217	return size == 0;	1✔
218	}
219	finally {
220	lock.unlock();	1✔
221	}
222	}
223
224	@Override
225	public boolean offer(E e) {
226	return add(e);	1✔
227	}
228
229	@Override
230	public E peek() {
231	final ReentrantLock lock = this.lock;	1✔
232	lock.lock();	1✔
233	try {
234	return size == 0 ? null : elements[read];	1✔
235	}
236	finally {
237	lock.unlock();	1✔
238	}
239	}
240
241	@Override
242	public E poll() {
243	final ReentrantLock lock = this.lock;	1✔
244	lock.lock();	1✔
245	try {
246	return size == 0 ? null : internalRemove();	1✔
247	}
248	finally {
249	lock.unlock();	1✔
250	}
251	}
252
253	@Override
254	public E remove() {
255	final ReentrantLock lock = this.lock;	1✔
256	lock.lock();	1✔
257	try {
258	if (size == 0) {	1✔
259	throw new NoSuchElementException("queue is empty");	1✔
260	}
261
262	return internalRemove();	1✔
263	}
264	finally {
265	lock.unlock();	1✔
266	}
267	}
268
269	@Override
270	public boolean remove(Object o) {
271	if (null == o) {	1✔
272	return false;	×
273	}
274
275	final ReentrantLock lock = this.lock;	1✔
276	lock.lock();	1✔
277	try {
278	if (size == 0) {	1✔
279	return false;	1✔
280	}
281
282	int idx = this.read;	1✔
283	do {
284	if (o.equals(elements[idx])) {	1✔
285	internalRemoveAtIndex(idx);	1✔
286	return true;	1✔
287	}
288
289	idx = increment(idx);	1✔
290	} while (idx != write);	1✔
291
292	return false;	1✔
293	}
294	finally {
295	lock.unlock();	1✔
296	}
297	}
298
299	@Override
300	public boolean removeAll(Collection<?> c) {
301	Objects.requireNonNull(c);	1✔
302
303	final ReentrantLock lock = this.lock;	1✔
304	lock.lock();	1✔
305	try {
306	return removeIf(c::contains);	1✔
307	}
308	finally {
309	lock.unlock();	1✔
310	}
311	}
312
313	@Override
314	public boolean retainAll(Collection<?> c) {
315	Objects.requireNonNull(c);	1✔
316
317	final ReentrantLock lock = this.lock;	1✔
318	lock.lock();	1✔
319	try {
320	return removeIf(o -> !c.contains(o));	1✔
321	}
322	finally {
323	lock.unlock();	1✔
324	}
325	}
326
327	@Override
328	public int size() {
329	final ReentrantLock lock = this.lock;	1✔
330	lock.lock();	1✔
331	try {
332	return size;	1✔
333	}
334	finally {
335	lock.unlock();	1✔
336	}
337	}
338
339	@Override
340	public Object[] toArray() {
341	return toArray(new Object[0]);	1✔
342	}
343
344	@Override
345	@SuppressWarnings("unchecked")
346	public <T> T[] toArray(T[] a) {
347	Objects.requireNonNull(a);	1✔
348
349	final ReentrantLock lock = this.lock;	1✔
350	lock.lock();	1✔
351	try {
352	final int size = this.size;	1✔
353	final T[] result = a.length < size ? (T[]) Array.newInstance(a.getClass().getComponentType(), size) : a;	1✔
354
355	final int n = elements.length - read;	1✔
356	if (size <= n) {	1✔
357	System.arraycopy(elements, read, result, 0, size);	1✔
358	} else {
359	System.arraycopy(elements, read, result, 0, n);	1✔
360	System.arraycopy(elements, 0, result, n, size - n);	1✔
361	}
362
363	if (result.length > size) {	1✔
364	for (int i = Math.max(0, size - 1); i < result.length; i++) {	1✔
365	result[i] = null;	1✔
366	}
367	}
368
369	return result;	1✔
370	}
371	finally {
372	lock.unlock();	1✔
373	}
374	}
375
376	public String toString() {
377	final ReentrantLock lock = this.lock;	1✔
378	lock.lock();	1✔
379	try {
380	if (size == 0) {	1✔
381	return "[]";	1✔
382	}
383
384	StringBuilder sb = new StringBuilder();	1✔
385	sb.append('[');	1✔
386
387	int idx = read;	1✔
388	while (true) {
389	E e = elements[idx];	1✔
390	sb.append(e == this ? "(this Collection)" : e);	1✔
391
392	idx = (idx + 1) % maxElements;	1✔
393
394	if (idx == write) {	1✔
395	return sb.append(']').toString();	1✔
396	} else {
397	sb.append(',').append(' ');	1✔
398	}
399	}	1✔
400	}
401	finally {
402	lock.unlock();	1✔
403	}
404	}
405
406	/* Internal implementations: MUST BE USED INSIDE LOCKS */
407
408	private int increment(int i) {
409	return (i + 1) % maxElements;	1✔
410	}
411
412	private int decrement(int i) {
413	return ((i == 0) ? maxElements : i) - 1;	1✔
414	}
415
416	private void internalAdd(E e) {
417	if (size == maxElements) {	1✔
418	internalRemove();	1✔
419	}
420
421	size++;	1✔
422	elements[write] = e;	1✔
423	write = increment(write);	1✔
424	}	1✔
425
426	private boolean internalContains(Object o) {
427	if (size > 0) {	1✔
428	int idx = read;	1✔
429	do {
430	if (o.equals(elements[idx])) {	1✔
431	return true;	1✔
432	}
433
434	idx = (idx + 1) % maxElements;	1✔
435	} while (idx != write);	1✔
436	}
437
438	return false;	1✔
439	}
440
441	private E internalRemove() {
442	final E element = elements[read];	1✔
443
444	if (null != element) {	1✔
445	internalRemoveAtIndex(read);	1✔
446	}
447
448	return element;	1✔
449	}
450
451	private void internalRemoveAtIndex(int idx) {
452	if (idx == read) {	1✔
453	elements[read] = null;	1✔
454
455	read = increment(read);	1✔
456
457	size--;	1✔
458	if (iterators != null) {	1✔
459	iterators.elementRemoved();	1✔
460	}
461	} else {
462	int i = idx;	1✔
463	while (true) {
464	int next = increment(i);	1✔
465
466	if (next != write) {	1✔
467	elements[i] = elements[next];	1✔
468	i = next;	1✔
469	} else {
470	elements[i] = null;	1✔
471	write = i;	1✔
472	break;	1✔
473	}
474	}	1✔
475
476	size--;	1✔
477	if (iterators != null) {	1✔
478	iterators.removedAt(idx);	1✔
479	}
480	}
481	}	1✔
482
483	private void readObject(final ObjectInputStream in) throws IOException, ClassNotFoundException {
484	in.defaultReadObject();	×
485	lock = new ReentrantLock();	×
486	}	×
487
488	/**
489	* A linked list maintaining references between the queue and any iterators
490	*
491	* This class exists to ensure that iterator objects are properly updated when items are removed from the queue and
492	* are invalidated if the underlying queue becomes incompatible with the iterator's view of it.
493	*
494	* This is based on the implementation of ArrayBlockingQueue.Itrs and involves the same garbage collection scheme
495	* described there.
496	*/
497	private class Iterators {
498
499	private static final int SHORT_SWEEP_PROBES = 4;
500
501	private static final int LONG_SWEEP_PROBES = 16;
502
503	private Node head;
504
505	private Node sweeper = null;	1✔
506
507	int cycles = 0;	1✔
508
509	Iterators(Iterator iterator) {	1✔
510	register(iterator);	1✔
511	}	1✔
512
513	void register(Iterator iterator) {
514	head = new Node(iterator, head);	1✔
515	}	1✔
516
517	void elementRemoved() {
518	if (size == 0) {	1✔
519	queueEmptied();	1✔
520	} else if (read == 0) {	1✔
521	readWrapped();	1✔
522	}
523	}	1✔
524
525	void queueEmptied() {
526	for (Node p = head; p != null; p = p.next) {	1✔
527	Iterator it = p.get();	1✔
528	if (it != null) {	1✔
529	p.clear();	1✔
530	it.shutdown();	1✔
531	}
532	}
533
534	head = null;	1✔
535	iterators = null;	1✔
536	}	1✔
537
538	void removedAt(int removedIndex) {
539	prune(it -> it.removedAt(removedIndex));	1✔
540	}	1✔
541
542	void readWrapped() {
543	cycles++;	1✔
544	prune(Iterator::readWrapped);	1✔
545	}	1✔
546
547	void sweep(boolean tryHarder) {
548	int probes = tryHarder ? LONG_SWEEP_PROBES : SHORT_SWEEP_PROBES;	1✔
549	Node o, p;
550	final Node sweeper = this.sweeper;	1✔
551	boolean completeCycle; // to limit search to one full sweep
552
553	if (sweeper == null) {	1✔
554	o = null;	1✔
555	p = head;	1✔
556	completeCycle = true;	1✔
557	} else {
558	o = sweeper;	1✔
559	p = o.next;	1✔
560	completeCycle = false;	1✔
561	}
562
563	for (; probes > 0; probes--) {	1✔
564	if (p == null) {	1✔
565	if (completeCycle) {	1✔
566	break;	1✔
567	}
568
569	o = null;	1✔
570	p = head;	1✔
571	completeCycle = true;	1✔
572	}
573
574	final Iterator it = p.get();	1✔
575	final Node next = p.next;	1✔
576	if (it == null \|\| it.isDetached()) {	1✔
577	// found a discarded/exhausted iterator
578	probes = LONG_SWEEP_PROBES; // "try harder"	1✔
579	// unlink p
580	p.clear();	1✔
581	p.next = null;	1✔
582	if (o == null) {	1✔
583	head = next;	1✔
584	if (next == null) {	1✔
585	// We've run out of iterators to track; retire
586	iterators = null;	1✔
587	return;	1✔
588	}
589	} else {
590	o.next = next;	1✔
591	}
592	} else {
593	o = p;	1✔
594	}
595	p = next;	1✔
596	}
597
598	this.sweeper = (p == null) ? null : o;	1✔
599	}	1✔
600
601	private void prune(Predicate<Iterator> shouldRemove) {
602	for (Node o = null, p = head; p != null; ) {	1✔
603	final Iterator it = p.get();	1✔
604	final Node next = p.next;	1✔
605
606	if (it == null \|\| shouldRemove.test(it)) {	1✔
607	p.clear();	1✔
608	p.next = null;	1✔
609
610	if (o == null) {	1✔
611	head = next;	1✔
612	} else {
613	o.next = next;	×
614	}
615	} else {
616	o = p;	1✔
617	}
618
619	p = next;	1✔
620	}	1✔
621
622	if (head == null) {	1✔
623	ThreadSafeCircularFifoQueue.this.iterators = null;	1✔
624	}
625	}	1✔
626
627	/**
628	* The actual list node implementation
629	*/
630	private class Node extends WeakReference<Iterator> {
631
632	Node next;
633
634	Node(Iterator iterator, Node next) {	1✔
635	super(iterator);	1✔
636	this.next = next;	1✔
637	}	1✔
638	}
639	}
640
641	/**
642	* An attempt to be a straight-forward iterator implementation for a ThreadSafeCircularFifoQueue.
643	*
644	* It should iterate over each member in the queue only once, assuming that the queue is not modified while this
645	* iterator remains in use. If the underlying queue is modified, the iterator will attempt to recover and keep going.
646	*
647	* If it becomes too far out of synch with the underlying data, an iterator may fail before iterating over all elements
648	* in a queue. However if {@link #hasNext()} returns true, {@link #next()} will always return a result.
649	*/
650	private class Iterator implements java.util.Iterator<E> {
651
652	/* Special index values */
653
654	// indicates an index that is invalid or empty
655	private static final int NONE = -1;
656
657	// used as a value for prevRead when the iterator is no longer valid
658	private static final int DETACHED = -2;
659
660	private int nextIndex;
661
662	private E nextItem;
663
664	private int prevIndex = NONE;	1✔
665
666	private E prevItem = null;	1✔
667
668	private int prevRead;
669
670	private int prevCycles;
671
672	Iterator() {	1✔
673	final ReentrantLock lock = ThreadSafeCircularFifoQueue.this.lock;	1✔
674	lock.lock();	1✔
675	try {
676	if (size == 0) {	1✔
677	nextIndex = NONE;	1✔
678	prevRead = DETACHED;	1✔
679	} else {
680	nextItem = elements[read];	1✔
681	nextIndex = read;	1✔
682	prevRead = read;	1✔
683
684	if (iterators == null) {	1✔
685	iterators = new Iterators(this);	1✔
686	} else {
687	iterators.register(this);	1✔
688	iterators.sweep(false);	1✔
689	}
690
691	prevCycles = iterators.cycles;	1✔
692	}
693	}
694	finally {
695	lock.unlock();	1✔
696	}
697	}	1✔
698
699	@Override
700	public boolean hasNext() {
701	// no locks for the simplest case
702	if (nextItem != null) {	1✔
703	return true;	1✔
704	}
705
706	final ReentrantLock lock = ThreadSafeCircularFifoQueue.this.lock;	1✔
707	lock.lock();	1✔
708	try {
709	if (!isDetached()) {	1✔
710	updateIndices();	1✔
711	if (prevIndex >= 0) {	1✔
712	prevItem = elements[prevIndex];	1✔
713	detach();	1✔
714	}
715	}
716	}
717	finally {
718	lock.unlock();	1✔
719	}
720
721	return false;	1✔
722	}
723
724	@Override
725	public E next() {
726	final E it = nextItem;	1✔
727	if (it == null) {	1✔
728	throw new NoSuchElementException();	1✔
729	}
730
731	final ReentrantLock lock = ThreadSafeCircularFifoQueue.this.lock;	1✔
732	lock.lock();	1✔
733	try {
734	if (!isDetached()) {	1✔
735	updateIndices();	1✔
736	}
737
738	prevIndex = nextIndex;	1✔
739	prevItem = it;	1✔
740
741	if (nextIndex < 0) {	1✔
742	nextIndex = NONE;	1✔
743	nextItem = null;	1✔
744	} else {
745	nextIndex = increment(nextIndex);	1✔
746	if (nextIndex == write) {	1✔
747	nextIndex = NONE;	1✔
748	nextItem = null;	1✔
749	} else {
750	nextItem = elements[nextIndex];	1✔
751	}
752	}
753
754	return it;	1✔
755	}
756	finally {
757	lock.unlock();	1✔
758	}
759	}
760
761	@Override
762	public void remove() {
763	final ReentrantLock lock = ThreadSafeCircularFifoQueue.this.lock;	1✔
764	lock.lock();	1✔
765	try {
766	if (!isDetached()) {	1✔
767	updateIndices();	1✔
768	}
769
770	if (prevIndex == NONE) {	1✔
771	throw new IllegalStateException();	1✔
772	} else if (prevIndex >= 0 && elements[prevIndex] == prevItem) {	1✔
773	internalRemoveAtIndex(prevIndex);	1✔
774
775	if (prevIndex != read) {	1✔
776	nextIndex = Math.max(prevIndex, read);	1✔
777	}
778	}
779
780	prevIndex = NONE;	1✔
781	prevItem = null;	1✔
782
783	if (nextIndex < 0) {	1✔
784	detach();	×
785	}
786	}
787	finally {
788	lock.unlock();	1✔
789	}
790	}	1✔
791
792	boolean readWrapped() {
793	if (isDetached()) {	1✔
794	return true;	×
795	}
796
797	if (iterators.cycles - prevCycles > 1) {	1✔
798	shutdown();	1✔
799	return true;	1✔
800	}
801
802	return false;	1✔
803	}
804
805	boolean removedAt(int removedIndex) {
806	if (isDetached()) {	1✔
807	return true;	×
808	}
809
810	final int cycles = iterators.cycles;	1✔
811	final int read = ThreadSafeCircularFifoQueue.this.read;	1✔
812
813	int cycleDiff = cycles - prevCycles;	1✔
814
815	if (removedIndex < read) {	1✔
816	cycleDiff++;	×
817	}
818
819	final int removedDistance = (cycleDiff * maxElements) + (removedIndex - prevRead);	1✔
820
821	if (prevIndex >= 0) {	1✔
822	int x = distance(prevIndex);	1✔
823	if (x == removedDistance) {	1✔
824	prevIndex = NONE;	1✔
825	} else if (x > removedDistance) {	×
826	prevIndex = decrement(prevIndex);	×
827	}
828	}
829
830	if (nextIndex >= 0) {	1✔
831	int x = distance(nextIndex);	1✔
832	if (x == removedDistance) {	1✔
833	nextIndex = NONE;	×
834	} else if (x > removedDistance) {	1✔
835	nextIndex = decrement(nextIndex);	1✔
836	}
837	} else if (prevIndex < 0) {	1✔
838	// don't call detach() as returning true will trigger a full sweep anyways
839	prevRead = DETACHED;	1✔
840	return true;	1✔
841	}
842
843	return false;	1✔
844	}
845
846	void shutdown() {
847	// nextItem is not set to null as it has been cached and can be returned
848	nextIndex = nextIndex >= 0 ? NONE : nextIndex;	1✔
849	prevIndex = prevIndex >= 0 ? NONE : nextIndex;	1✔
850	prevItem = null;	1✔
851	prevRead = DETACHED;	1✔
852	}	1✔
853
854	/**
855	* Detach the iterator and trigger a sweep of the iterator queue
856	*/
857	private void detach() {
858	if (prevRead >= 0) {	1✔
859	prevRead = DETACHED;	1✔
860	iterators.sweep(true);	1✔
861	}
862	}	1✔
863
864	private boolean isDetached() {
865	return prevRead < 0;	1✔
866	}
867
868	private int distance(int index) {
869	int distance = index - prevRead;	1✔
870	if (distance < 0) {	1✔
871	distance += maxElements;	1✔
872	}
873	return distance;	1✔
874	}
875
876	private boolean indexInvalidated(int index, long dequeues) {
877	if (index < 0) {	1✔
878	return false;	1✔
879	}
880
881	int distance = distance(index);	1✔
882
883	return dequeues > distance;	1✔
884	}
885
886	private void updateIndices() {
887	final int cycles = ThreadSafeCircularFifoQueue.this.iterators.cycles;	1✔
888	if (cycles != prevCycles \|\| read != prevRead) {	1✔
889	long dequeues = (cycles - prevCycles) * maxElements + (read - prevRead);	1✔
890
891	if (indexInvalidated(prevIndex, dequeues)) {	1✔
892	prevIndex = NONE;	1✔
893	}
894
895	if (indexInvalidated(nextIndex, dequeues)) {	1✔
896	nextIndex = NONE;	1✔
897	}
898
899	if (prevIndex < 0 && nextIndex < 0) {	1✔
900	detach();	1✔
901	} else {
902	prevCycles = cycles;	1✔
903	prevRead = read;	1✔
904	}
905	}
906	}	1✔
907	}
908	}

openmrs / openmrs-core / 31024361752

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous