update SubmissionPublisher to CVS rev. 1.83

Author: stefan-zobel

src/main/java/java9/util/concurrent/SubmissionPublisher.java

@@ -14,6 +14,7 @@
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.LockSupport;
+import java.util.concurrent.locks.ReentrantLock;
 
 import java9.util.Objects;
 import java9.util.function.BiConsumer;
@@ -146,15 +147,15 @@
  * @since 9
  */
 public class SubmissionPublisher<T> implements Publisher<T> {
-// CVS rev. 1.81
+// CVS rev. 1.83
     /*
      * Most mechanics are handled by BufferedSubscription. This class
      * mainly tracks subscribers and ensures sequentiality, by using
-     * built-in synchronization locks across public methods. Using
-     * built-in locks works well in the most typical case in which
-     * only one thread submits items. We extend this idea in
-     * submission methods by detecting single-ownership to reduce
-     * producer-consumer synchronization strength.
+     * locks across public methods, to ensure thread-safety in the
+     * presence of multiple sources and maintain acquire-release
+     * ordering around user operations. However, we also track whether
+     * there is only a single source, and if so streamline some buffer
+     * operations by avoiding some atomics.
      */
 
     /** The largest possible power of two array size. */
@@ -209,6 +210,8 @@ private static final class ThreadPerTaskExecutor implements Executor {
      */
     BufferedSubscription<T> clients;
 
+    /** Lock for exclusion across multiple sources */
+    final ReentrantLock lock;
     /** Run status, updated only within locks */
     volatile boolean closed;
     /** Set true on first call to subscribe, to initialize possible owner */
@@ -248,6 +251,7 @@ public SubmissionPublisher(Executor executor, int maxBufferCapacity,
         Objects.requireNonNull(executor);
         if (maxBufferCapacity <= 0)
             throw new IllegalArgumentException("capacity must be positive");
+        this.lock = new ReentrantLock();
         this.executor = executor;
         this.onNextHandler = handler;
         this.maxBufferCapacity = roundCapacity(maxBufferCapacity);
@@ -311,13 +315,15 @@ public SubmissionPublisher() {
      */
     public void subscribe(Subscriber<? super T> subscriber) {
         Objects.requireNonNull(subscriber);
+        ReentrantLock lock = this.lock;
         int max = maxBufferCapacity; // allocate initial array
         Object[] array = new Object[max < INITIAL_CAPACITY ?
                                     max : INITIAL_CAPACITY];
         BufferedSubscription<T> subscription =
             new BufferedSubscription<T>(subscriber, executor, onNextHandler,
                                         array, max);
-        synchronized (this) {
+        lock.lock();
+        try {
             if (!subscribed) {
                 subscribed = true;
                 owner = Thread.currentThread();
@@ -352,6 +358,8 @@ else if (subscriber.equals(b.subscriber)) {
                     pred = b;
                 b = next;
             }
+        } finally {
+            lock.unlock();
         }
     }
 
@@ -364,7 +372,9 @@ private int doOffer(T item, long nanos,
         Objects.requireNonNull(item);
         int lag = 0;
         boolean complete, unowned;
-        synchronized (this) {
+        ReentrantLock lock = this.lock;
+        lock.lock();
+        try {
             Thread t = Thread.currentThread(), o;
             BufferedSubscription<T> b = clients;
             if ((unowned = ((o = owner) != t)) && o != null)
@@ -395,6 +405,8 @@ else if (stat > lag)
                 if (retries != null || cleanMe)
                     lag = retryOffer(item, nanos, onDrop, retries, lag, cleanMe);
             }
+        } finally {
+            lock.unlock();
         }
         if (complete)
             throw new IllegalStateException("Closed");
@@ -583,14 +595,18 @@ public int offer(T item, long timeout, TimeUnit unit,
      * subscribers have yet completed.
      */
     public void close() {
+        ReentrantLock lock = this.lock;
         if (!closed) {
             BufferedSubscription<T> b;
-            synchronized (this) {
+            lock.lock();
+            try {
                 // no need to re-check closed here
                 b = clients;
                 clients = null;
                 owner = null;
                 closed = true;
+            } finally {
+                lock.unlock();
             }
             while (b != null) {
                 BufferedSubscription<T> next = b.next;
@@ -614,16 +630,20 @@ public void close() {
      */
     public void closeExceptionally(Throwable error) {
         Objects.requireNonNull(error);
+        ReentrantLock lock = this.lock;
         if (!closed) {
             BufferedSubscription<T> b;
-            synchronized (this) {
+            lock.lock();
+            try {
                 b = clients;
                 if (!closed) {  // don't clobber racing close
                     closedException = error;
                     clients = null;
                     owner = null;
                     closed = true;
                 }
+            } finally {
+                lock.unlock();
             }
             while (b != null) {
                 BufferedSubscription<T> next = b.next;
@@ -661,7 +681,9 @@ public Throwable getClosedException() {
      */
     public boolean hasSubscribers() {
         boolean nonEmpty = false;
-        synchronized (this) {
+        ReentrantLock lock = this.lock;
+        lock.lock();
+        try {
             for (BufferedSubscription<T> b = clients; b != null;) {
                 BufferedSubscription<T> next = b.next;
                 if (b.isClosed()) {
@@ -673,6 +695,8 @@ public boolean hasSubscribers() {
                     break;
                 }
             }
+        } finally {
+            lock.unlock();
         }
         return nonEmpty;
     }
@@ -683,9 +707,15 @@ public boolean hasSubscribers() {
      * @return the number of current subscribers
      */
     public int getNumberOfSubscribers() {
-        synchronized (this) {
-            return cleanAndCount();
+        int n;
+        ReentrantLock lock = this.lock;
+        lock.lock();
+        try {
+            n = cleanAndCount();
+        } finally {
+            lock.unlock();
         }
+        return n;
     }
 
     /**
@@ -715,7 +745,9 @@ public int getMaxBufferCapacity() {
      */
     public List<Subscriber<? super T>> getSubscribers() {
         ArrayList<Subscriber<? super T>> subs = new ArrayList<Subscriber<? super T>>();
-        synchronized (this) {
+        ReentrantLock lock = this.lock;
+        lock.lock();
+        try {
             BufferedSubscription<T> pred = null, next;
             for (BufferedSubscription<T> b = clients; b != null; b = next) {
                 next = b.next;
@@ -731,6 +763,8 @@ public List<Subscriber<? super T>> getSubscribers() {
                     pred = b;
                 }
             }
+        } finally {
+            lock.unlock();
         }
         return subs;
     }
@@ -744,8 +778,11 @@ public List<Subscriber<? super T>> getSubscribers() {
      */
     public boolean isSubscribed(Subscriber<? super T> subscriber) {
         Objects.requireNonNull(subscriber);
+        boolean subscribed = false;
+        ReentrantLock lock = this.lock;
         if (!closed) {
-            synchronized (this) {
+            lock.lock();
+            try {
                 BufferedSubscription<T> pred = null, next;
                 for (BufferedSubscription<T> b = clients; b != null; b = next) {
                     next = b.next;
@@ -756,14 +793,16 @@ public boolean isSubscribed(Subscriber<? super T> subscriber) {
                         else
                             pred.next = next;
                     }
-                    else if (subscriber.equals(b.subscriber))
-                        return true;
+                    else if (subscribed = subscriber.equals(b.subscriber))
+                        break;
                     else
                         pred = b;
                 }
+            } finally {
+                lock.unlock();
             }
         }
-        return false;
+        return subscribed;
     }
 
     /**
@@ -776,7 +815,9 @@ else if (subscriber.equals(b.subscriber))
     public long estimateMinimumDemand() {
         long min = Long.MAX_VALUE;
         boolean nonEmpty = false;
-        synchronized (this) {
+        ReentrantLock lock = this.lock;
+        lock.lock();
+        try {
             BufferedSubscription<T> pred = null, next;
             for (BufferedSubscription<T> b = clients; b != null; b = next) {
                 int n; long d;
@@ -795,6 +836,8 @@ public long estimateMinimumDemand() {
                     pred = b;
                 }
             }
+        } finally {
+            lock.unlock();
         }
         return nonEmpty ? min : 0;
     }
@@ -807,7 +850,9 @@ public long estimateMinimumDemand() {
      */
     public int estimateMaximumLag() {
         int max = 0;
-        synchronized (this) {
+        ReentrantLock lock = this.lock;
+        lock.lock();
+        try {
             BufferedSubscription<T> pred = null, next;
             for (BufferedSubscription<T> b = clients; b != null; b = next) {
                 int n;
@@ -825,6 +870,8 @@ public int estimateMaximumLag() {
                     pred = b;
                 }
             }
+        } finally {
+            lock.unlock();
         }
         return max;
     }