ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/test/tck/SubmissionPublisherTest.java
(Generate patch)

Comparing jsr166/src/test/tck/SubmissionPublisherTest.java (file contents):
Revision 1.11 by jsr166, Sat Sep 12 17:11:12 2015 UTC vs.
Revision 1.26 by jsr166, Sun Jan 7 22:59:18 2018 UTC

# Line 6 | Line 6
6   */
7  
8   import java.util.concurrent.CompletableFuture;
9 + import java.util.concurrent.CountDownLatch;
10   import java.util.concurrent.Executor;
11   import java.util.concurrent.Executors;
12   import java.util.concurrent.Flow;
13   import java.util.concurrent.ForkJoinPool;
13 – import java.util.concurrent.LinkedBlockingQueue;
14   import java.util.concurrent.SubmissionPublisher;
15 – import java.util.concurrent.ThreadFactory;
16 – import java.util.concurrent.ThreadPoolExecutor;
17 – import java.util.concurrent.TimeUnit;
15   import java.util.concurrent.atomic.AtomicInteger;
19 – import java.util.function.BiConsumer;
20 – import java.util.function.BiFunction;
21 – import java.util.function.BiPredicate;
22 – import java.util.stream.Stream;
16   import junit.framework.Test;
17   import junit.framework.TestSuite;
18  
26 – import static java.util.concurrent.Flow.Publisher;
19   import static java.util.concurrent.Flow.Subscriber;
20   import static java.util.concurrent.Flow.Subscription;
21   import static java.util.concurrent.TimeUnit.MILLISECONDS;
30 – import static java.util.concurrent.TimeUnit.SECONDS;
22  
23   public class SubmissionPublisherTest extends JSR166TestCase {
24  
# Line 38 | Line 29 | public class SubmissionPublisherTest ext
29          return new TestSuite(SubmissionPublisherTest.class);
30      }
31  
32 <    // Factory for single thread pool in case commonPool parallelism is zero
42 <    static final class DaemonThreadFactory implements ThreadFactory {
43 <        public Thread newThread(Runnable r) {
44 <            Thread t = new Thread(r);
45 <            t.setDaemon(true);
46 <            return t;
47 <        }
48 <    }
49 <
50 <    static final Executor basicExecutor =
51 <        (ForkJoinPool.getCommonPoolParallelism() > 1) ?
52 <        ForkJoinPool.commonPool() :
53 <        new ThreadPoolExecutor(1, 1, 60, SECONDS,
54 <                               new LinkedBlockingQueue<Runnable>(),
55 <                               new DaemonThreadFactory());
32 >    final Executor basicExecutor = basicPublisher().getExecutor();
33  
34      static SubmissionPublisher<Integer> basicPublisher() {
35 <        return new SubmissionPublisher<Integer>(basicExecutor,
59 <                                                Flow.defaultBufferSize());
35 >        return new SubmissionPublisher<Integer>();
36      }
37  
38      static class SPException extends RuntimeException {}
# Line 168 | Line 144 | public class SubmissionPublisherTest ext
144      /**
145       * A default-constructed SubmissionPublisher has no subscribers,
146       * is not closed, has default buffer size, and uses the
147 <     * ForkJoinPool.commonPool executor
147 >     * defaultExecutor
148       */
149      public void testConstructor1() {
150 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>();
150 >        SubmissionPublisher<Integer> p = new SubmissionPublisher<>();
151          checkInitialState(p);
176 –        assertSame(p.getExecutor(), ForkJoinPool.commonPool());
152          assertEquals(p.getMaxBufferCapacity(), Flow.defaultBufferSize());
153 +        Executor e = p.getExecutor(), c = ForkJoinPool.commonPool();
154 +        if (ForkJoinPool.getCommonPoolParallelism() > 1)
155 +            assertSame(e, c);
156 +        else
157 +            assertNotSame(e, c);
158      }
159  
160      /**
# Line 183 | Line 163 | public class SubmissionPublisherTest ext
163       */
164      public void testConstructor2() {
165          Executor e = Executors.newFixedThreadPool(1);
166 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(e, 8);
166 >        SubmissionPublisher<Integer> p = new SubmissionPublisher<>(e, 8);
167          checkInitialState(p);
168          assertSame(p.getExecutor(), e);
169          assertEquals(8, p.getMaxBufferCapacity());
170      }
171  
172      /**
173 <     * A null Executor argument to SubmissionPublisher constructor throws NPE
173 >     * A null Executor argument to SubmissionPublisher constructor
174 >     * throws NullPointerException
175       */
176      public void testConstructor3() {
177          try {
# Line 201 | Line 182 | public class SubmissionPublisherTest ext
182  
183      /**
184       * A negative capacity argument to SubmissionPublisher constructor
185 <     * throws IAE
185 >     * throws IllegalArgumentException
186       */
187      public void testConstructor4() {
188          Executor e = Executors.newFixedThreadPool(1);
# Line 213 | Line 194 | public class SubmissionPublisherTest ext
194  
195      /**
196       * A closed publisher reports isClosed with no closedException and
197 <     * throws ISE upon attempted submission; a subsequent close or
198 <     * closeExceptionally has no additional effect.
197 >     * throws IllegalStateException upon attempted submission; a
198 >     * subsequent close or closeExceptionally has no additional
199 >     * effect.
200       */
201      public void testClose() {
202          SubmissionPublisher<Integer> p = basicPublisher();
# Line 234 | Line 216 | public class SubmissionPublisherTest ext
216  
217      /**
218       * A publisher closedExceptionally reports isClosed with the
219 <     * closedException and throws ISE upon attempted submission; a
220 <     * subsequent close or closeExceptionally has no additional
221 <     * effect.
219 >     * closedException and throws IllegalStateException upon attempted
220 >     * submission; a subsequent close or closeExceptionally has no
221 >     * additional effect.
222       */
223      public void testCloseExceptionally() {
224          SubmissionPublisher<Integer> p = basicPublisher();
# Line 396 | Line 378 | public class SubmissionPublisherTest ext
378  
379      /**
380       * Closing a publisher exceptionally causes onError to subscribers
381 +     * after they are subscribed
382       */
383      public void testCloseExceptionallyError() {
384          SubmissionPublisher<Integer> p = basicPublisher();
# Line 406 | Line 389 | public class SubmissionPublisherTest ext
389          p.submit(1);
390          p.closeExceptionally(new SPException());
391          assertTrue(p.isClosed());
392 +        s1.awaitSubscribe();
393          s1.awaitError();
394          assertTrue(s1.nexts <= 1);
395          assertEquals(1, s1.errors);
396 +        s2.awaitSubscribe();
397          s2.awaitError();
398          assertTrue(s2.nexts <= 1);
399          assertEquals(1, s2.errors);
# Line 418 | Line 403 | public class SubmissionPublisherTest ext
403       * Cancelling a subscription eventually causes no more onNexts to be issued
404       */
405      public void testCancel() {
406 <        SubmissionPublisher<Integer> p = basicPublisher();
406 >        SubmissionPublisher<Integer> p =
407 >            new SubmissionPublisher<>(basicExecutor, 4); // must be < 20
408          TestSubscriber s1 = new TestSubscriber();
409          TestSubscriber s2 = new TestSubscriber();
410          p.subscribe(s1);
# Line 462 | Line 448 | public class SubmissionPublisherTest ext
448       */
449      public void testThrowOnNextHandler() {
450          AtomicInteger calls = new AtomicInteger();
451 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>
452 <            (basicExecutor, 8,
467 <             (s, e) -> calls.getAndIncrement());
451 >        SubmissionPublisher<Integer> p = new SubmissionPublisher<>(
452 >            basicExecutor, 8, (s, e) -> calls.getAndIncrement());
453          TestSubscriber s1 = new TestSubscriber();
454          TestSubscriber s2 = new TestSubscriber();
455          p.subscribe(s1);
# Line 511 | Line 496 | public class SubmissionPublisherTest ext
496          s1.request = false;
497          p.subscribe(s1);
498          s1.awaitSubscribe();
499 <        assertTrue(p.estimateMinimumDemand() == 0);
499 >        assertEquals(0, p.estimateMinimumDemand());
500          TestSubscriber s2 = new TestSubscriber();
501          p.subscribe(s2);
502          p.submit(1);
# Line 552 | Line 537 | public class SubmissionPublisherTest ext
537      }
538  
539      /**
540 <     * Negative request causes error
540 >     * Non-positive request causes error
541       */
542      public void testRequest3() {
543          SubmissionPublisher<Integer> p = basicPublisher();
544          TestSubscriber s1 = new TestSubscriber();
545          TestSubscriber s2 = new TestSubscriber();
546 +        TestSubscriber s3 = new TestSubscriber();
547          p.subscribe(s1);
548          p.subscribe(s2);
549 +        p.subscribe(s3);
550 +        s3.awaitSubscribe();
551          s2.awaitSubscribe();
552          s1.awaitSubscribe();
553          s1.sn.request(-1L);
554 +        s3.sn.request(0L);
555          p.submit(1);
556          p.submit(2);
557          p.close();
# Line 572 | Line 561 | public class SubmissionPublisherTest ext
561          s1.awaitError();
562          assertEquals(1, s1.errors);
563          assertTrue(s1.lastError instanceof IllegalArgumentException);
564 +        s3.awaitError();
565 +        assertEquals(1, s3.errors);
566 +        assertTrue(s3.lastError instanceof IllegalArgumentException);
567      }
568  
569      /**
570       * estimateMinimumDemand reports 0 until request, nonzero after
571 <     * request, and zero again after delivery
571 >     * request
572       */
573      public void testEstimateMinimumDemand() {
574          TestSubscriber s = new TestSubscriber();
# Line 587 | Line 579 | public class SubmissionPublisherTest ext
579          assertEquals(0, p.estimateMinimumDemand());
580          s.sn.request(1);
581          assertEquals(1, p.estimateMinimumDemand());
590 –        p.submit(1);
591 –        s.awaitNext(1);
592 –        assertEquals(0, p.estimateMinimumDemand());
582      }
583  
584      /**
# Line 645 | Line 634 | public class SubmissionPublisherTest ext
634       * submit eventually issues requested items when buffer capacity is 1
635       */
636      public void testCap1Submit() {
637 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
638 <            basicExecutor, 1);
637 >        SubmissionPublisher<Integer> p
638 >            = new SubmissionPublisher<>(basicExecutor, 1);
639          TestSubscriber s1 = new TestSubscriber();
640          TestSubscriber s2 = new TestSubscriber();
641          p.subscribe(s1);
642          p.subscribe(s2);
643          for (int i = 1; i <= 20; ++i) {
655 –            assertTrue(p.estimateMinimumDemand() <= 1);
644              assertTrue(p.submit(i) >= 0);
645          }
646          p.close();
# Line 724 | Line 712 | public class SubmissionPublisherTest ext
712       * offer reports drops if saturated
713       */
714      public void testDroppedOffer() {
715 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
716 <            basicExecutor, 4);
715 >        SubmissionPublisher<Integer> p
716 >            = new SubmissionPublisher<>(basicExecutor, 4);
717          TestSubscriber s1 = new TestSubscriber();
718          s1.request = false;
719          TestSubscriber s2 = new TestSubscriber();
# Line 753 | Line 741 | public class SubmissionPublisherTest ext
741       */
742      public void testHandledDroppedOffer() {
743          AtomicInteger calls = new AtomicInteger();
744 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
745 <            basicExecutor, 4);
744 >        SubmissionPublisher<Integer> p
745 >            = new SubmissionPublisher<>(basicExecutor, 4);
746          TestSubscriber s1 = new TestSubscriber();
747          s1.request = false;
748          TestSubscriber s2 = new TestSubscriber();
# Line 781 | Line 769 | public class SubmissionPublisherTest ext
769       */
770      public void testRecoveredHandledDroppedOffer() {
771          AtomicInteger calls = new AtomicInteger();
772 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
773 <            basicExecutor, 4);
772 >        SubmissionPublisher<Integer> p
773 >            = new SubmissionPublisher<>(basicExecutor, 4);
774          TestSubscriber s1 = new TestSubscriber();
775          s1.request = false;
776          TestSubscriber s2 = new TestSubscriber();
# Line 862 | Line 850 | public class SubmissionPublisherTest ext
850       * Timed offer reports drops if saturated
851       */
852      public void testDroppedTimedOffer() {
853 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
854 <            basicExecutor, 4);
853 >        SubmissionPublisher<Integer> p
854 >            = new SubmissionPublisher<>(basicExecutor, 4);
855          TestSubscriber s1 = new TestSubscriber();
856          s1.request = false;
857          TestSubscriber s2 = new TestSubscriber();
# Line 894 | Line 882 | public class SubmissionPublisherTest ext
882       */
883      public void testHandledDroppedTimedOffer() {
884          AtomicInteger calls = new AtomicInteger();
885 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
886 <            basicExecutor, 4);
885 >        SubmissionPublisher<Integer> p
886 >            = new SubmissionPublisher<>(basicExecutor, 4);
887          TestSubscriber s1 = new TestSubscriber();
888          s1.request = false;
889          TestSubscriber s2 = new TestSubscriber();
# Line 924 | Line 912 | public class SubmissionPublisherTest ext
912       */
913      public void testRecoveredHandledDroppedTimedOffer() {
914          AtomicInteger calls = new AtomicInteger();
915 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
916 <            basicExecutor, 4);
915 >        SubmissionPublisher<Integer> p
916 >            = new SubmissionPublisher<>(basicExecutor, 4);
917          TestSubscriber s1 = new TestSubscriber();
918          s1.request = false;
919          TestSubscriber s2 = new TestSubscriber();
# Line 957 | Line 945 | public class SubmissionPublisherTest ext
945          AtomicInteger sum = new AtomicInteger();
946          SubmissionPublisher<Integer> p = basicPublisher();
947          CompletableFuture<Void> f =
948 <            p.consume((Integer x) -> { sum.getAndAdd(x.intValue()); });
948 >            p.consume((Integer x) -> sum.getAndAdd(x.intValue()));
949          int n = 20;
950          for (int i = 1; i <= n; ++i)
951              p.submit(i);
# Line 991 | Line 979 | public class SubmissionPublisherTest ext
979          assertTrue(count.get() < n);
980      }
981  
982 +    /**
983 +     * Tests scenario for
984 +     * JDK-8187947: A race condition in SubmissionPublisher
985 +     * cvs update -D '2017-11-25' src/main/java/util/concurrent/SubmissionPublisher.java && ant -Djsr166.expensiveTests=true -Djsr166.tckTestClass=SubmissionPublisherTest -Djsr166.methodFilter=testMissedSignal tck; cvs update -A src/main/java/util/concurrent/SubmissionPublisher.java
986 +     */
987 +    public void testMissedSignal_8187947() throws Exception {
988 +        if (!atLeastJava9()) return; // backport to jdk8 too hard
989 +        final int N = expensiveTests ? (1 << 20) : (1 << 10);
990 +        final CountDownLatch finished = new CountDownLatch(1);
991 +        final SubmissionPublisher<Boolean> pub = new SubmissionPublisher<>();
992 +        class Sub implements Subscriber<Boolean> {
993 +            int received;
994 +            public void onSubscribe(Subscription s) {
995 +                s.request(N);
996 +            }
997 +            public void onNext(Boolean item) {
998 +                if (++received == N)
999 +                    finished.countDown();
1000 +                else
1001 +                    CompletableFuture.runAsync(() -> pub.submit(Boolean.TRUE));
1002 +            }
1003 +            public void onError(Throwable t) { throw new AssertionError(t); }
1004 +            public void onComplete() {}
1005 +        }
1006 +        pub.subscribe(new Sub());
1007 +        CompletableFuture.runAsync(() -> pub.submit(Boolean.TRUE));
1008 +        await(finished);
1009 +    }
1010   }

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines