Fork SpscLinkedQueue and SpscLinkedAtomicQueue from JCTools

Motivation:

See netty#3746.

Modifications:

Fork SpscLinkedQueue and SpscLinkedAtomicQueue from JCTools based on https://github.com/JCTools/JCTools/tree/7846450e2817d6eb9653111e3b7a85b76fbbcfd1

Result:

Add SpscLinkedQueue and SpscLinkedAtomicQueue and apply it in LocalChannel.

common/src/main/java/io/netty/util/internal/BaseLinkedAtomicQueue.java

@@ -0,0 +1,110 @@
+/*
+ * Copyright 2016 The Netty Project
+ *
+ * The Netty Project licenses this file to you 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.
+ */
+/*
+ * 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.
+ */
+package io.netty.util.internal;
+
+import java.util.AbstractQueue;
+import java.util.Iterator;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Forked from <a href="https://github.com/JCTools/JCTools">JCTools</a>.
+ */
+abstract class BaseLinkedAtomicQueue<E> extends AbstractQueue<E> {
+    private final AtomicReference<LinkedQueueAtomicNode<E>> producerNode;
+    private final AtomicReference<LinkedQueueAtomicNode<E>> consumerNode;
+    public BaseLinkedAtomicQueue() {
+        producerNode = new AtomicReference<LinkedQueueAtomicNode<E>>();
+        consumerNode = new AtomicReference<LinkedQueueAtomicNode<E>>();
+    }
+    protected final LinkedQueueAtomicNode<E> lvProducerNode() {
+        return producerNode.get();
+    }
+    protected final LinkedQueueAtomicNode<E> lpProducerNode() {
+        return producerNode.get();
+    }
+    protected final void spProducerNode(LinkedQueueAtomicNode<E> node) {
+        producerNode.lazySet(node);
+    }
+    protected final LinkedQueueAtomicNode<E> xchgProducerNode(LinkedQueueAtomicNode<E> node) {
+        return producerNode.getAndSet(node);
+    }
+    protected final LinkedQueueAtomicNode<E> lvConsumerNode() {
+        return consumerNode.get();
+    }
+
+    protected final LinkedQueueAtomicNode<E> lpConsumerNode() {
+        return consumerNode.get();
+    }
+    protected final void spConsumerNode(LinkedQueueAtomicNode<E> node) {
+        consumerNode.lazySet(node);
+    }
+    @Override
+    public final Iterator<E> iterator() {
+        throw new UnsupportedOperationException();
+    }
+
+    /**
+     * {@inheritDoc} <br>
+     * <p>
+     * IMPLEMENTATION NOTES:<br>
+     * This is an O(n) operation as we run through all the nodes and count them.<br>
+     *
+     * @see java.util.Queue#size()
+     */
+    @Override
+    public final int size() {
+        LinkedQueueAtomicNode<E> chaserNode = lvConsumerNode();
+        final LinkedQueueAtomicNode<E> producerNode = lvProducerNode();
+        int size = 0;
+        // must chase the nodes all the way to the producer node, but there's no need to chase a moving target.
+        while (chaserNode != producerNode && size < Integer.MAX_VALUE) {
+            LinkedQueueAtomicNode<E> next;
+            while ((next = chaserNode.lvNext()) == null) {
+                continue;
+            }
+            chaserNode = next;
+            size++;
+        }
+        return size;
+    }
+    /**
+     * {@inheritDoc} <br>
+     * <p>
+     * IMPLEMENTATION NOTES:<br>
+     * Queue is empty when producerNode is the same as consumerNode. An alternative implementation would be to observe
+     * the producerNode.value is null, which also means an empty queue because only the consumerNode.value is allowed to
+     * be null.
+     *
+     * @see MessagePassingQueue#isEmpty()
+     */
+    @Override
+    public final boolean isEmpty() {
+        return lvConsumerNode() == lvProducerNode();
+    }
+}

common/src/main/java/io/netty/util/internal/BaseLinkedQueue.java

@@ -0,0 +1,158 @@
+/*
+ * Copyright 2016 The Netty Project
+ *
+ * The Netty Project licenses this file to you 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.
+ */
+/*
+ * 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.
+ */
+package io.netty.util.internal;
+
+import java.util.AbstractQueue;
+import java.util.Iterator;
+
+/**
+ * Forked from <a href="https://github.com/JCTools/JCTools">JCTools</a>.
+ *
+ * A base data structure for concurrent linked queues.
+ *
+ * @param <E>
+ */
+abstract class BaseLinkedQueue<E> extends BaseLinkedQueueConsumerNodeRef<E> {
+    long p01, p02, p03, p04, p05, p06, p07;
+    long p10, p11, p12, p13, p14, p15, p16, p17;
+
+    @Override
+    public final Iterator<E> iterator() {
+        throw new UnsupportedOperationException();
+    }
+
+    /**
+     * {@inheritDoc} <br>
+     * <p>
+     * IMPLEMENTATION NOTES:<br>
+     * This is an O(n) operation as we run through all the nodes and count them.<br>
+     *
+     * @see java.util.Queue#size()
+     */
+    @Override
+    public final int size() {
+        // Read consumer first, this is important because if the producer is node is 'older' than the consumer the
+        // consumer may overtake it (consume past it). This will lead to an infinite loop below.
+        LinkedQueueNode<E> chaserNode = lvConsumerNode();
+        final LinkedQueueNode<E> producerNode = lvProducerNode();
+        int size = 0;
+        // must chase the nodes all the way to the producer node, but there's no need to chase a moving target.
+        while (chaserNode != producerNode && size < Integer.MAX_VALUE) {
+            LinkedQueueNode<E> next;
+            while ((next = chaserNode.lvNext()) == null) {
+                continue;
+            }
+            chaserNode = next;
+            size++;
+        }
+        return size;
+    }
+
+    /**
+     * {@inheritDoc} <br>
+     * <p>
+     * IMPLEMENTATION NOTES:<br>
+     * Queue is empty when producerNode is the same as consumerNode. An alternative implementation would be to observe
+     * the producerNode.value is null, which also means an empty queue because only the consumerNode.value is allowed to
+     * be null.
+     *
+     * @see MessagePassingQueue#isEmpty()
+     */
+    @Override
+    public final boolean isEmpty() {
+        return lvConsumerNode() == lvProducerNode();
+    }
+
+    @Override
+    public int capacity() {
+        return UNBOUNDED_CAPACITY;
+    }
+}
+
+abstract class BaseLinkedQueuePad0<E> extends AbstractQueue<E> implements MessagePassingQueue<E> {
+    long p00, p01, p02, p03, p04, p05, p06, p07;
+    long p10, p11, p12, p13, p14, p15, p16;
+}
+
+abstract class BaseLinkedQueueProducerNodeRef<E> extends BaseLinkedQueuePad0<E> {
+    protected static final long P_NODE_OFFSET;
+
+    static {
+        try {
+            P_NODE_OFFSET = PlatformDependent0.objectFieldOffset(
+                    BaseLinkedQueueProducerNodeRef.class.getDeclaredField("producerNode"));
+        } catch (NoSuchFieldException e) {
+            throw new RuntimeException(e);
+        }
+    }
+    protected LinkedQueueNode<E> producerNode;
+    protected final void spProducerNode(LinkedQueueNode<E> node) {
+        producerNode = node;
+    }
+
+    @SuppressWarnings("unchecked")
+    protected final LinkedQueueNode<E> lvProducerNode() {
+        return (LinkedQueueNode<E>) PlatformDependent0.getObjectVolatile(this, P_NODE_OFFSET);
+    }
+
+    protected final LinkedQueueNode<E> lpProducerNode() {
+        return producerNode;
+    }
+}
+
+abstract class BaseLinkedQueuePad1<E> extends BaseLinkedQueueProducerNodeRef<E> {
+    long p01, p02, p03, p04, p05, p06, p07;
+    long p10, p11, p12, p13, p14, p15, p16, p17;
+}
+
+abstract class BaseLinkedQueueConsumerNodeRef<E> extends BaseLinkedQueuePad1<E> {
+    protected static final long C_NODE_OFFSET;
+
+    static {
+        try {
+            C_NODE_OFFSET = PlatformDependent0.objectFieldOffset(
+                    BaseLinkedQueueConsumerNodeRef.class.getDeclaredField("consumerNode"));
+        } catch (NoSuchFieldException e) {
+            throw new RuntimeException(e);
+        }
+    }
+    protected LinkedQueueNode<E> consumerNode;
+    protected final void spConsumerNode(LinkedQueueNode<E> node) {
+        consumerNode = node;
+    }
+
+    @SuppressWarnings("unchecked")
+    protected final LinkedQueueNode<E> lvConsumerNode() {
+        return (LinkedQueueNode<E>) PlatformDependent0.getObjectVolatile(this, C_NODE_OFFSET);
+    }
+
+    protected final LinkedQueueNode<E> lpConsumerNode() {
+        return consumerNode;
+    }
+}

common/src/main/java/io/netty/util/internal/LinkedQueueAtomicNode.java

@@ -0,0 +1,71 @@
+/*
+ * Copyright 2016 The Netty Project
+ *
+ * The Netty Project licenses this file to you 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.
+ */
+/*
+ * 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.
+ */
+package io.netty.util.internal;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Forked from <a href="https://github.com/JCTools/JCTools">JCTools</a>.
+ */
+public final class LinkedQueueAtomicNode<E> extends AtomicReference<LinkedQueueAtomicNode<E>> {
+    /** */
+    private static final long serialVersionUID = 2404266111789071508L;
+    private E value;
+    LinkedQueueAtomicNode() {
+    }
+    LinkedQueueAtomicNode(E val) {
+        spValue(val);
+    }
+    /**
+     * Gets the current value and nulls out the reference to it from this node.
+     *
+     * @return value
+     */
+    public E getAndNullValue() {
+        E temp = lpValue();
+        spValue(null);
+        return temp;
+    }
+
+    public E lpValue() {
+        return value;
+    }
+
+    public void spValue(E newValue) {
+        value = newValue;
+    }
+
+    public void soNext(LinkedQueueAtomicNode<E> n) {
+        lazySet(n);
+    }
+
+    public LinkedQueueAtomicNode<E> lvNext() {
+        return get();
+    }
+}