下面列出了java.util.concurrent.atomic.AtomicBoolean#notifyAll() 实例代码，或者点击链接到github查看源代码，也可以在右侧发表评论。
/** GemStoneAddition - tell the waiter it can start its countdown now */
public final void wakeWaiter(AtomicBoolean waiter_sync) {
synchronized(waiter_sync) {
waiter_sync.set(true);
waiter_sync.notifyAll();
}
}
/** GemStoneAddition - tell the waiter it can start its countdown now */
public final void wakeWaiter(AtomicBoolean waiter_sync) {
synchronized(waiter_sync) {
waiter_sync.set(true);
waiter_sync.notifyAll();
}
}
@Override
public Void call(ConfigBag parameters) {
Entity child = Iterables.getOnlyElement(entity().getChildren());
AtomicBoolean lock = new AtomicBoolean();
Task<Void> dummyTask = null;
try {
// Queue a (DST secondary) task which waits until notified, so that tasks queued later will get blocked
queue(Effectors.invocation(entity(), STALL, ImmutableMap.of("lock", lock)));
// Start a new task - submitted directly to child's ExecutionContext, as well as added as a
// DST secondary of the current effector.
dummyTask = child.invoke(DUMMY, ImmutableMap.<String, Object>of());
dummyTask.getUnchecked();
// Execution completed in the child's ExecutionContext, but still queued as a secondary.
// Destroy the child entity so that no subsequent tasks can be executed in its context.
Entities.destroy(child);
} finally {
// Let STALL complete
synchronized(lock) {
lock.set(true);
lock.notifyAll();
}
// At this point DUMMY will be unblocked and the DST will try to execute it as a secondary.
// Submission will be ignored because DUMMY already executed.
// If it's not ignored then submission will fail because entity is already unmanaged.
}
return null;
}
@Test(timeout = 10000)
public void testInterrupt() throws Exception {
Atomix atomix = atomix();
DistributedSemaphore semaphore = atomix.semaphoreBuilder("test-semaphore-interrupt")
.withProtocol(protocol())
.withInitialCapacity(10)
.build();
DistributedSemaphore semaphore2 = atomix.semaphoreBuilder("test-semaphore-interrupt")
.withProtocol(protocol())
.withInitialCapacity(10)
.build();
AtomicBoolean interrupted = new AtomicBoolean();
Thread t = new Thread(() -> {
try {
semaphore.acquire(11);
} catch (InterruptedException e) {
synchronized (interrupted) {
interrupted.set(true);
interrupted.notifyAll();
}
}
});
t.start();
synchronized (interrupted) {
t.interrupt();
interrupted.wait();
}
assertTrue(interrupted.get());
semaphore2.increasePermits(1);
// wait asynchronous release.
Thread.sleep(1000);
assertEquals(11, semaphore.availablePermits());
}
@Test(timeout = 10000)
public void testInterrupt() throws Exception {
Atomix atomix = atomix();
AtomicSemaphore semaphore = atomix.atomicSemaphoreBuilder("test-semaphore-interrupt")
.withProtocol(protocol())
.withInitialCapacity(10)
.build();
AtomicSemaphore semaphore2 = atomix.atomicSemaphoreBuilder("test-semaphore-interrupt")
.withProtocol(protocol())
.withInitialCapacity(10)
.build();
AtomicBoolean interrupted = new AtomicBoolean();
Thread t = new Thread(() -> {
try {
semaphore.acquire(11);
} catch (PrimitiveException.Interrupted e) {
synchronized (interrupted) {
interrupted.set(true);
interrupted.notifyAll();
}
}
});
t.start();
synchronized (interrupted) {
t.interrupt();
interrupted.wait();
}
assertTrue(interrupted.get());
semaphore2.increasePermits(1);
// wait asynchronous release.
Thread.sleep(1000);
assertEquals(11, semaphore.availablePermits());
}
@Test
public void testExceptionPropagationReader() {
try {
// use atomic boolean as a boolean reference
final AtomicBoolean handlerCalled = new AtomicBoolean();
final AtomicBoolean exceptionForwarded = new AtomicBoolean();
ReadRequest req = new ReadRequest() {
@Override
public void requestDone(IOException ioex) {
if (ioex instanceof TestIOException) {
exceptionForwarded.set(true);
}
synchronized (handlerCalled) {
handlerCalled.set(true);
handlerCalled.notifyAll();
}
}
@Override
public void read() throws IOException {
throw new TestIOException();
}
};
// test the read queue
RequestQueue<ReadRequest> rq = ioManager.getReadRequestQueue(ioManager.createChannel());
rq.add(req);
// wait until the asynchronous request has been handled
synchronized (handlerCalled) {
while (!handlerCalled.get()) {
handlerCalled.wait();
}
}
assertTrue(exceptionForwarded.get());
}
catch (Exception e) {
e.printStackTrace();
fail(e.getMessage());
}
}
@Test
public void testExceptionPropagationWriter() {
try {
// use atomic boolean as a boolean reference
final AtomicBoolean handlerCalled = new AtomicBoolean();
final AtomicBoolean exceptionForwarded = new AtomicBoolean();
WriteRequest req = new WriteRequest() {
@Override
public void requestDone(IOException ioex) {
if (ioex instanceof TestIOException) {
exceptionForwarded.set(true);
}
synchronized (handlerCalled) {
handlerCalled.set(true);
handlerCalled.notifyAll();
}
}
@Override
public void write() throws IOException {
throw new TestIOException();
}
};
// test the read queue
RequestQueue<WriteRequest> rq = ioManager.getWriteRequestQueue(ioManager.createChannel());
rq.add(req);
// wait until the asynchronous request has been handled
synchronized (handlerCalled) {
while (!handlerCalled.get()) {
handlerCalled.wait();
}
}
assertTrue(exceptionForwarded.get());
}
catch (Exception e) {
e.printStackTrace();
fail(e.getMessage());
}
}
@Test
public void testExceptionPropagationReader() {
try {
// use atomic boolean as a boolean reference
final AtomicBoolean handlerCalled = new AtomicBoolean();
final AtomicBoolean exceptionForwarded = new AtomicBoolean();
ReadRequest req = new ReadRequest() {
@Override
public void requestDone(IOException ioex) {
if (ioex instanceof TestIOException) {
exceptionForwarded.set(true);
}
synchronized (handlerCalled) {
handlerCalled.set(true);
handlerCalled.notifyAll();
}
}
@Override
public void read() throws IOException {
throw new TestIOException();
}
};
// test the read queue
RequestQueue<ReadRequest> rq = ioManager.getReadRequestQueue(ioManager.createChannel());
rq.add(req);
// wait until the asynchronous request has been handled
synchronized (handlerCalled) {
while (!handlerCalled.get()) {
handlerCalled.wait();
}
}
assertTrue(exceptionForwarded.get());
}
catch (Exception e) {
e.printStackTrace();
fail(e.getMessage());
}
}
@Test
public void testExceptionPropagationWriter() {
try {
// use atomic boolean as a boolean reference
final AtomicBoolean handlerCalled = new AtomicBoolean();
final AtomicBoolean exceptionForwarded = new AtomicBoolean();
WriteRequest req = new WriteRequest() {
@Override
public void requestDone(IOException ioex) {
if (ioex instanceof TestIOException) {
exceptionForwarded.set(true);
}
synchronized (handlerCalled) {
handlerCalled.set(true);
handlerCalled.notifyAll();
}
}
@Override
public void write() throws IOException {
throw new TestIOException();
}
};
// test the read queue
RequestQueue<WriteRequest> rq = ioManager.getWriteRequestQueue(ioManager.createChannel());
rq.add(req);
// wait until the asynchronous request has been handled
synchronized (handlerCalled) {
while (!handlerCalled.get()) {
handlerCalled.wait();
}
}
assertTrue(exceptionForwarded.get());
}
catch (Exception e) {
e.printStackTrace();
fail(e.getMessage());
}
}
public void notify(AtomicBoolean running) {
running.set(false);
synchronized (running) {
running.notifyAll();
}
}
@Test
public void testExceptionPropagationReader() {
try {
// use atomic boolean as a boolean reference
final AtomicBoolean handlerCalled = new AtomicBoolean();
final AtomicBoolean exceptionForwarded = new AtomicBoolean();
ReadRequest req = new ReadRequest() {
@Override
public void requestDone(IOException ioex) {
if (ioex instanceof TestIOException) {
exceptionForwarded.set(true);
}
synchronized (handlerCalled) {
handlerCalled.set(true);
handlerCalled.notifyAll();
}
}
@Override
public void read() throws IOException {
throw new TestIOException();
}
};
// test the read queue
RequestQueue<ReadRequest> rq = ioManager.getReadRequestQueue(ioManager.createChannel());
rq.add(req);
// wait until the asynchronous request has been handled
synchronized (handlerCalled) {
while (!handlerCalled.get()) {
handlerCalled.wait();
}
}
assertTrue(exceptionForwarded.get());
}
catch (Exception e) {
e.printStackTrace();
fail(e.getMessage());
}
}
@Test
public void testExceptionPropagationWriter() {
try {
// use atomic boolean as a boolean reference
final AtomicBoolean handlerCalled = new AtomicBoolean();
final AtomicBoolean exceptionForwarded = new AtomicBoolean();
WriteRequest req = new WriteRequest() {
@Override
public void requestDone(IOException ioex) {
if (ioex instanceof TestIOException) {
exceptionForwarded.set(true);
}
synchronized (handlerCalled) {
handlerCalled.set(true);
handlerCalled.notifyAll();
}
}
@Override
public void write() throws IOException {
throw new TestIOException();
}
};
// test the read queue
RequestQueue<WriteRequest> rq = ioManager.getWriteRequestQueue(ioManager.createChannel());
rq.add(req);
// wait until the asynchronous request has been handled
synchronized (handlerCalled) {
while (!handlerCalled.get()) {
handlerCalled.wait();
}
}
assertTrue(exceptionForwarded.get());
}
catch (Exception e) {
e.printStackTrace();
fail(e.getMessage());
}
}
@Test
public void testMetricsCollect() throws Exception {
int maxThreads = 5;
int maxTries = 10;
int sleepInterval = 10;
Server mockedServer = mock(Server.class);
when(mockedServer.getConfiguration()).thenReturn(HBaseConfiguration.create());
// Start an executor service pool with max 5 threads
ExecutorService executorService = new ExecutorService("unit_test");
executorService.startExecutorService(
ExecutorType.RS_PARALLEL_SEEK, maxThreads);
MetricsRegionServerSource serverSource = CompatibilitySingletonFactory
.getInstance(MetricsRegionServerSourceFactory.class).createServer(null);
assertTrue(serverSource instanceof MetricsRegionServerSourceImpl);
ExecutorStatusChore statusChore = new ExecutorStatusChore(60000,
mockedServer, executorService, serverSource);
AtomicBoolean lock = new AtomicBoolean(true);
AtomicInteger counter = new AtomicInteger(0);
for (int i = 0; i < maxThreads + 1; i++) {
executorService.submit(new TestEventHandler(mockedServer,
EventType.RS_PARALLEL_SEEK, lock, counter));
}
// The TestEventHandler will increment counter when it starts.
int tries = 0;
while (counter.get() < maxThreads && tries < maxTries) {
LOG.info("Waiting for all event handlers to start...");
Thread.sleep(sleepInterval);
tries++;
}
// Assert that pool is at max threads.
assertEquals(maxThreads, counter.get());
statusChore.chore();
Pair<Long, Long> executorStatus = statusChore.getExecutorStatus("RS_PARALLEL_SEEK");
assertEquals(maxThreads, executorStatus.getFirst().intValue()); // running
assertEquals(1, executorStatus.getSecond().intValue()); // pending
// Now interrupt the running Executor
synchronized (lock) {
lock.set(false);
lock.notifyAll();
}
executorService.shutdown();
}
@Test
public void testExecutorService() throws Exception {
int maxThreads = 5;
int maxTries = 10;
int sleepInterval = 10;
Server mockedServer = mock(Server.class);
when(mockedServer.getConfiguration()).thenReturn(HBaseConfiguration.create());
// Start an executor service pool with max 5 threads
ExecutorService executorService = new ExecutorService("unit_test");
executorService.startExecutorService(
ExecutorType.MASTER_SERVER_OPERATIONS, maxThreads);
Executor executor =
executorService.getExecutor(ExecutorType.MASTER_SERVER_OPERATIONS);
ThreadPoolExecutor pool = executor.threadPoolExecutor;
// Assert no threads yet
assertEquals(0, pool.getPoolSize());
AtomicBoolean lock = new AtomicBoolean(true);
AtomicInteger counter = new AtomicInteger(0);
// Submit maxThreads executors.
for (int i = 0; i < maxThreads; i++) {
executorService.submit(
new TestEventHandler(mockedServer, EventType.M_SERVER_SHUTDOWN,
lock, counter));
}
// The TestEventHandler will increment counter when it starts.
int tries = 0;
while (counter.get() < maxThreads && tries < maxTries) {
LOG.info("Waiting for all event handlers to start...");
Thread.sleep(sleepInterval);
tries++;
}
// Assert that pool is at max threads.
assertEquals(maxThreads, counter.get());
assertEquals(maxThreads, pool.getPoolSize());
ExecutorStatus status = executor.getStatus();
assertTrue(status.queuedEvents.isEmpty());
assertEquals(5, status.running.size());
checkStatusDump(status);
// Now interrupt the running Executor
synchronized (lock) {
lock.set(false);
lock.notifyAll();
}
// Executor increments counter again on way out so.... test that happened.
while (counter.get() < (maxThreads * 2) && tries < maxTries) {
System.out.println("Waiting for all event handlers to finish...");
Thread.sleep(sleepInterval);
tries++;
}
assertEquals(maxThreads * 2, counter.get());
assertEquals(maxThreads, pool.getPoolSize());
// Add more than the number of threads items.
// Make sure we don't get RejectedExecutionException.
for (int i = 0; i < (2 * maxThreads); i++) {
executorService.submit(
new TestEventHandler(mockedServer, EventType.M_SERVER_SHUTDOWN,
lock, counter));
}
// Now interrupt the running Executor
synchronized (lock) {
lock.set(false);
lock.notifyAll();
}
// Make sure threads are still around even after their timetolive expires.
Thread.sleep(ExecutorService.Executor.keepAliveTimeInMillis * 2);
assertEquals(maxThreads, pool.getPoolSize());
executorService.shutdown();
assertEquals(0, executorService.getAllExecutorStatuses().size());
// Test that submit doesn't throw NPEs
executorService.submit(
new TestEventHandler(mockedServer, EventType.M_SERVER_SHUTDOWN,
lock, counter));
}