ViewVC Help

View File | Revision Log | Show Annotations | Download File | Root Listing

root/jsr166/jsr166/src/test/tck/SubmissionPublisherTest.java

Comparing jsr166/src/test/tck/SubmissionPublisherTest.java (file contents):
Revision 1.8 by jsr166, Mon Sep 7 21:36:03 2015 UTC vs.
Revision 1.26 by jsr166, Sun Jan 7 22:59:18 2018 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() > 0) ?
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 285 | Line 268 | public class SubmissionPublisherTest ext
268		assertEquals(0, s2.nexts);
269		assertEquals(0, s2.errors);
270		assertEquals(0, s2.completes);
271	+	p.close();
272		}
273
274		/**
#	Line 394 | 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 404 | 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 416 | 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 460 | 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,
465	<	(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 509 | 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 550 | 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 570 | 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 585 | Line 579 | public class SubmissionPublisherTest ext
579		assertEquals(0, p.estimateMinimumDemand());
580		s.sn.request(1);
581		assertEquals(1, p.estimateMinimumDemand());
588	–	p.submit(1);
589	–	s.awaitNext(1);
590	–	assertEquals(0, p.estimateMinimumDemand());
582		}
583
584		/**
#	Line 643 | 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) {
653	–	assertTrue(p.estimateMinimumDemand() <= 1);
644		assertTrue(p.submit(i) >= 0);
645		}
646		p.close();
#	Line 722 | 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 751 | 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 779 | 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 860 | 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 870 | Line 860 | public class SubmissionPublisherTest ext
860		p.subscribe(s2);
861		s2.awaitSubscribe();
862		s1.awaitSubscribe();
863	+	long delay = timeoutMillis();
864		for (int i = 1; i <= 4; ++i)
865	<	assertTrue(p.offer(i, SHORT_DELAY_MS, MILLISECONDS, null) >= 0);
866	<	p.offer(5, SHORT_DELAY_MS, MILLISECONDS, null);
867	<	assertTrue(p.offer(6, SHORT_DELAY_MS, MILLISECONDS, null) < 0);
865	>	assertTrue(p.offer(i, delay, MILLISECONDS, null) >= 0);
866	>	long startTime = System.nanoTime();
867	>	assertTrue(p.offer(5, delay, MILLISECONDS, null) < 0);
868		s1.sn.request(64);
869	<	assertTrue(p.offer(7, SHORT_DELAY_MS, MILLISECONDS, null) < 0);
869	>	assertTrue(p.offer(6, delay, MILLISECONDS, null) < 0);
870	>	// 2 * delay should elapse but check only 1 * delay to allow timer slop
871	>	assertTrue(millisElapsedSince(startTime) >= delay);
872		s2.sn.request(64);
873		p.close();
874		s2.awaitComplete();
#	Line 889 | 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 899 | Line 892 | public class SubmissionPublisherTest ext
892		p.subscribe(s2);
893		s2.awaitSubscribe();
894		s1.awaitSubscribe();
895	+	long delay = timeoutMillis();
896		for (int i = 1; i <= 4; ++i)
897	<	assertTrue(p.offer(i, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) >= 0);
898	<	p.offer(5, (s, x) -> noopHandle(calls));
899	<	assertTrue(p.offer(6, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
897	>	assertTrue(p.offer(i, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) >= 0);
898	>	long startTime = System.nanoTime();
899	>	assertTrue(p.offer(5, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
900		s1.sn.request(64);
901	<	assertTrue(p.offer(7, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
901	>	assertTrue(p.offer(6, delay, MILLISECONDS, (s, x) -> noopHandle(calls)) < 0);
902	>	assertTrue(millisElapsedSince(startTime) >= delay);
903		s2.sn.request(64);
904		p.close();
905		s2.awaitComplete();
#	Line 917 | 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 928 | Line 923 | public class SubmissionPublisherTest ext
923		s2.awaitSubscribe();
924		s1.awaitSubscribe();
925		int n = 0;
926	<	for (int i = 1; i <= 8; ++i) {
927	<	int d = p.offer(i, SHORT_DELAY_MS, MILLISECONDS, (s, x) -> reqHandle(calls, s));
926	>	long delay = timeoutMillis();
927	>	long startTime = System.nanoTime();
928	>	for (int i = 1; i <= 6; ++i) {
929	>	int d = p.offer(i, delay, MILLISECONDS, (s, x) -> reqHandle(calls, s));
930		n = n + 2 + (d < 0 ? d : 0);
931		}
932	+	assertTrue(millisElapsedSince(startTime) >= delay);
933		p.close();
934		s2.awaitComplete();
935		s1.awaitComplete();
#	Line 939 | Line 937 | public class SubmissionPublisherTest ext
937		assertTrue(calls.get() >= 2);
938		}
939
940	+	/**
941	+	* consume returns a CompletableFuture that is done when
942	+	* publisher completes
943	+	*/
944	+	public void testConsume() {
945	+	AtomicInteger sum = new AtomicInteger();
946	+	SubmissionPublisher<Integer> p = basicPublisher();
947	+	CompletableFuture<Void> f =
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);
952	+	p.close();
953	+	f.join();
954	+	assertEquals((n * (n + 1)) / 2, sum.get());
955	+	}
956	+
957	+	/**
958	+	* consume(null) throws NPE
959	+	*/
960	+	public void testConsumeNPE() {
961	+	SubmissionPublisher<Integer> p = basicPublisher();
962	+	try {
963	+	CompletableFuture<Void> f = p.consume(null);
964	+	shouldThrow();
965	+	} catch (NullPointerException success) {}
966	+	}
967	+
968	+	/**
969	+	* consume eventually stops processing published items if cancelled
970	+	*/
971	+	public void testCancelledConsume() {
972	+	AtomicInteger count = new AtomicInteger();
973	+	SubmissionPublisher<Integer> p = basicPublisher();
974	+	CompletableFuture<Void> f = p.consume(x -> count.getAndIncrement());
975	+	f.cancel(true);
976	+	int n = 1000000; // arbitrary limit
977	+	for (int i = 1; i <= n; ++i)
978	+	p.submit(i);
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