From c89ea3238b0e56b22a12f8a14a6e4e94c6be6a7c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?D=C3=A1vid=20Karnok?= Date: Mon, 9 Nov 2015 21:56:59 +0100 Subject: [PATCH] 1.x: eager concatMap to choose safe or unsafe queue based on platform. I forgot to add the choice because 2.x SpscArrayQueue doesn't use Unsafe. I copied the SpscAtomicArrayQueue from #3169 and I hope it won't conflict. --- .../operators/OperatorEagerConcatMap.java | 11 +- .../atomic/AtomicReferenceArrayQueue.java | 75 +++++++++++ .../util/atomic/SpscAtomicArrayQueue.java | 124 ++++++++++++++++++ 3 files changed, 208 insertions(+), 2 deletions(-) create mode 100644 src/main/java/rx/internal/util/atomic/AtomicReferenceArrayQueue.java create mode 100644 src/main/java/rx/internal/util/atomic/SpscAtomicArrayQueue.java diff --git a/src/main/java/rx/internal/operators/OperatorEagerConcatMap.java b/src/main/java/rx/internal/operators/OperatorEagerConcatMap.java index 127f2fbd51..4df115b7ae 100644 --- a/src/main/java/rx/internal/operators/OperatorEagerConcatMap.java +++ b/src/main/java/rx/internal/operators/OperatorEagerConcatMap.java @@ -24,7 +24,8 @@ import rx.Observable.Operator; import rx.exceptions.Exceptions; import rx.functions.*; -import rx.internal.util.unsafe.SpscArrayQueue; +import rx.internal.util.atomic.SpscAtomicArrayQueue; +import rx.internal.util.unsafe.*; import rx.subscriptions.Subscriptions; public final class OperatorEagerConcatMap implements Operator { @@ -278,7 +279,13 @@ static final class EagerInnerSubscriber extends Subscriber { public EagerInnerSubscriber(EagerOuterSubscriber parent, int bufferSize) { super(); this.parent = parent; - this.queue = new SpscArrayQueue(bufferSize); + Queue q; + if (UnsafeAccess.isUnsafeAvailable()) { + q = new SpscArrayQueue(bufferSize); + } else { + q = new SpscAtomicArrayQueue(bufferSize); + } + this.queue = q; request(bufferSize); } diff --git a/src/main/java/rx/internal/util/atomic/AtomicReferenceArrayQueue.java b/src/main/java/rx/internal/util/atomic/AtomicReferenceArrayQueue.java new file mode 100644 index 0000000000..f7594ba20a --- /dev/null +++ b/src/main/java/rx/internal/util/atomic/AtomicReferenceArrayQueue.java @@ -0,0 +1,75 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + * Original License: https://github.com/JCTools/JCTools/blob/master/LICENSE + * Original location: https://github.com/JCTools/JCTools/blob/master/jctools-core/src/main/java/org/jctools/queues/atomic/AtomicReferenceArrayQueue.java + */ +package rx.internal.util.atomic; + +import java.util.*; +import java.util.concurrent.atomic.AtomicReferenceArray; + +import rx.internal.util.unsafe.Pow2; + +abstract class AtomicReferenceArrayQueue extends AbstractQueue { + protected final AtomicReferenceArray buffer; + protected final int mask; + public AtomicReferenceArrayQueue(int capacity) { + int actualCapacity = Pow2.roundToPowerOfTwo(capacity); + this.mask = actualCapacity - 1; + this.buffer = new AtomicReferenceArray(actualCapacity); + } + @Override + public Iterator iterator() { + throw new UnsupportedOperationException(); + } + @Override + public void clear() { + // we have to test isEmpty because of the weaker poll() guarantee + while (poll() != null || !isEmpty()) + ; + } + protected final int calcElementOffset(long index, int mask) { + return (int)index & mask; + } + protected final int calcElementOffset(long index) { + return (int)index & mask; + } + protected final E lvElement(AtomicReferenceArray buffer, int offset) { + return buffer.get(offset); + } + protected final E lpElement(AtomicReferenceArray buffer, int offset) { + return buffer.get(offset); // no weaker form available + } + protected final E lpElement(int offset) { + return buffer.get(offset); // no weaker form available + } + protected final void spElement(AtomicReferenceArray buffer, int offset, E value) { + buffer.lazySet(offset, value); // no weaker form available + } + protected final void spElement(int offset, E value) { + buffer.lazySet(offset, value); // no weaker form available + } + protected final void soElement(AtomicReferenceArray buffer, int offset, E value) { + buffer.lazySet(offset, value); + } + protected final void soElement(int offset, E value) { + buffer.lazySet(offset, value); + } + protected final void svElement(AtomicReferenceArray buffer, int offset, E value) { + buffer.set(offset, value); + } + protected final E lvElement(int offset) { + return lvElement(buffer, offset); + } +} diff --git a/src/main/java/rx/internal/util/atomic/SpscAtomicArrayQueue.java b/src/main/java/rx/internal/util/atomic/SpscAtomicArrayQueue.java new file mode 100644 index 0000000000..65c29e3ce8 --- /dev/null +++ b/src/main/java/rx/internal/util/atomic/SpscAtomicArrayQueue.java @@ -0,0 +1,124 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + * Original License: https://github.com/JCTools/JCTools/blob/master/LICENSE + * Original location: https://github.com/JCTools/JCTools/blob/master/jctools-core/src/main/java/org/jctools/queues/atomic/SpscAtomicArrayQueue.java + */ +package rx.internal.util.atomic; + +import java.util.concurrent.atomic.*; + +/** + * A Single-Producer-Single-Consumer queue backed by a pre-allocated buffer. + *

+ * This implementation is a mashup of the Fast Flow + * algorithm with an optimization of the offer method taken from the BQueue algorithm (a variation on Fast + * Flow), and adjusted to comply with Queue.offer semantics with regards to capacity.
+ * For convenience the relevant papers are available in the resources folder:
+ * 2010 - Pisa - SPSC Queues on Shared Cache Multi-Core Systems.pdf
+ * 2012 - Junchang- BQueue- Efficient and Practical Queuing.pdf
+ * This implementation is wait free. + * + * @param + */ +public final class SpscAtomicArrayQueue extends AtomicReferenceArrayQueue { + private static final Integer MAX_LOOK_AHEAD_STEP = Integer.getInteger("jctools.spsc.max.lookahead.step", 4096); + final AtomicLong producerIndex; + protected long producerLookAhead; + final AtomicLong consumerIndex; + final int lookAheadStep; + public SpscAtomicArrayQueue(int capacity) { + super(capacity); + this.producerIndex = new AtomicLong(); + this.consumerIndex = new AtomicLong(); + lookAheadStep = Math.min(capacity / 4, MAX_LOOK_AHEAD_STEP); + } + + @Override + public boolean offer(E e) { + if (null == e) { + throw new NullPointerException("Null is not a valid element"); + } + // local load of field to avoid repeated loads after volatile reads + final AtomicReferenceArray buffer = this.buffer; + final int mask = this.mask; + final long index = producerIndex.get(); + final int offset = calcElementOffset(index, mask); + if (index >= producerLookAhead) { + int step = lookAheadStep; + if (null == lvElement(buffer, calcElementOffset(index + step, mask))) {// LoadLoad + producerLookAhead = index + step; + } + else if (null != lvElement(buffer, offset)){ + return false; + } + } + soProducerIndex(index + 1); // ordered store -> atomic and ordered for size() + soElement(buffer, offset, e); // StoreStore + return true; + } + + @Override + public E poll() { + final long index = consumerIndex.get(); + final int offset = calcElementOffset(index); + // local load of field to avoid repeated loads after volatile reads + final AtomicReferenceArray lElementBuffer = buffer; + final E e = lvElement(lElementBuffer, offset);// LoadLoad + if (null == e) { + return null; + } + soConsumerIndex(index + 1); // ordered store -> atomic and ordered for size() + soElement(lElementBuffer, offset, null);// StoreStore + return e; + } + + @Override + public E peek() { + return lvElement(calcElementOffset(consumerIndex.get())); + } + + @Override + public int size() { + /* + * It is possible for a thread to be interrupted or reschedule between the read of the producer and consumer + * indices, therefore protection is required to ensure size is within valid range. In the event of concurrent + * polls/offers to this method the size is OVER estimated as we read consumer index BEFORE the producer index. + */ + long after = lvConsumerIndex(); + while (true) { + final long before = after; + final long currentProducerIndex = lvProducerIndex(); + after = lvConsumerIndex(); + if (before == after) { + return (int) (currentProducerIndex - after); + } + } + } + + private void soProducerIndex(long newIndex) { + producerIndex.lazySet(newIndex); + } + + private void soConsumerIndex(long newIndex) { + consumerIndex.lazySet(newIndex); + } + + private long lvConsumerIndex() { + return consumerIndex.get(); + } + private long lvProducerIndex() { + return producerIndex.get(); + } +}