This is an automated email from the ASF dual-hosted git repository.
adoroszlai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new b3a18ae057 HDDS-10039. Remove the flushLatches set from FlushNotifier.
(#5900)
b3a18ae057 is described below
commit b3a18ae057e3cb3b3ebfe5e1c69b0ea35a2fbe19
Author: Tsz-Wo Nicholas Sze <[email protected]>
AuthorDate: Wed Jan 10 00:47:13 2024 -0800
HDDS-10039. Remove the flushLatches set from FlushNotifier. (#5900)
---
.../ozone/om/ratis/OzoneManagerDoubleBuffer.java | 64 ++++++++++++++++------
.../om/ratis/TestOzoneManagerDoubleBuffer.java | 36 ++++--------
2 files changed, 57 insertions(+), 43 deletions(-)
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerDoubleBuffer.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerDoubleBuffer.java
index 1b4e3cafc4..96bad99e87 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerDoubleBuffer.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerDoubleBuffer.java
@@ -29,11 +29,10 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
-import java.util.Set;
+import java.util.TreeMap;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
@@ -597,7 +596,7 @@ public final class OzoneManagerDoubleBuffer {
*/
private synchronized boolean canFlush() {
try {
- while (currentBuffer.size() == 0) {
+ while (currentBuffer.isEmpty()) {
// canFlush() only gets called when the readyBuffer is empty.
// Since both buffers are empty, notify once for each.
flushNotifier.notifyFlush();
@@ -649,29 +648,58 @@ public final class OzoneManagerDoubleBuffer {
isRunning.set(true);
}
+ CompletableFuture<Integer> awaitFlushAsync() {
+ return flushNotifier.await();
+ }
+
public void awaitFlush() throws InterruptedException {
- flushNotifier.await();
+ try {
+ awaitFlushAsync().get();
+ } catch (ExecutionException e) {
+ // the future will never be completed exceptionally.
+ throw new IllegalStateException(e);
+ }
}
static class FlushNotifier {
- private final Set<CountDownLatch> flushLatches =
- ConcurrentHashMap.newKeySet();
+ static class Entry {
+ private final CompletableFuture<Integer> future = new
CompletableFuture<>();
+ private int count;
- void await() throws InterruptedException {
+ private CompletableFuture<Integer> await() {
+ count++;
+ return future;
+ }
+
+ private int complete() {
+ Preconditions.checkState(future.complete(count));
+ return future.join();
+ }
+ }
- // Wait until both the current and ready buffers are flushed.
- CountDownLatch latch = new CountDownLatch(2);
- flushLatches.add(latch);
- latch.await();
- flushLatches.remove(latch);
+ /** The size of the map is at most two since it uses {@link #flushCount} +
2 in {@link #await()} .*/
+ private final Map<Integer, Entry> flushFutures = new TreeMap<>();
+ private int awaitCount;
+ private int flushCount;
+
+ synchronized CompletableFuture<Integer> await() {
+ awaitCount++;
+ final int flush = flushCount + 2;
+ LOG.debug("await flush {}", flush);
+ final Entry entry = flushFutures.computeIfAbsent(flush, key -> new
Entry());
+ Preconditions.checkState(flushFutures.size() <= 2);
+ return entry.await();
}
- int notifyFlush() {
- int retval = flushLatches.size();
- for (CountDownLatch l : flushLatches) {
- l.countDown();
+ synchronized int notifyFlush() {
+ final int await = awaitCount;
+ final int flush = ++flushCount;
+ final Entry removed = flushFutures.remove(flush);
+ if (removed != null) {
+ awaitCount -= removed.complete();
}
- return retval;
+ LOG.debug("notifyFlush {}, awaitCount: {} -> {}", flush, await,
awaitCount);
+ return await;
}
}
}
diff --git
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerDoubleBuffer.java
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerDoubleBuffer.java
index 729eca6f47..859d54eb5f 100644
---
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerDoubleBuffer.java
+++
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerDoubleBuffer.java
@@ -265,8 +265,7 @@ class TestOzoneManagerDoubleBuffer {
}
@Test
- public void testAwaitFlush()
- throws ExecutionException, InterruptedException {
+ public void testAwaitFlush() throws Exception {
List<OMClientResponse> omClientResponses =
Arrays.asList(omKeyCreateResponse,
omBucketCreateResponse);
@@ -287,9 +286,9 @@ class TestOzoneManagerDoubleBuffer {
notifyCounter.incrementAndGet();
assertEquals(0, doubleBuffer.getCurrentBufferSize());
assertEquals(0, doubleBuffer.getReadyBufferSize());
- flushNotifier.notifyFlush();
- return null;
+ return flushNotifier.notifyFlush();
}).when(spyFlushNotifier).notifyFlush();
+ doAnswer(i -> flushNotifier.await()).when(spyFlushNotifier).await();
// Init double buffer.
for (OMClientResponse omClientResponse : omClientResponses) {
@@ -299,7 +298,7 @@ class TestOzoneManagerDoubleBuffer {
doubleBuffer.getCurrentBufferSize());
// Start double buffer and wait for flush.
- Future<?> await = awaitFlush(executorService);
+ final Future<?> await = awaitFlush();
Future<Boolean> flusher = flushTransactions(executorService);
await.get();
@@ -312,8 +311,7 @@ class TestOzoneManagerDoubleBuffer {
assertEquals(0, doubleBuffer.getReadyBufferSize());
// Run again to make sure it works when double buffer is empty
- await = awaitFlush(executorService);
- await.get();
+ awaitFlush().get();
// Clean up.
flusher.cancel(false);
@@ -405,11 +403,8 @@ class TestOzoneManagerDoubleBuffer {
}
// Return a future that waits for the flush.
- private Future<Boolean> awaitFlush(ExecutorService executorService) {
- return executorService.submit(() -> {
- doubleBuffer.awaitFlush();
- return true;
- });
+ private Future<?> awaitFlush() {
+ return doubleBuffer.awaitFlushAsync();
}
private Future<Boolean> flushTransactions(ExecutorService executorService) {
@@ -433,12 +428,11 @@ class TestOzoneManagerDoubleBuffer {
// Confirm nothing waiting yet.
assertEquals(0, fn.notifyFlush());
- ExecutorService executorService = Executors.newCachedThreadPool();
List<Future<Boolean>> tasks = new ArrayList<>();
// Simulate 3 waiting.
for (int i = 0; i < 3; i++) {
- tasks.add(waitFN(fn, executorService));
+ tasks.add(waitFN(fn));
}
Thread.sleep(2000);
@@ -449,7 +443,7 @@ class TestOzoneManagerDoubleBuffer {
assertEquals(3, fn.notifyFlush());
// Add a fourth.
- tasks.add(waitFN(fn, executorService));
+ tasks.add(waitFN(fn));
Thread.sleep(2000);
assertEquals(4, fn.notifyFlush());
@@ -467,15 +461,7 @@ class TestOzoneManagerDoubleBuffer {
}
- // Have a thread wait until notified.
- private Future<Boolean> waitFN(OzoneManagerDoubleBuffer.FlushNotifier fn,
- ExecutorService executorService) {
- return executorService.submit(() -> {
- try {
- fn.await();
- } catch (InterruptedException e) {
- }
- return true;
- });
+ private static Future<Boolean> waitFN(OzoneManagerDoubleBuffer.FlushNotifier
fn) {
+ return fn.await().thenApply(n -> true);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
