27011900216

Committed 05 Jun 2026 11:20AM UTC coverage: 63.664% (-0.01%) from 63.677%

Build # 27011900216

Build Type

push

github

Committed by

web-flow

Commit Message

TRUNK-6558 Return null from getTaskByName when the task does not exist (#6162)

JobRunrSchedulerService.getTaskByName propagated the DAO's
ObjectRetrievalFailureException when no task matched, breaking the
long-standing SchedulerService contract (the pre-JobRunr
TimerSchedulerServiceImpl caught the exception and returned null).
Modules that probe for their task before registering it - e.g.
schedulerService.getTaskByName(NAME) == null - failed to start on any
fresh database.

Restore the historical behavior: catch ObjectRetrievalFailureException,
log a warning, and return null.

Coverage Stats

0 of 4 new or added lines in 1 file covered. (0.0%)

4 existing lines in 2 files now uncovered.

23827 of 37426 relevant lines covered (63.66%)

0.64 hits per line

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

94.27

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

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

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

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

        private static final long serialVersionUID = -89162358098721L;

        // queue capacity
        private final int maxElements;

        // Underlying storage
        private final E[] elements;

        private transient ReentrantLock lock = new ReentrantLock();

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

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

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

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

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

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

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

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

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

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

                return true;
        }

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

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

                return true;
        }

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

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

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

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

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

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

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

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

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

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

        }

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

                                idx = (idx + 1) % maxElements;

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

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

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

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

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

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

                return false;
        }

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

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

                return element;
        }

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

                        read = increment(read);

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

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

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

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

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

                private static final int SHORT_SWEEP_PROBES = 4;

                private static final int LONG_SWEEP_PROBES = 16;

                private Node head;

                private Node sweeper = null;

                int cycles = 0;

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

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

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

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

                        head = null;
                        iterators = null;
                }

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

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

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

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

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

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

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

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

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

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

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

                                p = next;
                        }

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

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

                        Node next;

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

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

                /* Special index values */

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

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

                private int nextIndex;

                private E nextItem;

                private int prevIndex = NONE;

                private E prevItem = null;

                private int prevRead;

                private int prevCycles;

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

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

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

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

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

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

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

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

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

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

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

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

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

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

                        return false;
                }

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

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

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

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

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

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

                        return false;
                }

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

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

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

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

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

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

openmrs / openmrs-core / 27011900216

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