Fixed padding of the integer and node classes.

rxjava-core/src/main/java/rx/internal/util/FrontPadding.java

@@ -0,0 +1,31 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ *
+ * 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 rx.internal.util;
+
+import java.io.Serializable;
+
+/**
+ * Padding up to 128 bytes at the front.
+ * Based on netty's implementation
+ */
+abstract class FrontPadding implements Serializable {
+    /** */
+    private static final long serialVersionUID = -596356687591714352L;
+    /** Padding. */
+    public transient long p1, p2, p3, p4, p5, p6;           // 48 bytes (header is 16 bytes)
+    /** Padding. */
+    public transient long p8, p9, p10, p11, p12, p13, p14, p15; // 64 bytes
+}

rxjava-core/src/main/java/rx/internal/util/MpscPaddedQueue.java

@@ -15,6 +15,7 @@
  */
 package rx.internal.util;
 
+import java.io.Serializable;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 
@@ -26,8 +27,6 @@
  * @param <E> the element type
  */
 public final class MpscPaddedQueue<E> extends AtomicReference<MpscPaddedQueue.Node<E>> {
-    @SuppressWarnings(value = "rawtypes")
-    static final AtomicReferenceFieldUpdater<PaddedNode, Node> TAIL_UPDATER = AtomicReferenceFieldUpdater.newUpdater(PaddedNode.class, Node.class, "tail");
     /** */
     private static final long serialVersionUID = 1L;
     /** The padded tail reference. */
@@ -39,7 +38,7 @@ public final class MpscPaddedQueue<E> extends AtomicReference<MpscPaddedQueue.No
     public MpscPaddedQueue() {
         Node<E> first = new Node<E>(null);
         tail = new PaddedNode<E>();
-        tail.tail = first;
+        tail.node = first;
         set(first);
     }
 
@@ -64,7 +63,7 @@ public E poll() {
         }
         E v = n.value;
         n.value = null; // do not retain this value as the node still stays in the queue
-        TAIL_UPDATER.lazySet(tail, n);
+        tail.lazySet(n);
         return v;
     }
 
@@ -75,7 +74,7 @@ public E poll() {
     private Node<E> peekNode() {
         for (;;) {
             @SuppressWarnings(value = "unchecked")
-            Node<E> t = TAIL_UPDATER.get(tail);
+            Node<E> t = tail.node;
             Node<E> n = t.get();
             if (n != null || get() == t) {
                 return n;
@@ -93,29 +92,30 @@ public void clear() {
             }
         }
     }
-
-    /** Class that contains a Node reference padded around to fit a typical cache line. */
-    static final class PaddedNode<E> {
-        /** Padding, public to prevent optimizing it away. */
-        public int p1;
-        volatile Node<E> tail;
-        /** Padding, public to prevent optimizing it away. */
-        public long p2;
-        /** Padding, public to prevent optimizing it away. */
-        public long p3;
-        /** Padding, public to prevent optimizing it away. */
-        public long p4;
-        /** Padding, public to prevent optimizing it away. */
-        public long p5;
-        /** Padding, public to prevent optimizing it away. */
-        public long p6;
+    /** The front-padded node class housing the actual value. */
+    static abstract class PaddedNodeBase<E> extends FrontPadding {
+        private static final long serialVersionUID = 2L;
+        volatile Node<E> node;
+        @SuppressWarnings(value = "rawtypes")
+        static final AtomicReferenceFieldUpdater<PaddedNodeBase, Node> NODE_UPDATER = AtomicReferenceFieldUpdater.newUpdater(PaddedNodeBase.class, Node.class, "node");
+        public void lazySet(Node<E> newValue) {
+            NODE_UPDATER.lazySet(this, newValue);
+        }
+    }
+    /** Post-padding of the padded node base class.  */
+    static final class PaddedNode<E> extends PaddedNodeBase<E> {
+        private static final long serialVersionUID = 3L;
+        /** Padding. */
+        public transient long p16, p17, p18, p19, p20, p21, p22;      // 56 bytes (the remaining 8 is in the base)
+        /** Padding. */
+        public transient long p24, p25, p26, p27, p28, p29, p30, p31; // 64 bytes
     }
 
     /**
      * Regular node with value and reference to the next node.
      */
-    static final class Node<E> {
-
+    static final class Node<E> implements Serializable {
+        private static final long serialVersionUID = 4L;
         E value;
         @SuppressWarnings(value = "rawtypes")
         static final AtomicReferenceFieldUpdater<Node, Node> TAIL_UPDATER = AtomicReferenceFieldUpdater.newUpdater(Node.class, Node.class, "tail");

rxjava-core/src/main/java/rx/internal/util/PaddedAtomicInteger.java

@@ -15,45 +15,16 @@
  */
 package rx.internal.util;
 
-import java.util.concurrent.atomic.AtomicInteger;
-
 /**
- * An AtomicInteger with extra fields to pad it out to fit a typical cache line.
+ * A padded atomic integer to fill in 4 cache lines to avoid any false sharing or
+ * adjacent prefetch.
+ * Based on Netty's implementation.
  */
-public final class PaddedAtomicInteger extends AtomicInteger {
-    private static final long serialVersionUID = 1L;
-    /** Padding, public to prevent optimizing it away. */
-    public int p1;
-    /** Padding, public to prevent optimizing it away. */
-    public int p2;
-    /** Padding, public to prevent optimizing it away. */
-    public int p3;
-    /** Padding, public to prevent optimizing it away. */
-    public int p4;
-    /** Padding, public to prevent optimizing it away. */
-    public int p5;
-    /** Padding, public to prevent optimizing it away. */
-    public int p6;
-    /** Padding, public to prevent optimizing it away. */
-    public int p7;
-    /** Padding, public to prevent optimizing it away. */
-    public int p8;
-    /** Padding, public to prevent optimizing it away. */
-    public int p9;
-    /** Padding, public to prevent optimizing it away. */
-    public int p10;
-    /** Padding, public to prevent optimizing it away. */
-    public int p11;
-    /** Padding, public to prevent optimizing it away. */
-    public int p12;
-    /** Padding, public to prevent optimizing it away. */
-    public int p13;
-    /**
-     * @warn description missing
-     * @return prevents optimizing away the fields, most likely.
-     */
-    public int noopt() {
-        return p1 + p2 + p3 + p4 + p5 + p6 + p7 + p8 + p9 + p10 + p11 + p12 + p13;
-    }
-
+public final class PaddedAtomicInteger extends PaddedAtomicIntegerBase {
+    /** */
+    private static final long serialVersionUID = 8781891581317286855L;
+    /** Padding. */
+    public transient long p16, p17, p18, p19, p20, p21, p22;      // 56 bytes (the remaining 8 is in the base)
+    /** Padding. */
+    public transient long p24, p25, p26, p27, p28, p29, p30, p31; // 64 bytes
 }

rxjava-core/src/main/java/rx/internal/util/PaddedAtomicIntegerBase.java

@@ -0,0 +1,84 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ *
+ * 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 rx.internal.util;
+
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+
+/**
+ * The atomic integer base padded at the front.
+ * Based on Netty's implementation.
+ */
+abstract class PaddedAtomicIntegerBase extends FrontPadding {
+
+    private static final long serialVersionUID = 6513142711280243198L;
+
+    private static final AtomicIntegerFieldUpdater<PaddedAtomicIntegerBase> updater;
+
+    static {
+        updater = AtomicIntegerFieldUpdater.newUpdater(PaddedAtomicIntegerBase.class, "value");
+    }
+
+    private volatile int value; // 8-byte object field (or 4-byte + padding)
+
+    public final int get() {
+        return value;
+    }
+
+    public final void set(int newValue) {
+        this.value = newValue;
+    }
+
+    public final void lazySet(int newValue) {
+        updater.lazySet(this, newValue);
+    }
+
+    public final boolean compareAndSet(int expect, int update) {
+        return updater.compareAndSet(this, expect, update);
+    }
+
+    public final boolean weakCompareAndSet(int expect, int update) {
+        return updater.weakCompareAndSet(this, expect, update);
+    }
+
+    public final int getAndSet(int newValue) {
+        return updater.getAndSet(this, value);
+    }
+
+    public final int getAndAdd(int delta) {
+    	return updater.getAndAdd(this, delta);
+    }
+    public final int incrementAndGet() {
+    	return updater.incrementAndGet(this);
+    }
+    public final int decrementAndGet() {
+    	return updater.decrementAndGet(this);
+    }
+    public final int getAndIncrement() {
+    	return updater.getAndIncrement(this);
+    }
+    public final int getAndDecrement() {
+    	return updater.getAndDecrement(this);
+    }
+    public final int addAndGet(int delta) {
+    	return updater.addAndGet(this, delta);
+    }
+
+    @Override
+    public String toString() {
+        return String.valueOf(get());
+    }
+}