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.9 by dl, Tue Sep 8 19:44:10 2015 UTC vs.
Revision 1.30 by jsr166, Mon Dec 16 22:55:54 2019 UTC

# Line 5 | Line 5
5   * http://creativecommons.org/publicdomain/zero/1.0/
6   */
7  
8 < import static java.util.concurrent.TimeUnit.MILLISECONDS;
9 < import static java.util.concurrent.TimeUnit.SECONDS;
10 <
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;
14 – import static java.util.concurrent.Flow.Publisher;
15 – import static java.util.concurrent.Flow.Subscriber;
16 – import static java.util.concurrent.Flow.Subscription;
17 – import java.util.concurrent.LinkedBlockingQueue;
13   import java.util.concurrent.ForkJoinPool;
14   import java.util.concurrent.SubmissionPublisher;
20 – import java.util.concurrent.ThreadFactory;
21 – import java.util.concurrent.ThreadPoolExecutor;
22 – import java.util.concurrent.TimeUnit;
15   import java.util.concurrent.atomic.AtomicInteger;
24 – import java.util.function.BiConsumer;
25 – import java.util.function.BiPredicate;
26 – import java.util.function.BiFunction;
27 –
16   import junit.framework.Test;
17   import junit.framework.TestSuite;
18  
19 + import static java.util.concurrent.Flow.Subscriber;
20 + import static java.util.concurrent.Flow.Subscription;
21 + import static java.util.concurrent.TimeUnit.MILLISECONDS;
22 +
23   public class SubmissionPublisherTest extends JSR166TestCase {
24  
25      public static void main(String[] args) {
# Line 37 | 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
41 <    static final class DaemonThreadFactory implements ThreadFactory {
42 <        public Thread newThread(Runnable r) {
43 <            Thread t = new Thread(r);
44 <            t.setDaemon(true);
45 <            return t;
46 <        }
47 <    }
48 <
49 <    static final Executor basicExecutor =
50 <        (ForkJoinPool.getCommonPoolParallelism() > 1) ?
51 <        ForkJoinPool.commonPool() :
52 <        new ThreadPoolExecutor(1, 1, 60, SECONDS,
53 <                               new LinkedBlockingQueue<Runnable>(),
54 <                               new DaemonThreadFactory());
32 >    final Executor basicExecutor = basicPublisher().getExecutor();
33  
34      static SubmissionPublisher<Integer> basicPublisher() {
35 <        return new SubmissionPublisher<Integer>(basicExecutor,
58 <                                                Flow.defaultBufferSize());
35 >        return new SubmissionPublisher<Integer>();
36      }
37  
38      static class SPException extends RuntimeException {}
# Line 167 | 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);
175 –        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 182 | 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 200 | 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 212 | 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 233 | 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 351 | Line 334 | public class SubmissionPublisherTest ext
334          TestSubscriber s = new TestSubscriber();
335          SubmissionPublisher<Integer> p = basicPublisher();
336          s.throwOnCall = true;
337 <        try {
355 <            p.subscribe(s);
356 <        } catch (Exception ok) {}
337 >        p.subscribe(s);
338          s.awaitError();
339          assertEquals(0, s.nexts);
340          assertEquals(1, s.errors);
# Line 395 | Line 376 | public class SubmissionPublisherTest ext
376  
377      /**
378       * Closing a publisher exceptionally causes onError to subscribers
379 +     * after they are subscribed
380       */
381      public void testCloseExceptionallyError() {
382          SubmissionPublisher<Integer> p = basicPublisher();
# Line 405 | Line 387 | public class SubmissionPublisherTest ext
387          p.submit(1);
388          p.closeExceptionally(new SPException());
389          assertTrue(p.isClosed());
390 +        s1.awaitSubscribe();
391          s1.awaitError();
392          assertTrue(s1.nexts <= 1);
393          assertEquals(1, s1.errors);
394 +        s2.awaitSubscribe();
395          s2.awaitError();
396          assertTrue(s2.nexts <= 1);
397          assertEquals(1, s2.errors);
# Line 417 | Line 401 | public class SubmissionPublisherTest ext
401       * Cancelling a subscription eventually causes no more onNexts to be issued
402       */
403      public void testCancel() {
404 <        SubmissionPublisher<Integer> p = basicPublisher();
404 >        SubmissionPublisher<Integer> p =
405 >            new SubmissionPublisher<>(basicExecutor, 4); // must be < 20
406          TestSubscriber s1 = new TestSubscriber();
407          TestSubscriber s2 = new TestSubscriber();
408          p.subscribe(s1);
# Line 461 | Line 446 | public class SubmissionPublisherTest ext
446       */
447      public void testThrowOnNextHandler() {
448          AtomicInteger calls = new AtomicInteger();
449 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>
450 <            (basicExecutor, 8,
466 <             (s, e) -> calls.getAndIncrement());
449 >        SubmissionPublisher<Integer> p = new SubmissionPublisher<>(
450 >            basicExecutor, 8, (s, e) -> calls.getAndIncrement());
451          TestSubscriber s1 = new TestSubscriber();
452          TestSubscriber s2 = new TestSubscriber();
453          p.subscribe(s1);
# Line 510 | Line 494 | public class SubmissionPublisherTest ext
494          s1.request = false;
495          p.subscribe(s1);
496          s1.awaitSubscribe();
497 <        assertTrue(p.estimateMinimumDemand() == 0);
497 >        assertEquals(0, p.estimateMinimumDemand());
498          TestSubscriber s2 = new TestSubscriber();
499          p.subscribe(s2);
500          p.submit(1);
# Line 551 | Line 535 | public class SubmissionPublisherTest ext
535      }
536  
537      /**
538 <     * Negative request causes error
538 >     * Non-positive request causes error
539       */
540      public void testRequest3() {
541          SubmissionPublisher<Integer> p = basicPublisher();
542          TestSubscriber s1 = new TestSubscriber();
543          TestSubscriber s2 = new TestSubscriber();
544 +        TestSubscriber s3 = new TestSubscriber();
545          p.subscribe(s1);
546          p.subscribe(s2);
547 +        p.subscribe(s3);
548 +        s3.awaitSubscribe();
549          s2.awaitSubscribe();
550          s1.awaitSubscribe();
551          s1.sn.request(-1L);
552 +        s3.sn.request(0L);
553          p.submit(1);
554          p.submit(2);
555          p.close();
# Line 571 | Line 559 | public class SubmissionPublisherTest ext
559          s1.awaitError();
560          assertEquals(1, s1.errors);
561          assertTrue(s1.lastError instanceof IllegalArgumentException);
562 +        s3.awaitError();
563 +        assertEquals(1, s3.errors);
564 +        assertTrue(s3.lastError instanceof IllegalArgumentException);
565      }
566  
567      /**
568       * estimateMinimumDemand reports 0 until request, nonzero after
569 <     * request, and zero again after delivery
569 >     * request
570       */
571      public void testEstimateMinimumDemand() {
572          TestSubscriber s = new TestSubscriber();
# Line 586 | Line 577 | public class SubmissionPublisherTest ext
577          assertEquals(0, p.estimateMinimumDemand());
578          s.sn.request(1);
579          assertEquals(1, p.estimateMinimumDemand());
589 –        p.submit(1);
590 –        s.awaitNext(1);
591 –        assertEquals(0, p.estimateMinimumDemand());
580      }
581  
582      /**
# Line 644 | Line 632 | public class SubmissionPublisherTest ext
632       * submit eventually issues requested items when buffer capacity is 1
633       */
634      public void testCap1Submit() {
635 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
636 <            basicExecutor, 1);
635 >        SubmissionPublisher<Integer> p
636 >            = new SubmissionPublisher<>(basicExecutor, 1);
637          TestSubscriber s1 = new TestSubscriber();
638          TestSubscriber s2 = new TestSubscriber();
639          p.subscribe(s1);
640          p.subscribe(s2);
641          for (int i = 1; i <= 20; ++i) {
654 –            assertTrue(p.estimateMinimumDemand() <= 1);
642              assertTrue(p.submit(i) >= 0);
643          }
644          p.close();
# Line 723 | Line 710 | public class SubmissionPublisherTest ext
710       * offer reports drops if saturated
711       */
712      public void testDroppedOffer() {
713 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
714 <            basicExecutor, 4);
713 >        SubmissionPublisher<Integer> p
714 >            = new SubmissionPublisher<>(basicExecutor, 4);
715          TestSubscriber s1 = new TestSubscriber();
716          s1.request = false;
717          TestSubscriber s2 = new TestSubscriber();
# Line 752 | Line 739 | public class SubmissionPublisherTest ext
739       */
740      public void testHandledDroppedOffer() {
741          AtomicInteger calls = new AtomicInteger();
742 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
743 <            basicExecutor, 4);
742 >        SubmissionPublisher<Integer> p
743 >            = new SubmissionPublisher<>(basicExecutor, 4);
744          TestSubscriber s1 = new TestSubscriber();
745          s1.request = false;
746          TestSubscriber s2 = new TestSubscriber();
# Line 780 | Line 767 | public class SubmissionPublisherTest ext
767       */
768      public void testRecoveredHandledDroppedOffer() {
769          AtomicInteger calls = new AtomicInteger();
770 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
771 <            basicExecutor, 4);
770 >        SubmissionPublisher<Integer> p
771 >            = new SubmissionPublisher<>(basicExecutor, 4);
772          TestSubscriber s1 = new TestSubscriber();
773          s1.request = false;
774          TestSubscriber s2 = new TestSubscriber();
# Line 861 | Line 848 | public class SubmissionPublisherTest ext
848       * Timed offer reports drops if saturated
849       */
850      public void testDroppedTimedOffer() {
851 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
852 <            basicExecutor, 4);
851 >        SubmissionPublisher<Integer> p
852 >            = new SubmissionPublisher<>(basicExecutor, 4);
853          TestSubscriber s1 = new TestSubscriber();
854          s1.request = false;
855          TestSubscriber s2 = new TestSubscriber();
# Line 893 | Line 880 | public class SubmissionPublisherTest ext
880       */
881      public void testHandledDroppedTimedOffer() {
882          AtomicInteger calls = new AtomicInteger();
883 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
884 <            basicExecutor, 4);
883 >        SubmissionPublisher<Integer> p
884 >            = new SubmissionPublisher<>(basicExecutor, 4);
885          TestSubscriber s1 = new TestSubscriber();
886          s1.request = false;
887          TestSubscriber s2 = new TestSubscriber();
# Line 923 | Line 910 | public class SubmissionPublisherTest ext
910       */
911      public void testRecoveredHandledDroppedTimedOffer() {
912          AtomicInteger calls = new AtomicInteger();
913 <        SubmissionPublisher<Integer> p = new SubmissionPublisher<Integer>(
914 <            basicExecutor, 4);
913 >        SubmissionPublisher<Integer> p
914 >            = new SubmissionPublisher<>(basicExecutor, 4);
915          TestSubscriber s1 = new TestSubscriber();
916          s1.request = false;
917          TestSubscriber s2 = new TestSubscriber();
# Line 948 | Line 935 | public class SubmissionPublisherTest ext
935          assertTrue(calls.get() >= 2);
936      }
937  
938 +    /**
939 +     * consume returns a CompletableFuture that is done when
940 +     * publisher completes
941 +     */
942 +    public void testConsume() {
943 +        AtomicInteger sum = new AtomicInteger();
944 +        SubmissionPublisher<Integer> p = basicPublisher();
945 +        CompletableFuture<Void> f =
946 +            p.consume((Integer x) -> sum.getAndAdd(x.intValue()));
947 +        int n = 20;
948 +        for (int i = 1; i <= n; ++i)
949 +            p.submit(i);
950 +        p.close();
951 +        f.join();
952 +        assertEquals((n * (n + 1)) / 2, sum.get());
953 +    }
954 +
955 +    /**
956 +     * consume(null) throws NPE
957 +     */
958 +    public void testConsumeNPE() {
959 +        SubmissionPublisher<Integer> p = basicPublisher();
960 +        try {
961 +            CompletableFuture<Void> unused = p.consume(null);
962 +            shouldThrow();
963 +        } catch (NullPointerException success) {}
964 +    }
965 +
966 +    /**
967 +     * consume eventually stops processing published items if cancelled
968 +     */
969 +    public void testCancelledConsume() {
970 +        AtomicInteger count = new AtomicInteger();
971 +        SubmissionPublisher<Integer> p = basicPublisher();
972 +        CompletableFuture<Void> f = p.consume(x -> count.getAndIncrement());
973 +        f.cancel(true);
974 +        int n = 1000000; // arbitrary limit
975 +        for (int i = 1; i <= n; ++i)
976 +            p.submit(i);
977 +        assertTrue(count.get() < n);
978 +    }
979 +
980 +    /**
981 +     * Tests scenario for
982 +     * JDK-8187947: A race condition in SubmissionPublisher
983 +     * 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
984 +     */
985 +    public void testMissedSignal_8187947() throws Exception {
986 +        if (!atLeastJava9()) return; // backport to jdk8 too hard
987 +        final int N =
988 +            ((ForkJoinPool.getCommonPoolParallelism() < 2) // JDK-8212899
989 +             ? (1 << 5)
990 +             : (1 << 10))
991 +            * (expensiveTests ? (1 << 10) : 1);
992 +        final CountDownLatch finished = new CountDownLatch(1);
993 +        final SubmissionPublisher<Boolean> pub = new SubmissionPublisher<>();
994 +        class Sub implements Subscriber<Boolean> {
995 +            int received;
996 +            public void onSubscribe(Subscription s) {
997 +                s.request(N);
998 +            }
999 +            public void onNext(Boolean item) {
1000 +                if (++received == N)
1001 +                    finished.countDown();
1002 +                else
1003 +                    CompletableFuture.runAsync(() -> pub.submit(Boolean.TRUE));
1004 +            }
1005 +            public void onError(Throwable t) { throw new AssertionError(t); }
1006 +            public void onComplete() {}
1007 +        }
1008 +        pub.subscribe(new Sub());
1009 +        checkTimedGet(
1010 +            CompletableFuture.runAsync(() -> pub.submit(Boolean.TRUE)),
1011 +            null);
1012 +        await(finished);
1013 +    }
1014   }

Diff Legend

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