24249879133

Committed 10 Apr 2026 03:12PM UTC coverage: 65.248% (+0.01%) from 65.236%

Build # 24249879133

Build Type

push

github

Committed by

openmrs-bot

Commit Message

Setting new SNAPSHOT version

Coverage Stats

23617 of 36196 relevant lines covered (65.25%)

1.3 hits per line

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

95.04

/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;
        }

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

openmrs / openmrs-core / 24249879133

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