22781579539

Committed 06 Mar 2026 08:53PM UTC coverage: 65.392%. Remained the same

Build # 22781579539

Build Type

push

github

Committed by

ibacher

Commit Message

Fix the Mockito hack

Previously, the way we were controlling mockito resulted in an inconsistent
dependency tree. This fixes the profile so we only need one and updates
the version property instead of the raw dependency.

Run Details

23842 of 36460 relevant lines covered (65.39%)

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

openmrs / openmrs-core / 22781579539

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