Changed lazySet to regular volatile write to avoid potential visibility issues.

Author: akarnokd

rxjava-core/src/main/java/rx/operators/BlockingOperatorMostRecent.java

@@ -16,7 +16,6 @@
 package rx.operators;
 
 import java.util.Iterator;
-import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 
 import rx.Observable;
 import rx.Subscriber;
@@ -80,27 +79,24 @@ public void remove() {
     private static class MostRecentObserver<T> extends Subscriber<T> {
         static final NotificationLite<Object> nl = NotificationLite.instance();
         volatile Object value;
-        @SuppressWarnings("rawtypes")
-        static final AtomicReferenceFieldUpdater<MostRecentObserver, Object> VALUE_UPDATER
-                = AtomicReferenceFieldUpdater.newUpdater(MostRecentObserver.class, Object.class, "value");
 
         private MostRecentObserver(T value) {
-            VALUE_UPDATER.lazySet(this, nl.next(value));
+            this.value = nl.next(value);
         }
 
         @Override
         public void onCompleted() {
-            VALUE_UPDATER.lazySet(this, nl.completed());
+            value = nl.completed();
         }
 
         @Override
         public void onError(Throwable e) {
-            VALUE_UPDATER.lazySet(this, nl.error(e));
+            value = nl.error(e);
         }
 
         @Override
         public void onNext(T args) {
-            VALUE_UPDATER.lazySet(this, nl.next(args));
+            value = nl.next(args);
         }
 
         private boolean isCompleted() {

rxjava-core/src/main/java/rx/operators/BlockingOperatorNext.java

@@ -178,7 +178,7 @@ public Notification<? extends T> takeNext() throws InterruptedException {
             return buf.take();
         }
         void setWaiting(int value) {
-            WAITING_UPDATER.lazySet(this, value);
+            waiting = value;
         }
     }
 }

rxjava-core/src/main/java/rx/operators/BufferUntilSubscriber.java

@@ -70,7 +70,7 @@ boolean casFirst(int expected, int next) {
             return FIRST_UPDATER.compareAndSet(this, expected, next);
         }
         void setObserverRef(Observer<? super T> o) {
-            OBSERVER_UPDATER.lazySet(this, o);
+            observerRef = o;
         }
         boolean casObserverRef(Observer<? super T> expected, Observer<? super T> next) {
             return OBSERVER_UPDATER.compareAndSet(this, expected, next);

rxjava-core/src/main/java/rx/operators/OperatorMerge.java

@@ -57,7 +57,7 @@ public MergeSubscriber(Subscriber<T> actual, CompositeSubscription childrenSubsc
             super(actual);
             this.actual = actual;
             this.childrenSubscriptions = childrenSubscriptions;
-            WIP_UPDATER.lazySet(this, 1);
+            this.wip = 1;
         }
 
         @Override

rxjava-core/src/main/java/rx/operators/OperatorMergeDelayError.java

@@ -67,7 +67,7 @@ public MergeDelayErrorSubscriber(Subscriber<? super T> s, CompositeSubscription
             super(s);
             this.s = s;
             this.csub = csub;
-            WIP_UPDATER.lazySet(this, 1);
+            this.wip = 1;
         }
 
         @Override

rxjava-core/src/main/java/rx/operators/OperatorMergeMapPair.java

@@ -79,7 +79,7 @@ public SourceSubscriber(Subscriber<R> s, CompositeSubscription csub,
             this.csub = csub;
             this.collectionSelector = collectionSelector;
             this.resultSelector = resultSelector;
-            WIP_UPDATER.lazySet(this, 1);
+            this.wip = 1;
         }
 
         @Override

rxjava-core/src/main/java/rx/operators/OperatorMergeMapTransform.java

@@ -71,7 +71,7 @@ public SourceSubscriber(Subscriber<R> s, CompositeSubscription csub, Func1<? sup
             this.onNext = onNext;
             this.onError = onError;
             this.onCompleted = onCompleted;
-            WIP_UPDATER.lazySet(this, 1);
+            this.wip = 1;
         }
 
         @Override

rxjava-core/src/main/java/rx/operators/OperatorMergeMaxConcurrent.java

@@ -72,7 +72,7 @@ public SourceSubscriber(int maxConcurrency, Subscriber<T> s, CompositeSubscripti
             this.csub = csub;
             this.guard = new Object();
             this.queue = new LinkedList<Observable<? extends T>>();
-            WIP_UPDATER.lazySet(this, 1);
+            this.wip = 1;
         }
 
         @Override

rxjava-core/src/main/java/rx/operators/OperatorSampleWithTime.java

@@ -72,7 +72,7 @@ public SamplerSubscriber(Subscriber<? super T> subscriber) {
         }
         @Override
         public void onNext(T t) {
-            VALUE_UPDATER.lazySet(this, t);
+            value = t;
         }
 
         @Override

rxjava-core/src/main/java/rx/plugins/RxJavaPlugins.java

@@ -15,7 +15,7 @@
  */
 package rx.plugins;
 
-import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+import java.util.concurrent.atomic.AtomicReference;
 
 /**
  * Registry for plugin implementations that allows global override and handles the retrieval of correct implementation based on order of precedence:
@@ -29,17 +29,10 @@
 public class RxJavaPlugins {
     private final static RxJavaPlugins INSTANCE = new RxJavaPlugins();
 
-    volatile RxJavaErrorHandler errorHandler;
-    volatile RxJavaObservableExecutionHook observableExecutionHook;
-    volatile RxJavaDefaultSchedulers schedulerOverrides;
+    private final AtomicReference<RxJavaErrorHandler> errorHandler = new AtomicReference<RxJavaErrorHandler>();
+    private final AtomicReference<RxJavaObservableExecutionHook> observableExecutionHook = new AtomicReference<RxJavaObservableExecutionHook>();
+    private final AtomicReference<RxJavaDefaultSchedulers> schedulerOverrides = new AtomicReference<RxJavaDefaultSchedulers>();
 
-    static final AtomicReferenceFieldUpdater<RxJavaPlugins, RxJavaErrorHandler> ERROR_HANDLER_UPDATER
-            = AtomicReferenceFieldUpdater.newUpdater(RxJavaPlugins.class, RxJavaErrorHandler.class, "errorHandler");
-    static final AtomicReferenceFieldUpdater<RxJavaPlugins, RxJavaObservableExecutionHook> EXECUTION_HOOK_UPDATER
-            = AtomicReferenceFieldUpdater.newUpdater(RxJavaPlugins.class, RxJavaObservableExecutionHook.class, "observableExecutionHook");
-    static final AtomicReferenceFieldUpdater<RxJavaPlugins, RxJavaDefaultSchedulers> SCHEDULER_OVERRIDE_UPDATER
-            = AtomicReferenceFieldUpdater.newUpdater(RxJavaPlugins.class, RxJavaDefaultSchedulers.class, "schedulerOverrides");
-    
     public static RxJavaPlugins getInstance() {
         return INSTANCE;
     }
@@ -48,9 +41,9 @@ public static RxJavaPlugins getInstance() {
 
     }
 
-    /* package accessible for unit tests */void reset() {
-        ERROR_HANDLER_UPDATER.lazySet(this, null);
-        EXECUTION_HOOK_UPDATER.lazySet(this, null);
+    /* package accessible for ujnit tests */void reset() {
+        INSTANCE.errorHandler.set(null);
+        INSTANCE.observableExecutionHook.set(null);
     }
 
     /**
@@ -62,19 +55,19 @@ public static RxJavaPlugins getInstance() {
      * @return {@link RxJavaErrorHandler} implementation to use
      */
     public RxJavaErrorHandler getErrorHandler() {
-        if (errorHandler == null) {
+        if (errorHandler.get() == null) {
             // check for an implementation from System.getProperty first
             Object impl = getPluginImplementationViaProperty(RxJavaErrorHandler.class);
             if (impl == null) {
                 // nothing set via properties so initialize with default 
-                ERROR_HANDLER_UPDATER.compareAndSet(this, null, RxJavaErrorHandlerDefault.getInstance());
+                errorHandler.compareAndSet(null, RxJavaErrorHandlerDefault.getInstance());
                 // we don't return from here but call get() again in case of thread-race so the winner will always get returned
             } else {
                 // we received an implementation from the system property so use it
-                ERROR_HANDLER_UPDATER.compareAndSet(this, null, (RxJavaErrorHandler) impl);
+                errorHandler.compareAndSet(null, (RxJavaErrorHandler) impl);
             }
         }
-        return errorHandler;
+        return errorHandler.get();
     }
 
     /**
@@ -86,8 +79,8 @@ public RxJavaErrorHandler getErrorHandler() {
      *             if called more than once or after the default was initialized (if usage occurs before trying to register)
      */
     public void registerErrorHandler(RxJavaErrorHandler impl) {
-        if (!ERROR_HANDLER_UPDATER.compareAndSet(this, null, impl)) {
-            throw new IllegalStateException("Another strategy was already registered: " + errorHandler);
+        if (!errorHandler.compareAndSet(null, impl)) {
+            throw new IllegalStateException("Another strategy was already registered: " + errorHandler.get());
         }
     }
 
@@ -100,19 +93,19 @@ public void registerErrorHandler(RxJavaErrorHandler impl) {
      * @return {@link RxJavaObservableExecutionHook} implementation to use
      */
     public RxJavaObservableExecutionHook getObservableExecutionHook() {
-        if (observableExecutionHook == null) {
+        if (observableExecutionHook.get() == null) {
             // check for an implementation from System.getProperty first
             Object impl = getPluginImplementationViaProperty(RxJavaObservableExecutionHook.class);
             if (impl == null) {
                 // nothing set via properties so initialize with default 
-                EXECUTION_HOOK_UPDATER.compareAndSet(this, null, RxJavaObservableExecutionHookDefault.getInstance());
+                observableExecutionHook.compareAndSet(null, RxJavaObservableExecutionHookDefault.getInstance());
                 // we don't return from here but call get() again in case of thread-race so the winner will always get returned
             } else {
                 // we received an implementation from the system property so use it
-                EXECUTION_HOOK_UPDATER.compareAndSet(this, null, (RxJavaObservableExecutionHook) impl);
+                observableExecutionHook.compareAndSet(null, (RxJavaObservableExecutionHook) impl);
             }
         }
-        return observableExecutionHook;
+        return observableExecutionHook.get();
     }
 
     /**
@@ -124,8 +117,8 @@ public RxJavaObservableExecutionHook getObservableExecutionHook() {
      *             if called more than once or after the default was initialized (if usage occurs before trying to register)
      */
     public void registerObservableExecutionHook(RxJavaObservableExecutionHook impl) {
-        if (!EXECUTION_HOOK_UPDATER.compareAndSet(this, null, impl)) {
-            throw new IllegalStateException("Another strategy was already registered: " + observableExecutionHook);
+        if (!observableExecutionHook.compareAndSet(null, impl)) {
+            throw new IllegalStateException("Another strategy was already registered: " + observableExecutionHook.get());
         }
     }
 
@@ -168,19 +161,19 @@ private static Object getPluginImplementationViaProperty(Class<?> pluginClass) {
      * @return {@link RxJavaErrorHandler} implementation to use
      */
     public RxJavaDefaultSchedulers getDefaultSchedulers() {
-        if (schedulerOverrides == null) {
+        if (schedulerOverrides.get() == null) {
             // check for an implementation from System.getProperty first
             Object impl = getPluginImplementationViaProperty(RxJavaDefaultSchedulers.class);
             if (impl == null) {
                 // nothing set via properties so initialize with default 
-                SCHEDULER_OVERRIDE_UPDATER.compareAndSet(this, null, RxJavaDefaultSchedulersDefault.getInstance());
+                schedulerOverrides.compareAndSet(null, RxJavaDefaultSchedulersDefault.getInstance());
                 // we don't return from here but call get() again in case of thread-race so the winner will always get returned
             } else {
                 // we received an implementation from the system property so use it
-                SCHEDULER_OVERRIDE_UPDATER.compareAndSet(this, null, (RxJavaDefaultSchedulers) impl);
+                schedulerOverrides.compareAndSet(null, (RxJavaDefaultSchedulers) impl);
             }
         }
-        return schedulerOverrides;
+        return schedulerOverrides.get();
     }
 
     /**
@@ -192,8 +185,8 @@ public RxJavaDefaultSchedulers getDefaultSchedulers() {
      *             if called more than once or after the default was initialized (if usage occurs before trying to register)
      */
     public void registerDefaultSchedulers(RxJavaDefaultSchedulers impl) {
-        if (!SCHEDULER_OVERRIDE_UPDATER.compareAndSet(this, null, impl)) {
-            throw new IllegalStateException("Another strategy was already registered: " + schedulerOverrides);
+        if (!schedulerOverrides.compareAndSet(null, impl)) {
+            throw new IllegalStateException("Another strategy was already registered: " + schedulerOverrides.get());
         }
     }
 }