| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| package org.openide.util; |
| |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Set; |
| import java.util.concurrent.Callable; |
| import java.util.concurrent.CancellationException; |
| import java.util.concurrent.CopyOnWriteArrayList; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.Future; |
| import java.util.concurrent.ScheduledFuture; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import java.util.logging.Logger; |
| import org.netbeans.junit.NbTestCase; |
| import org.netbeans.junit.RandomlyFails; |
| import org.openide.util.Cancellable; |
| import org.openide.util.Exceptions; |
| import org.openide.util.RequestProcessor; |
| |
| /** |
| * |
| * @author Tim Boudreau |
| */ |
| public class RequestProcessor180386Test extends NbTestCase { |
| private static final Logger LOG = Logger.getLogger(RequestProcessor180386Test.class.getName()); |
| |
| public RequestProcessor180386Test(java.lang.String testName) { |
| super(testName); |
| } |
| |
| public void testSubmit() throws Exception { |
| class C implements Callable<String> { |
| |
| volatile boolean hasRun; |
| |
| @Override |
| public String call() throws Exception { |
| String result = "Hello"; |
| hasRun = true; |
| return result; |
| } |
| } |
| C c = new C(); |
| Future<String> f = RequestProcessor.getDefault().submit(c); |
| assertEquals("Hello", f.get()); |
| assertTrue(c.hasRun); |
| |
| class R implements Runnable { |
| |
| volatile boolean hasRun; |
| |
| @Override |
| public void run() { |
| hasRun = true; |
| } |
| } |
| R r = new R(); |
| f = RequestProcessor.getDefault().submit(r, "Goodbye"); |
| assertEquals("Goodbye", f.get()); |
| assertTrue(r.hasRun); |
| } |
| |
| @RandomlyFails // NB-Core-Build #4352: notRun.empty |
| public void testSomeTasksNotRunIfShutDown() throws Exception { |
| final Object lock = new Object(); |
| int count = 10; |
| final CountDownLatch waitAllLaunched = new CountDownLatch(count); |
| final CountDownLatch waitOneFinished = new CountDownLatch(1); |
| final RequestProcessor notificationThread = new RequestProcessor("notifier", 1, true, true); |
| |
| RequestProcessor rp = new RequestProcessor("TestRP", count * 2); |
| class R implements Runnable { |
| |
| volatile boolean hasStarted; |
| volatile boolean hasFinished; |
| |
| @Override |
| public void run() { |
| hasStarted = true; |
| waitAllLaunched.countDown(); |
| synchronized (lock) { |
| try { |
| lock.wait(); |
| if (Thread.interrupted()) { |
| return; |
| } |
| } catch (InterruptedException ex) { |
| return; |
| } finally { |
| new N(waitOneFinished).launch(); |
| } |
| hasFinished = true; |
| } |
| } |
| |
| class N implements Runnable { |
| private final CountDownLatch l; |
| N (CountDownLatch l) { |
| this.l = l; |
| } |
| |
| void launch() { |
| notificationThread.create(this).schedule(20); |
| } |
| |
| @Override |
| public void run() { |
| l.countDown(); |
| } |
| |
| } |
| } |
| Set<Future<String>> s = new HashSet<Future<String>>(); |
| Set<R> rs = new HashSet<R>(); |
| for (int i = 0; i < count; i++) { |
| String currName = "Runnable " + i; |
| R r = new R(); |
| rs.add(r); |
| s.add(rp.submit(r, currName)); |
| } |
| waitAllLaunched.await(); |
| synchronized (lock) { |
| //Notify just one thread |
| lock.notify(); |
| } |
| waitOneFinished.await(); |
| List<Runnable> notRun = rp.shutdownNow(); |
| synchronized (lock) { |
| lock.notifyAll(); |
| } |
| boolean allFinished = true; |
| int finishedCount = 0; |
| for (R r : rs) { |
| assertTrue(r.hasStarted); |
| allFinished &= r.hasFinished; |
| if (r.hasFinished) { |
| finishedCount++; |
| } |
| } |
| assertFalse("All tasks should not have completed", allFinished); |
| assertTrue("At least one task shall complete", finishedCount >= 1); |
| assertTrue(notRun.isEmpty()); |
| assertTrue(rp.isShutdown()); |
| //Technically not provable due to "spurious wakeups" |
| // assertEquals (1, finishedCount); |
| |
| try { |
| RequestProcessor.getDefault().shutdown(); |
| fail("Should not be able to shutdown() default RP"); |
| } catch (Exception e) { |
| } |
| try { |
| RequestProcessor.getDefault().shutdownNow(); |
| fail("Should not be able to shutdownNow() default RP"); |
| } catch (Exception e) { |
| } |
| } |
| |
| public void testAwaitTermination() throws Exception { |
| int count = 20; |
| final Object lock = new Object(); |
| final CountDownLatch waitAllLaunched = new CountDownLatch(count); |
| final CountDownLatch waitAll = new CountDownLatch(count); |
| final RequestProcessor rp = new RequestProcessor("TestRP", count); |
| class R implements Runnable { |
| |
| volatile boolean hasStarted; |
| volatile boolean hasFinished; |
| |
| @Override |
| public void run() { |
| hasStarted = true; |
| waitAllLaunched.countDown(); |
| synchronized (lock) { |
| try { |
| lock.wait(); |
| if (Thread.interrupted()) { |
| return; |
| } |
| } catch (InterruptedException ex) { |
| return; |
| } finally { |
| hasFinished = true; |
| waitAll.countDown(); |
| } |
| } |
| } |
| } |
| Set<Future<String>> s = new HashSet<Future<String>>(); |
| Set<R> rs = new HashSet<R>(); |
| for (int i = 0; i < count; i++) { |
| String currName = "Runnable " + i; |
| R r = new R(); |
| rs.add(r); |
| s.add(rp.submit(r, currName)); |
| } |
| waitAllLaunched.await(); |
| synchronized (lock) { |
| //Notify just one thread |
| lock.notifyAll(); |
| } |
| rp.shutdown(); |
| boolean awaitTermination = rp.awaitTermination(1, TimeUnit.DAYS); |
| assertTrue(awaitTermination); |
| assertTrue(rp.isShutdown()); |
| assertTrue(rp.isTerminated()); |
| } |
| |
| @RandomlyFails |
| public void testAwaitTerminationWaitsForNewlyAddedThreads() throws Exception { |
| final RequestProcessor rp = new RequestProcessor("testAwaitTerminationWaitsForNewlyAddedThreads", 50, false); |
| int count = 30; |
| final CountDownLatch waitLock = new CountDownLatch(1); |
| class R implements Runnable { |
| boolean done; |
| @Override |
| public void run() { |
| try { |
| waitLock.await(); |
| } catch (InterruptedException ex) { |
| done = true; |
| } finally { |
| done = true; |
| } |
| } |
| } |
| Set<R> rs = new HashSet<R>(); |
| for (int i= 0; i < count; i++) { |
| R r = new R(); |
| rs.add(r); |
| rp.submit(r); |
| } |
| final CountDownLatch shutdownBegun = new CountDownLatch(1); |
| Runnable shutdowner = new Runnable() { |
| @Override |
| public void run() { |
| try { |
| Thread.sleep(1000); |
| rp.shutdown(); |
| shutdownBegun.countDown(); |
| } catch (InterruptedException ex) { |
| Exceptions.printStackTrace(ex); |
| } |
| } |
| }; |
| waitLock.countDown(); |
| new Thread(shutdowner).start(); |
| assertTrue(rp.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS)); |
| Thread.sleep (600); |
| assertTrue (rp.isTerminated()); |
| } |
| |
| public void testInvokeAll() throws Exception { |
| int count = 20; |
| final CountDownLatch waitAll = new CountDownLatch(count); |
| final RequestProcessor notificationThread = new RequestProcessor("notifier", 1, true, true); |
| final RequestProcessor rp = new RequestProcessor("TestRP", count); |
| try { |
| class C implements Callable<String>, Runnable { |
| |
| private final String result; |
| volatile boolean ran; |
| |
| C(String result) { |
| this.result = result; |
| } |
| |
| @Override |
| public String call() throws Exception { |
| ran = true; |
| //#182637 - the waiting thread can be notified before |
| //the Done flag on this runnable's future has been set, |
| //so ensure this thread's runnable has time to exit |
| //before we do the notification |
| notificationThread.create(this).schedule (20); |
| return result; |
| } |
| |
| @Override |
| public void run() { |
| waitAll.countDown(); |
| } |
| } |
| List<C> callables = new ArrayList<C>(count); |
| List<Future<String>> fs; |
| Set<String> names = new HashSet<String>(count); |
| for (int i = 0; i < count; i++) { |
| String name = "R" + i; |
| names.add(name); |
| C c = new C(name); |
| callables.add(c); |
| } |
| fs = rp.invokeAll(callables); |
| |
| assertNotNull(fs); |
| waitAll.await(); |
| assertEquals(0, waitAll.getCount()); |
| for (Future<String> f : fs) { |
| assertTrue (f.isDone()); |
| } |
| for (C c : callables) { |
| assertTrue (c.ran); |
| } |
| Set<String> s = new HashSet<String>(count); |
| for (Future<String> f : fs) { |
| s.add(f.get()); |
| } |
| assertEquals(names, s); |
| } finally { |
| rp.stop(); |
| } |
| } |
| |
| public void testInvokeAllWithTimeout() throws Exception { |
| int count = 20; |
| final CountDownLatch blocker = new CountDownLatch(1); |
| final RequestProcessor rp = new RequestProcessor("TestRP", count); |
| try { |
| class C implements Callable<String> { |
| volatile boolean iAmSpecial; |
| |
| private final String result; |
| volatile boolean ran; |
| |
| C(String result) { |
| this.result = result; |
| } |
| |
| @Override |
| public String call() throws Exception { |
| //Only one will be allowed to run, the rest |
| //will be cancelled |
| if (!iAmSpecial) { |
| blocker.await(); |
| } |
| ran = true; |
| return result; |
| } |
| } |
| List<C> callables = new ArrayList<C>(count); |
| C special = new C("Special"); |
| special.iAmSpecial = true; |
| callables.add(special); |
| List<Future<String>> fs; |
| Set<String> names = new HashSet<String>(count); |
| for (int i = 0; i < count; i++) { |
| String name = "R" + i; |
| names.add(name); |
| C c = new C(name); |
| callables.add(c); |
| } |
| fs = rp.invokeAll(callables, 1000, TimeUnit.MILLISECONDS); |
| assertNotNull(fs); |
| for (Future<String> f : fs) { |
| assertTrue (f.isDone()); |
| } |
| for (C c : callables) { |
| if (c == special) { |
| assertTrue (c.ran); |
| } else { |
| assertFalse(c.ran); |
| } |
| } |
| } finally { |
| rp.stop(); |
| } |
| } |
| |
| public void testInvokeAllSingleThread() throws Exception { |
| int count = 20; |
| final CountDownLatch waitAll = new CountDownLatch(count); |
| final RequestProcessor rp = new RequestProcessor("TestRP", 1); |
| class C implements Callable<String> { |
| |
| private final String result; |
| |
| C(String result) { |
| this.result = result; |
| } |
| |
| @Override |
| public String call() throws Exception { |
| waitAll.countDown(); |
| return result; |
| } |
| } |
| List<C> l = new ArrayList<C>(count); |
| List<Future<String>> fs; |
| Set<String> names = new HashSet<String>(count); |
| for (int i = 0; i < count; i++) { |
| String name = "R" + i; |
| names.add(name); |
| C c = new C(name); |
| l.add(c); |
| } |
| fs = rp.invokeAll(l); |
| assertNotNull(fs); |
| Set<String> s = new HashSet<String>(count); |
| for (Future<String> f : fs) { |
| s.add(f.get()); |
| } |
| assertEquals(names, s); |
| } |
| |
| @RandomlyFails // NB-Core-Build #4165: res==null |
| public void testInvokeAny() throws Exception { |
| int count = 20; |
| final RequestProcessor rp = new RequestProcessor("TestRP", count + 1); |
| class C implements Callable<String> { |
| |
| private final String result; |
| |
| C(String result) { |
| this.result = result; |
| } |
| |
| @Override |
| public String call() throws Exception { |
| if (Thread.interrupted()) { |
| return null; |
| } |
| return result; |
| } |
| } |
| List<C> l = new ArrayList<C>(count); |
| Set<String> names = new HashSet<String>(count); |
| for (int i = 0; i < count; i++) { |
| String name = "R" + i; |
| names.add(name); |
| C c = new C(name); |
| l.add(c); |
| } |
| String res = rp.invokeAny(l); |
| assertNotNull(res); |
| assertTrue(res.startsWith("R")); |
| } |
| |
| public void testInvokeAnySingleThread() throws Exception { |
| int count = 1000; |
| final RequestProcessor rp = new RequestProcessor("TestRP", 20); |
| final CountDownLatch latch = new CountDownLatch(count); |
| final Set<Thread> ts = Collections.synchronizedSet(new HashSet<Thread>()); |
| class C implements Callable<String> { |
| |
| volatile boolean hasRun; |
| private final String name; |
| |
| C(String name) { |
| this.name = name; |
| } |
| |
| @Override |
| public String call() throws Exception { |
| latch.countDown(); |
| if (!"R17".equals(name)) { |
| //Block all but one thread until threads have entered |
| Thread.currentThread().suspend(); |
| } |
| hasRun = true; |
| return name; |
| } |
| } |
| List<C> l = new ArrayList<C>(count); |
| Set<String> names = new HashSet<String>(count); |
| for (int i = 0; i < count; i++) { |
| String name = "R" + i; |
| names.add(name); |
| C c = new C(name); |
| l.add(c); |
| } |
| String res = rp.invokeAny(l); |
| assertNotNull(res); |
| assertTrue(res.startsWith("R")); |
| int runCount = 0; |
| for (C c : l) { |
| if (c.hasRun) { |
| runCount++; |
| } |
| } |
| assertTrue("Not all " + count + " threads should have completed, but " + runCount + " did.", runCount < count); |
| for (Thread t : ts) { |
| t.resume(); |
| } |
| } |
| |
| public void testInvokeAnyWithTimeout() throws Exception { |
| int count = 20; |
| final RequestProcessor rp = new RequestProcessor("TestRP", count + 1); |
| final CountDownLatch latch = new CountDownLatch(1); |
| class C implements Callable<String> { |
| |
| volatile boolean hasRun; |
| private final String result; |
| |
| C(String result) { |
| this.result = result; |
| } |
| |
| @Override |
| public String call() throws Exception { |
| latch.await(); |
| if (Thread.interrupted()) { |
| return null; |
| } |
| hasRun = true; |
| return result; |
| } |
| } |
| List<C> l = new ArrayList<C>(count); |
| Set<String> names = new HashSet<String>(count); |
| for (int i = 0; i < count; i++) { |
| String name = "R" + i; |
| names.add(name); |
| C c = new C(name); |
| l.add(c); |
| } |
| //All threads are waiting on latch; we should time out |
| String res = rp.invokeAny(l, 400, TimeUnit.MILLISECONDS); |
| assertNull(res); |
| for (C c : l) { |
| assertFalse(c.hasRun); |
| } |
| } |
| |
| public void testCancellation() throws Exception { |
| final CountDownLatch latch = new CountDownLatch(1); |
| class C implements Callable<String> { |
| |
| volatile boolean hasRun; |
| volatile boolean interrupted; |
| |
| @Override |
| public String call() throws Exception { |
| try { |
| latch.await(); |
| } catch (InterruptedException e) { |
| interrupted = true; |
| return null; |
| } |
| if (Thread.interrupted()) { |
| interrupted = true; |
| return null; |
| } |
| hasRun = true; |
| return "Hello"; |
| } |
| } |
| C c = new C(); |
| Future<String> f = RequestProcessor.getDefault().submit(c); |
| f.cancel(true); |
| latch.countDown(); |
| String s = null; |
| try { |
| s = f.get(); |
| fail("CancellationException should have been thrown"); |
| } catch (CancellationException e) { |
| } |
| assertNull(s); |
| assertTrue(c.interrupted || !c.hasRun); |
| assertFalse(c.hasRun); |
| } |
| |
| public void testCancellablesGetCancelInvokedWithCallable() throws Exception { |
| final CountDownLatch latch = new CountDownLatch(1); |
| final CountDownLatch exit = new CountDownLatch(1); |
| class C implements Callable<String>, Cancellable { |
| |
| volatile boolean hasRun; |
| volatile boolean interrupted; |
| volatile boolean cancelled; |
| |
| @Override |
| public String call() throws Exception { |
| try { |
| try { |
| latch.await(); |
| } catch (InterruptedException e) { |
| interrupted = true; |
| return null; |
| } |
| if (Thread.interrupted()) { |
| interrupted = true; |
| return null; |
| } |
| if (cancelled) { |
| return null; |
| } |
| hasRun = true; |
| return "Hello"; |
| } finally { |
| exit.countDown(); |
| } |
| } |
| |
| @Override |
| public boolean cancel() { |
| cancelled = true; |
| exit.countDown(); |
| return true; |
| } |
| } |
| C c = new C(); |
| Future<String> f = RequestProcessor.getDefault().submit(c); |
| f.cancel(true); |
| assertTrue (c.cancelled); |
| latch.countDown(); |
| exit.await(); |
| String s = null; |
| try { |
| s = f.get(); |
| fail ("Should have gotten cancellation exception"); |
| } catch (CancellationException e) { |
| |
| } |
| } |
| |
| public void testCancellablesGetCancelInvokedWithRunnable() throws Exception { |
| final CountDownLatch latch = new CountDownLatch(1); |
| final CountDownLatch exit = new CountDownLatch(1); |
| class C implements Runnable, Cancellable { |
| |
| volatile boolean hasRun; |
| volatile boolean interrupted; |
| volatile boolean cancelled; |
| |
| @Override |
| public void run() { |
| try { |
| try { |
| latch.await(); |
| } catch (InterruptedException e) { |
| interrupted = true; |
| return; |
| } |
| if (Thread.interrupted()) { |
| interrupted = true; |
| return; |
| } |
| if (cancelled) { |
| return; |
| } |
| hasRun = true; |
| } finally { |
| exit.countDown(); |
| } |
| } |
| |
| @Override |
| public boolean cancel() { |
| cancelled = true; |
| exit.countDown(); |
| return true; |
| } |
| } |
| C c = new C(); |
| Future<?> f = RequestProcessor.getDefault().submit(c); |
| f.cancel(true); |
| assertTrue (c.cancelled); |
| latch.countDown(); |
| exit.await(); |
| try { |
| f.get(); |
| fail ("Should have gotten cancellation exception"); |
| } catch (CancellationException e) { |
| |
| } |
| assertFalse (c.hasRun); |
| } |
| |
| public void testCancellablesThatSayTheyCantBeCancelledAreNotCancelledViaFutureDotCancel() throws Exception { |
| final CountDownLatch latch = new CountDownLatch(1); |
| final CountDownLatch exit = new CountDownLatch(1); |
| class C implements Runnable, Cancellable { |
| |
| volatile boolean hasRun; |
| volatile boolean interrupted; |
| volatile boolean cancelCalled; |
| |
| @Override |
| public void run() { |
| try { |
| try { |
| latch.await(); |
| } catch (InterruptedException e) { |
| interrupted = true; |
| throw new AssertionError(e); |
| } |
| if (Thread.interrupted()) { |
| interrupted = true; |
| throw new AssertionError("Thread should not have been interrupted"); |
| } |
| hasRun = true; |
| } finally { |
| exit.countDown(); |
| } |
| } |
| |
| @Override |
| public boolean cancel() { |
| cancelCalled = true; |
| return false; |
| } |
| } |
| C c = new C(); |
| Future<?> f = RequestProcessor.getDefault().submit(c); |
| f.cancel(true); |
| assertFalse (f.isCancelled()); |
| assertTrue (c.cancelCalled); |
| latch.countDown(); |
| exit.await(); |
| f.get(); |
| assertFalse (f.isCancelled()); |
| assertTrue (c.hasRun); |
| } |
| |
| public void testInvokeAllCancellation() throws Exception { |
| int count = 20; |
| final CountDownLatch waitAll = new CountDownLatch(count); |
| final RequestProcessor rp = new RequestProcessor("TestRP", count); |
| class C implements Callable<String>, Cancellable { |
| |
| private final String result; |
| volatile boolean cancelCalled; |
| |
| C(String result) { |
| this.result = result; |
| } |
| |
| @Override |
| public String call() throws Exception { |
| waitAll.countDown(); |
| return cancelCalled ? null : result; |
| } |
| |
| @Override |
| public boolean cancel() { |
| cancelCalled = true; |
| return false; |
| } |
| } |
| List<C> l = new ArrayList<C>(count); |
| List<Future<String>> fs; |
| Set<String> names = new HashSet<String>(count); |
| for (int i = 0; i < count; i++) { |
| String name = "R" + i; |
| names.add(name); |
| C c = new C(name); |
| l.add(c); |
| } |
| fs = rp.invokeAll(l); |
| assertNotNull(fs); |
| Set<String> s = new HashSet<String>(count); |
| for (Future<String> f : fs) { |
| s.add(f.get()); |
| } |
| assertEquals(names, s); |
| } |
| |
| public void testCannotScheduleLongerThanIntegerMaxValue() throws Exception { |
| Runnable r = new Runnable() { |
| |
| @Override |
| public void run() { |
| fail ("Should not have been run"); |
| } |
| }; |
| try { |
| Future<?> f = RequestProcessor.getDefault().schedule(r, Long.MAX_VALUE, TimeUnit.DAYS); |
| f.cancel(true); |
| } catch (Exception e) {} |
| } |
| |
| public void testCannotScheduleNegativeDelay() throws Exception { |
| Runnable r = new Runnable() { |
| |
| @Override |
| public void run() { |
| fail ("Should not have been run"); |
| } |
| }; |
| try { |
| RequestProcessor.getDefault().schedule(r, -1L, TimeUnit.MILLISECONDS); |
| fail ("Negative value accepetd"); |
| } catch (Exception e) {} |
| try { |
| RequestProcessor.getDefault().scheduleAtFixedRate(r, -1L, 22L, TimeUnit.MILLISECONDS); |
| fail ("Negative value accepetd"); |
| } catch (Exception e) {} |
| try { |
| RequestProcessor.getDefault().scheduleAtFixedRate(r, 200, -22L, TimeUnit.MILLISECONDS); |
| fail ("Negative value accepetd"); |
| } catch (Exception e) {} |
| try { |
| RequestProcessor.getDefault().scheduleWithFixedDelay(r, -1L, 22L, TimeUnit.MILLISECONDS); |
| fail ("Negative value accepetd"); |
| } catch (Exception e) {} |
| try { |
| RequestProcessor.getDefault().scheduleWithFixedDelay(r, 1L, -22L, TimeUnit.MILLISECONDS); |
| fail ("Negative value accepetd"); |
| } catch (Exception e) {} |
| } |
| |
| public void testTaskCanRescheduleItself() throws Exception { |
| final CountDownLatch latch = new CountDownLatch(2); |
| class R implements Runnable { |
| volatile RequestProcessor.Task task; |
| volatile int runCount; |
| @Override |
| public void run() { |
| runCount++; |
| if (runCount == 1) { |
| task.schedule(0); |
| } |
| latch.countDown(); |
| } |
| } |
| R r = new R(); |
| RequestProcessor.Task t = RequestProcessor.getDefault().create(r); |
| r.task = t; |
| t.schedule(0); |
| latch.await (); |
| assertEquals (r.runCount, 2); |
| } |
| |
| public void testScheduleRepeatingSanityFixedRate() throws Exception { |
| final CountDownLatch latch = new CountDownLatch(5); |
| class C implements Runnable { |
| volatile int runCount; |
| @Override |
| public void run() { |
| runCount++; |
| latch.countDown(); |
| if (latch.getCount() <= 0) { |
| waitABitToGiveMainThreadChanceToRun(); |
| } |
| } |
| private void waitABitToGiveMainThreadChanceToRun() { |
| try { |
| Thread.sleep(50); |
| } catch (InterruptedException ex) { |
| Exceptions.printStackTrace(ex); |
| } |
| } |
| } |
| C c = new C(); |
| RequestProcessor.getDefault().scheduleWithFixedDelay(c, 0, 200, TimeUnit.MILLISECONDS); |
| // latch.await(5000, TimeUnit.MILLISECONDS); |
| latch.await(); |
| assertAtLeast (5, c.runCount); |
| } |
| |
| public void testScheduleRepeatingSanityFixedDelay() throws Exception { |
| final CountDownLatch latch = new CountDownLatch(5); |
| class C implements Runnable { |
| volatile int runCount; |
| @Override |
| public void run() { |
| runCount++; |
| latch.countDown(); |
| waitABitToGiveMainThreadChanceToRun(); |
| } |
| |
| private void waitABitToGiveMainThreadChanceToRun() { |
| try { |
| Thread.sleep(50); |
| } catch (InterruptedException ex) { |
| Exceptions.printStackTrace(ex); |
| } |
| } |
| } |
| C c = new C(); |
| RequestProcessor.getDefault().scheduleAtFixedRate(c, 0, 200, TimeUnit.MILLISECONDS); |
| latch.await(2000, TimeUnit.MILLISECONDS); |
| |
| assertAtLeast (5, c.runCount); |
| } |
| |
| public void testScheduleOneShot() throws Exception { |
| RequestProcessor rp = new RequestProcessor ("testScheduleOneShot", 5, true, true); |
| try { |
| class C implements Callable<String> { |
| volatile long start = System.currentTimeMillis(); |
| private volatile long end; |
| |
| @Override |
| public String call() throws Exception { |
| synchronized(this) { |
| end = System.currentTimeMillis(); |
| } |
| return "Hello"; |
| } |
| |
| synchronized long elapsed() { |
| return end - start; |
| } |
| } |
| C c = new C(); |
| long delay = 5000; |
| //Use a 20 second timeout to have a reasonable chance of accuracy |
| ScheduledFuture<String> f = rp.schedule(c, delay * 1000, TimeUnit.MICROSECONDS); |
| assertEquals (5000, f.getDelay(TimeUnit.MILLISECONDS)); |
| assertNotNull(f.get()); |
| //Allow 4 seconds fudge-factor |
| assertTrue (c.elapsed() > 4600); |
| assertTrue (f.isDone()); |
| } finally { |
| rp.stop(); |
| } |
| } |
| |
| @RandomlyFails // NB-Core-Build #8322: hung |
| public void testScheduleRepeatingIntervalsAreRoughlyCorrect() throws Exception { |
| int runCount = 5; |
| final CountDownLatch latch = new CountDownLatch(runCount); |
| final List<Long> intervals = Collections.synchronizedList(new ArrayList<Long> (runCount)); |
| // long initialDelay = 30000; |
| // long period = 20000; |
| // long fudgeFactor = 4000; |
| long initialDelay = 3000; |
| long period = 2000; |
| long fudgeFactor = 400; |
| long expectedInitialDelay = initialDelay - fudgeFactor; |
| long expectedPeriod = period - fudgeFactor; |
| class C implements Runnable { |
| volatile long start = System.currentTimeMillis(); |
| private int runCount; |
| @Override |
| public void run() { |
| runCount++; |
| try { |
| synchronized(this) { |
| long end = System.currentTimeMillis(); |
| intervals.add (end - start); |
| start = end; |
| } |
| } finally { |
| latch.countDown(); |
| } |
| } |
| } |
| C c = new C(); |
| RequestProcessor rp = new RequestProcessor ("testScheduleRepeating", 5, true); |
| try { |
| Future<?> f = rp.scheduleWithFixedDelay(c, initialDelay, period, TimeUnit.MILLISECONDS); |
| // latch.await(initialDelay + fudgeFactor + ((runCount - 1) * (period + fudgeFactor)), TimeUnit.MILLISECONDS); //XXX |
| latch.await(); |
| f.cancel(true); |
| for (int i= 0; i < Math.min(runCount, intervals.size()); i++) { |
| long expect = i == 0 ? expectedInitialDelay : expectedPeriod; |
| assertTrue ("Expected at least " + expect + " milliseconds before run " + i + " but was " + intervals.get(i), intervals.get(i) >= expect); |
| } |
| //Ensure we have really exited |
| try { |
| f.get(); |
| fail ("CancellationException should have been thrown"); |
| } catch (CancellationException e) {} |
| assertTrue(f.isCancelled()); |
| assertTrue(f.isDone()); |
| } finally { |
| rp.stop(); |
| } |
| } |
| |
| @RandomlyFails |
| public void testScheduleFixedRateAreRoughlyCorrect() throws Exception { |
| if (!TaskTest.canWait1s()) { |
| LOG.warning("Skipping testWaitWithTimeOutReturnsAfterTimeOutWhenTheTaskIsNotComputedAtAll, as the computer is not able to wait 1s!"); |
| return; |
| } |
| int runCount = 5; |
| final CountDownLatch latch = new CountDownLatch(runCount); |
| final List<Long> intervals = Collections.synchronizedList(new ArrayList<Long> (runCount)); |
| // long initialDelay = 30000; |
| // long period = 20000; |
| // long fudgeFactor = 4000; |
| long initialDelay = 3000; |
| long period = 2000; |
| long fudgeFactor = 400; |
| long expectedInitialDelay = initialDelay - fudgeFactor; |
| long expectedPeriod = period - fudgeFactor; |
| class C implements Runnable { |
| volatile long start = System.currentTimeMillis(); |
| private int runCount; |
| @Override |
| public void run() { |
| runCount++; |
| try { |
| synchronized(this) { |
| long end = System.currentTimeMillis(); |
| intervals.add (end - start); |
| start = end; |
| } |
| } finally { |
| latch.countDown(); |
| } |
| } |
| } |
| C c = new C(); |
| RequestProcessor rp = new RequestProcessor ("testScheduleFixedRateAreRoughlyCorrect", 5, true); |
| try { |
| Future<?> f = rp.scheduleAtFixedRate(c, initialDelay, period, TimeUnit.MILLISECONDS); |
| latch.await(); |
| f.cancel(true); |
| StringBuilder failures = new StringBuilder(); |
| failures.append("Expected at least ").append(expectedInitialDelay). |
| append(" milliseconds before run:\n"); |
| boolean fail = false; |
| for (int i= 0; i < Math.min(runCount, intervals.size()); i++) { |
| long expect = i == 0 ? expectedInitialDelay : expectedPeriod; |
| failures.append("Round ").append(i). |
| append(" expected delay ").append(expect). |
| append(" but was ").append(intervals.get(i)). |
| append("\n"); |
| if (intervals.get(i) < expect) { |
| fail = true; |
| } |
| } |
| if (fail) { |
| fail(failures.toString()); |
| } |
| //Ensure we have really exited |
| try { |
| f.get(); |
| fail ("CancellationException should have been thrown"); |
| } catch (CancellationException e) {} |
| assertTrue(f.isCancelled()); |
| assertTrue(f.isDone()); |
| } finally { |
| rp.stop(); |
| } |
| } |
| |
| public void testScheduleFixedRateOnMultiThreadPoolDoesNotCauseConcurrentExecution() throws Exception { |
| final AtomicInteger val = new AtomicInteger(0); |
| final CountDownLatch latch = new CountDownLatch(10); |
| class C implements Runnable { |
| boolean failed; |
| @Override |
| public void run() { |
| try { |
| int now = val.incrementAndGet(); |
| if (now > 1) { |
| failed = true; |
| fail (now + " threads simultaneously in run()"); |
| } |
| try { |
| //Intentionally sleep *longer* than the interval |
| //between executions. We *want* to pile up all of the |
| //RP threads entering run() - synchronization should |
| //serialize them. This test is to prove that this |
| //method will never be called concurrently from two threads |
| Thread.sleep(1000); |
| } catch (InterruptedException ex) { |
| |
| } |
| } finally { |
| val.decrementAndGet(); |
| latch.countDown(); |
| } |
| } |
| } |
| C c = new C(); |
| long initialDelay = 2000; |
| long period = 10; |
| RequestProcessor rp = new RequestProcessor("testScheduleFixedRateOnMultiThreadPoolDoesNotCauseConcurrentExecution", 10, true); |
| rp.scheduleAtFixedRate(c, initialDelay, period, TimeUnit.MILLISECONDS); |
| latch.await(); |
| assertFalse(c.failed); |
| rp.stop(); |
| } |
| |
| @RandomlyFails |
| public void testScheduleFixedRateWithShorterIntervalThanRunMethodTimeAreNotDelayed() throws Exception { |
| final CountDownLatch latch = new CountDownLatch(10); |
| final List<Long> intervals = new CopyOnWriteArrayList<Long>(); |
| class C implements Runnable { |
| long start = Long.MIN_VALUE; |
| |
| @Override |
| public void run() { |
| long end = System.currentTimeMillis(); |
| if (start != Long.MIN_VALUE) { |
| intervals.add(end - start); |
| } |
| try { |
| Thread.sleep(500); |
| } catch (InterruptedException ex) { |
| |
| } |
| start = System.currentTimeMillis(); |
| latch.countDown(); |
| } |
| } |
| C c = new C(); |
| long initialDelay = 100; |
| long period = 100; |
| RequestProcessor rp = new RequestProcessor("testScheduleFixedRateWithShorterIntervalThanRunMethodTimeAreNotDelayed", 10, true); |
| ScheduledFuture<?> f = rp.scheduleAtFixedRate(c, initialDelay, period, TimeUnit.MILLISECONDS); |
| latch.await(); |
| f.cancel(true); |
| rp.stop(); |
| int max = intervals.size(); |
| for (int i= 0; i < max; i++) { |
| long iv = intervals.get(i); |
| assertFalse ("Interval " + i + " should have been at least less than requested interval * 1.5 with fixed rate" + iv, iv > 150); |
| } |
| } |
| |
| public void testCancelFutureInterruptsThreadEvenIfRequestProcessorForbidsIt() throws Exception { |
| RequestProcessor rp = new RequestProcessor ("X", 3, false, true); |
| final CountDownLatch releaseForRun = new CountDownLatch(1); |
| final CountDownLatch enterLatch = new CountDownLatch(1); |
| final CountDownLatch exitLatch = new CountDownLatch(1); |
| class R implements Runnable { |
| volatile boolean interrupted; |
| @Override |
| public void run() { |
| enterLatch.countDown(); |
| try { |
| releaseForRun.await(); |
| } catch (InterruptedException ex) { |
| interrupted = true; |
| } |
| interrupted |= Thread.interrupted(); |
| exitLatch.countDown(); |
| } |
| } |
| R r = new R(); |
| Future<?> f = rp.submit(r); |
| enterLatch.await(); |
| f.cancel(true); |
| assertTrue (f.isCancelled()); |
| exitLatch.await(); |
| assertTrue (r.interrupted); |
| } |
| |
| public void testCancelDoesNotInterruptIfNotPassedToFutureDotCancel() throws Exception { |
| RequestProcessor rp = new RequestProcessor ("X", 3, false, true); |
| final CountDownLatch releaseForRun = new CountDownLatch(1); |
| final CountDownLatch enterLatch = new CountDownLatch(1); |
| final CountDownLatch exitLatch = new CountDownLatch(1); |
| class R implements Runnable { |
| volatile boolean interrupted; |
| @Override |
| public void run() { |
| enterLatch.countDown(); |
| try { |
| releaseForRun.await(); |
| } catch (InterruptedException ex) { |
| interrupted = true; |
| } |
| interrupted |= Thread.interrupted(); |
| exitLatch.countDown(); |
| } |
| } |
| R r = new R(); |
| Future<?> f = rp.submit(r); |
| enterLatch.await(); |
| f.cancel(false); |
| assertTrue (f.isCancelled()); |
| assertFalse (r.interrupted); |
| } |
| |
| public void testSubmittedTasksExecutedBeforeShutdown() throws InterruptedException { |
| final CountDownLatch startLatch = new CountDownLatch(1); |
| final CountDownLatch executedLatch = new CountDownLatch(2); |
| Runnable dummyRunnable = new Runnable() { |
| |
| @Override |
| public void run() { |
| try { |
| startLatch.await(); |
| Thread.sleep(100); |
| } catch (InterruptedException ex) { |
| Thread.currentThread().interrupt(); |
| } finally { |
| executedLatch.countDown(); |
| } |
| } |
| }; |
| |
| RequestProcessor rp = new RequestProcessor("X", 1); |
| rp.submit(dummyRunnable); |
| rp.submit(dummyRunnable); |
| rp.shutdown(); |
| startLatch.countDown(); |
| |
| assertTrue("Submitted tasks not executed", executedLatch.await(5, TimeUnit.SECONDS)); |
| } |
| |
| public void testExecutingTasksNotInterruptedOnShutdown() throws InterruptedException { |
| final CountDownLatch startLatch = new CountDownLatch(2); |
| final CountDownLatch blockingLatch = new CountDownLatch(1); |
| final CountDownLatch executedLatch = new CountDownLatch(1); |
| final AtomicBoolean interrupted = new AtomicBoolean(false); |
| Runnable dummyRunnable = new Runnable() { |
| |
| @Override |
| public void run() { |
| try { |
| startLatch.countDown(); |
| startLatch.await(); |
| blockingLatch.await(); |
| } catch (InterruptedException ex) { |
| interrupted.set(true); |
| Thread.currentThread().interrupt(); |
| } finally { |
| executedLatch.countDown(); |
| } |
| } |
| }; |
| |
| RequestProcessor rp = new RequestProcessor("X", 1); |
| rp.submit(dummyRunnable); |
| startLatch.countDown(); |
| try { |
| startLatch.await(); |
| } catch (InterruptedException ex) { |
| Exceptions.printStackTrace(ex); |
| } |
| rp.shutdown(); |
| blockingLatch.countDown(); |
| executedLatch.await(); |
| assertFalse("Executing tasks interrupted", interrupted.get()); |
| } |
| |
| public void testAwaitingTasksReturnedOnShutdownNow() throws InterruptedException { |
| final CountDownLatch startupLatch = new CountDownLatch(2); |
| final CountDownLatch blockingLatch = new CountDownLatch(1); |
| Runnable blockingRunnable = new Runnable() { |
| |
| @Override |
| public void run() { |
| try { |
| startupLatch.countDown(); |
| startupLatch.await(); |
| blockingLatch.await(); |
| } catch (InterruptedException ex) { |
| Thread.currentThread().interrupt(); |
| } |
| } |
| }; |
| |
| Runnable awaitingRunnable = new Runnable() { |
| |
| @Override |
| public void run() { |
| // noop |
| } |
| }; |
| |
| RequestProcessor rp = new RequestProcessor("X", 1); |
| rp.submit(blockingRunnable); |
| startupLatch.countDown(); |
| startupLatch.await(); |
| rp.submit(awaitingRunnable); |
| Set<Runnable> awaiting = new HashSet<Runnable>(rp.shutdownNow()); |
| assertTrue("Awaiting task not returned on shutdownNow()", awaiting.contains(awaitingRunnable)); |
| assertFalse("Running task returned on shutdownNow()", awaiting.contains(blockingRunnable)); |
| } |
| |
| private static void assertAtLeast(int exp, int real) { |
| if (exp > real) { |
| fail("Expecting at least " + exp + " but was only " + real); |
| } |
| } |
| } |
