From f71ab0dddeb7c26822978f820d1aaf8f32c13611 Mon Sep 17 00:00:00 2001 From: Kezhu Wang Date: Fri, 17 Dec 2021 18:19:03 +0800 Subject: [PATCH 1/4] Add testing code to deadlock Journal.ForceWriteThread on forceWriteRequests.put --- .../bookie/BookieWriteToJournalTest.java | 28 +++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieWriteToJournalTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieWriteToJournalTest.java index 873bb33db21..28c3e60e31e 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieWriteToJournalTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieWriteToJournalTest.java @@ -42,6 +42,7 @@ import org.apache.bookkeeper.conf.TestBKConfiguration; import org.apache.bookkeeper.net.BookieId; import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.WriteCallback; +import org.apache.bookkeeper.util.DiskChecker; import org.apache.commons.lang3.mutable.MutableBoolean; import org.junit.Rule; import org.junit.Test; @@ -205,6 +206,33 @@ public void testForceLedger() throws Exception { b.shutdown(); } + @Test + public void testSmallJournalQueueWithHighFlushFrequency() throws IOException, InterruptedException { + ServerConfiguration configuration = new ServerConfiguration(); + configuration.setJournalQueueSize(1); + configuration.setJournalFlushWhenQueueEmpty(true); + configuration.setJournalBufferedWritesThreshold(1); + + configuration.setJournalDirName(tempDir.newFolder().getPath()); + configuration.setLedgerDirNames(new String[]{tempDir.newFolder().getPath()}); + DiskChecker diskChecker = new DiskChecker(configuration.getDiskUsageThreshold(), configuration.getDiskUsageWarnThreshold()); + LedgerDirsManager ledgerDirsManager = new LedgerDirsManager(configuration, configuration.getLedgerDirs(), diskChecker); + Journal journal = new Journal(0, configuration.getJournalDirs()[0], configuration, ledgerDirsManager); + journal.start(); + + final int entries = 1000; + CountDownLatch entriesLatch = new CountDownLatch(entries); + for (int j = 1; j <= entries; j++) { + ByteBuf entry = buildEntry(1, j, -1); + journal.logAddEntry(entry, false, (int rc, long _ledgerId, long entryId, BookieId addr, Object ctx) -> { + entriesLatch.countDown(); + }, null); + } + entriesLatch.await(); + + journal.shutdown(); + } + private static ByteBuf buildEntry(long ledgerId, long entryId, long lastAddConfirmed) { final ByteBuf data = Unpooled.buffer(); data.writeLong(ledgerId); From b19f6bb3c6d4402474e6da3f05a29b74e2186c7b Mon Sep 17 00:00:00 2001 From: Kezhu Wang Date: Fri, 24 Dec 2021 16:40:02 +0800 Subject: [PATCH 2/4] Send force write group marker non-blocking to avoid deadlock ForceWriteThread Fixes #2948. --- .../main/java/org/apache/bookkeeper/bookie/Journal.java | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java index 2c054a1d93b..e89888c9ffe 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java @@ -497,6 +497,7 @@ public void run() { long busyStartTime = System.nanoTime(); while (running) { ForceWriteRequest req = null; + boolean forceWriteMarkerSent = false; try { forceWriteThreadTime.add(MathUtils.elapsedNanos(busyStartTime)); req = forceWriteRequests.take(); @@ -509,7 +510,7 @@ public void run() { // queue will benefit from this force write - post a marker prior to issuing // the flush so until this marker is encountered we can skip the force write if (enableGroupForceWrites) { - forceWriteRequests.put(createForceWriteRequest(req.logFile, 0, 0, null, false, true)); + forceWriteMarkerSent = forceWriteRequests.offer(createForceWriteRequest(req.logFile, 0, 0, null, false, true)); } // If we are about to issue a write, record the number of requests in @@ -527,6 +528,11 @@ public void run() { if (enableGroupForceWrites // if its a marker we should switch back to flushing && !req.isMarker + // If group marker sending failed, we can't figure out which writes are + // grouped in this force write. So, abandon it even if other writes could + // be grouped. This should be extremely rare as, usually, queue size is + // large enough to accommodate high flush frequencies. + && forceWriteMarkerSent // This indicates that this is the last request in a given file // so subsequent requests will go to a different file so we should // flush on the next request From 9f8980ef12dd002c2ddffb2049d39e1323c103bb Mon Sep 17 00:00:00 2001 From: Kezhu Wang Date: Sun, 9 Jan 2022 15:45:09 +0800 Subject: [PATCH 3/4] fixup! Add testing code to deadlock Journal.ForceWriteThread on forceWriteRequests.put --- .../bookie/BookieWriteToJournalTest.java | 22 +++++++++---------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieWriteToJournalTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieWriteToJournalTest.java index 28c3e60e31e..a03810fb1b4 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieWriteToJournalTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieWriteToJournalTest.java @@ -208,23 +208,23 @@ public void testForceLedger() throws Exception { @Test public void testSmallJournalQueueWithHighFlushFrequency() throws IOException, InterruptedException { - ServerConfiguration configuration = new ServerConfiguration(); - configuration.setJournalQueueSize(1); - configuration.setJournalFlushWhenQueueEmpty(true); - configuration.setJournalBufferedWritesThreshold(1); - - configuration.setJournalDirName(tempDir.newFolder().getPath()); - configuration.setLedgerDirNames(new String[]{tempDir.newFolder().getPath()}); - DiskChecker diskChecker = new DiskChecker(configuration.getDiskUsageThreshold(), configuration.getDiskUsageWarnThreshold()); - LedgerDirsManager ledgerDirsManager = new LedgerDirsManager(configuration, configuration.getLedgerDirs(), diskChecker); - Journal journal = new Journal(0, configuration.getJournalDirs()[0], configuration, ledgerDirsManager); + ServerConfiguration conf = new ServerConfiguration(); + conf.setJournalQueueSize(1); + conf.setJournalFlushWhenQueueEmpty(true); + conf.setJournalBufferedWritesThreshold(1); + + conf.setJournalDirName(tempDir.newFolder().getPath()); + conf.setLedgerDirNames(new String[]{tempDir.newFolder().getPath()}); + DiskChecker diskChecker = new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold()); + LedgerDirsManager ledgerDirsManager = new LedgerDirsManager(conf, conf.getLedgerDirs(), diskChecker); + Journal journal = new Journal(0, conf.getJournalDirs()[0], conf, ledgerDirsManager); journal.start(); final int entries = 1000; CountDownLatch entriesLatch = new CountDownLatch(entries); for (int j = 1; j <= entries; j++) { ByteBuf entry = buildEntry(1, j, -1); - journal.logAddEntry(entry, false, (int rc, long _ledgerId, long entryId, BookieId addr, Object ctx) -> { + journal.logAddEntry(entry, false, (int rc, long ledgerId, long entryId, BookieId addr, Object ctx) -> { entriesLatch.countDown(); }, null); } From 4365c30e6a814fad1d4792f3574ed9cc23da26ae Mon Sep 17 00:00:00 2001 From: Kezhu Wang Date: Sun, 9 Jan 2022 15:57:21 +0800 Subject: [PATCH 4/4] fixup! Send force write group marker non-blocking to avoid deadlock ForceWriteThread --- .../bookkeeper/bookie/BookKeeperServerStats.java | 1 + .../java/org/apache/bookkeeper/bookie/Journal.java | 14 +++++++++++++- .../bookkeeper/bookie/stats/JournalStats.java | 7 +++++++ 3 files changed, 21 insertions(+), 1 deletion(-) diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookKeeperServerStats.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookKeeperServerStats.java index 862affe284c..1e245a4dab6 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookKeeperServerStats.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookKeeperServerStats.java @@ -112,6 +112,7 @@ public interface BookKeeperServerStats { String JOURNAL_FORCE_WRITE_LATENCY = "JOURNAL_FORCE_WRITE_LATENCY"; String JOURNAL_FORCE_WRITE_BATCH_ENTRIES = "JOURNAL_FORCE_WRITE_BATCH_ENTRIES"; String JOURNAL_FORCE_WRITE_BATCH_BYTES = "JOURNAL_FORCE_WRITE_BATCH_BYTES"; + String JOURNAL_FORCE_WRITE_GROUPING_FAILURES = "JOURNAL_FORCE_WRITE_GROUPING_FAILURES"; String JOURNAL_FLUSH_LATENCY = "JOURNAL_FLUSH_LATENCY"; String JOURNAL_QUEUE_LATENCY = "JOURNAL_QUEUE_LATENCY"; String JOURNAL_QUEUE_MAX_SIZE = "JOURNAL_QUEUE_MAX_SIZE"; diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java index e89888c9ffe..de474984579 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java @@ -510,7 +510,19 @@ public void run() { // queue will benefit from this force write - post a marker prior to issuing // the flush so until this marker is encountered we can skip the force write if (enableGroupForceWrites) { - forceWriteMarkerSent = forceWriteRequests.offer(createForceWriteRequest(req.logFile, 0, 0, null, false, true)); + ForceWriteRequest marker = + createForceWriteRequest(req.logFile, 0, 0, null, false, true); + forceWriteMarkerSent = forceWriteRequests.offer(marker); + if (!forceWriteMarkerSent) { + marker.recycle(); + Counter failures = journalStats.getForceWriteGroupingFailures(); + failures.inc(); + LOG.error( + "Fail to send force write grouping marker," + + " Journal.forceWriteRequests queue(capacity {}) is full," + + " current failure counter is {}.", + conf.getJournalQueueSize(), failures.get()); + } } // If we are about to issue a write, record the number of requests in diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/stats/JournalStats.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/stats/JournalStats.java index 58124a9f8b8..133936de3e2 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/stats/JournalStats.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/stats/JournalStats.java @@ -31,6 +31,7 @@ import static org.apache.bookkeeper.bookie.BookKeeperServerStats.JOURNAL_FORCE_WRITE_BATCH_ENTRIES; import static org.apache.bookkeeper.bookie.BookKeeperServerStats.JOURNAL_FORCE_WRITE_ENQUEUE; import static org.apache.bookkeeper.bookie.BookKeeperServerStats.JOURNAL_FORCE_WRITE_GROUPING_COUNT; +import static org.apache.bookkeeper.bookie.BookKeeperServerStats.JOURNAL_FORCE_WRITE_GROUPING_FAILURES; import static org.apache.bookkeeper.bookie.BookKeeperServerStats.JOURNAL_FORCE_WRITE_QUEUE_SIZE; import static org.apache.bookkeeper.bookie.BookKeeperServerStats.JOURNAL_MEMORY_MAX; import static org.apache.bookkeeper.bookie.BookKeeperServerStats.JOURNAL_MEMORY_USED; @@ -132,6 +133,11 @@ public class JournalStats { help = "The distribution of number of bytes grouped together into a force write request" ) private final OpStatsLogger forceWriteBatchBytesStats; + @StatsDoc( + name = JOURNAL_FORCE_WRITE_GROUPING_FAILURES, + help = "The number of force write grouping failures" + ) + private final Counter forceWriteGroupingFailures; @StatsDoc( name = JOURNAL_QUEUE_SIZE, help = "The journal queue size" @@ -190,6 +196,7 @@ public JournalStats(StatsLogger statsLogger, final long maxJournalMemoryBytes, journalProcessTimeStats = statsLogger.getOpStatsLogger(BookKeeperServerStats.JOURNAL_PROCESS_TIME_LATENCY); forceWriteGroupingCountStats = statsLogger.getOpStatsLogger(BookKeeperServerStats.JOURNAL_FORCE_WRITE_GROUPING_COUNT); + forceWriteGroupingFailures = statsLogger.getCounter(JOURNAL_FORCE_WRITE_GROUPING_FAILURES); forceWriteBatchEntriesStats = statsLogger.getOpStatsLogger(BookKeeperServerStats.JOURNAL_FORCE_WRITE_BATCH_ENTRIES); forceWriteBatchBytesStats = statsLogger.getOpStatsLogger(BookKeeperServerStats.JOURNAL_FORCE_WRITE_BATCH_BYTES);