Repository: bookkeeper Updated Branches: refs/heads/master efd8ec269 -> b3b958c2b
BOOKKEEPER-1087: Ledger Recovery (part 2) - Add a parallel reading request in PendingReadOp THIS CHANGE IS BASED ON #176 (you can review f0fb89c) bookkeeper recovery improvement (part-2): add a parallel reading request in PendingReadOp - add a parallel reading request in PendingReadOp - allow PendingReadOp to configure whether to do parallel reading or not - add flag in ClientConfiguration to allow configuring whether to do parallel reading in LedgerRecoveryOp or not. Author: Sijie Guo <[email protected]> Author: Sijie Guo <[email protected]> Reviewers: Matteo Merli <[email protected]> Closes #177 from sijie/recovery_improvements_part2 Project: http://git-wip-us.apache.org/repos/asf/bookkeeper/repo Commit: http://git-wip-us.apache.org/repos/asf/bookkeeper/commit/b3b958c2 Tree: http://git-wip-us.apache.org/repos/asf/bookkeeper/tree/b3b958c2 Diff: http://git-wip-us.apache.org/repos/asf/bookkeeper/diff/b3b958c2 Branch: refs/heads/master Commit: b3b958c2b6a7195b23477d181db3fea1e7d1c248 Parents: efd8ec2 Author: Sijie Guo <[email protected]> Authored: Mon Jun 5 12:48:33 2017 -0700 Committer: Sijie Guo <[email protected]> Committed: Mon Jun 5 12:48:33 2017 -0700 ---------------------------------------------------------------------- .../apache/bookkeeper/client/LedgerHandle.java | 6 +- .../bookkeeper/client/LedgerRecoveryOp.java | 8 +- .../apache/bookkeeper/client/PendingReadOp.java | 89 +++++- .../bookkeeper/conf/ClientConfiguration.java | 23 +- .../bookkeeper/client/TestParallelRead.java | 276 +++++++++++++++++++ 5 files changed, 393 insertions(+), 9 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/b3b958c2/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java ---------------------------------------------------------------------- diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java index 43f6ff2..7fa8c61 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java @@ -1477,7 +1477,8 @@ public class LedgerHandle implements AutoCloseable { if (wasInRecovery) { // if metadata is already in recover, dont try to write again, // just do the recovery from the starting point - new LedgerRecoveryOp(LedgerHandle.this, cb).initiate(); + new LedgerRecoveryOp(LedgerHandle.this, cb) + .parallelRead(bk.getConf().getEnableParallelRecoveryRead()).initiate(); return; } @@ -1503,7 +1504,8 @@ public class LedgerHandle implements AutoCloseable { } }); } else if (rc == BKException.Code.OK) { - new LedgerRecoveryOp(LedgerHandle.this, cb).initiate(); + new LedgerRecoveryOp(LedgerHandle.this, cb) + .parallelRead(bk.getConf().getEnableParallelRecoveryRead()).initiate(); } else { LOG.error("Error writing ledger config " + rc + " of ledger " + ledgerId); cb.operationComplete(rc, null); http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/b3b958c2/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerRecoveryOp.java ---------------------------------------------------------------------- diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerRecoveryOp.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerRecoveryOp.java index 20e78ce..666dbe8 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerRecoveryOp.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerRecoveryOp.java @@ -49,6 +49,7 @@ class LedgerRecoveryOp implements ReadCallback, AddCallback { long entryToRead; // keep a copy of metadata for recovery. LedgerMetadata metadataForRecovery; + boolean parallelRead = false; GenericCallback<Void> cb; @@ -75,6 +76,11 @@ class LedgerRecoveryOp implements ReadCallback, AddCallback { this.lh = lh; } + LedgerRecoveryOp parallelRead(boolean enabled) { + this.parallelRead = enabled; + return this; + } + public void initiate() { ReadLastConfirmedOp rlcop = new ReadLastConfirmedOp(lh, new ReadLastConfirmedOp.LastConfirmedDataCallback() { @@ -110,7 +116,7 @@ class LedgerRecoveryOp implements ReadCallback, AddCallback { if (!callbackDone.get()) { entryToRead++; try { - new RecoveryReadOp(lh, lh.bk.scheduler, entryToRead, entryToRead, this, null).initiate(); + new RecoveryReadOp(lh, lh.bk.scheduler, entryToRead, entryToRead, this, null).parallelRead(parallelRead).initiate(); } catch (InterruptedException e) { readComplete(BKException.Code.InterruptedException, lh, null, null); } http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/b3b958c2/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingReadOp.java ---------------------------------------------------------------------- diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingReadOp.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingReadOp.java index ae566b5..7b01b7f 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingReadOp.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/PendingReadOp.java @@ -71,6 +71,8 @@ class PendingReadOp implements Enumeration<LedgerEntry>, ReadEntryCallback { OpStatsLogger readOpLogger; final int maxMissedReadsAllowed; + boolean parallelRead = false; + final AtomicBoolean complete = new AtomicBoolean(false); abstract class LedgerEntryRequest extends LedgerEntry { @@ -129,6 +131,22 @@ class PendingReadOp implements Enumeration<LedgerEntry>, ReadEntryCallback { } /** + * Fail the request with given result code <i>rc</i>. + * + * @param rc + * result code to fail the request. + * @return true if we managed to fail the entry; otherwise return false if it already failed or completed. + */ + boolean fail(int rc) { + if (complete.compareAndSet(false, true)) { + submitCallback(rc); + return true; + } else { + return false; + } + } + + /** * Log error <i>errMsg</i> and reattempt read from <i>host</i>. * * @param host @@ -138,7 +156,7 @@ class PendingReadOp implements Enumeration<LedgerEntry>, ReadEntryCallback { * @param rc * read result code */ - void logErrorAndReattemptRead(BookieSocketAddress host, String errMsg, int rc) { + synchronized void logErrorAndReattemptRead(BookieSocketAddress host, String errMsg, int rc) { if (BKException.Code.OK == firstError || BKException.Code.NoSuchEntryException == firstError || BKException.Code.NoSuchLedgerExistsException == firstError) { @@ -191,6 +209,52 @@ class PendingReadOp implements Enumeration<LedgerEntry>, ReadEntryCallback { } } + class ParallelReadRequest extends LedgerEntryRequest { + + int numPendings; + + ParallelReadRequest(ArrayList<BookieSocketAddress> ensemble, long lId, long eId) { + super(ensemble, lId, eId); + numPendings = writeSet.size(); + } + + @Override + void read() { + for (int bookieIndex : writeSet) { + BookieSocketAddress to = ensemble.get(bookieIndex); + try { + sendReadTo(to, this); + } catch (InterruptedException ie) { + LOG.error("Interrupted reading entry {} : ", this, ie); + Thread.currentThread().interrupt(); + fail(BKException.Code.InterruptedException); + return; + } + } + } + + @Override + synchronized void logErrorAndReattemptRead(BookieSocketAddress host, String errMsg, int rc) { + super.logErrorAndReattemptRead(host, errMsg, rc); + --numPendings; + // if received all responses or this entry doesn't meet quorum write, complete the request. + if (numMissedEntryReads > maxMissedReadsAllowed || numPendings == 0) { + if (BKException.Code.BookieHandleNotAvailableException == firstError && + numMissedEntryReads > maxMissedReadsAllowed) { + firstError = BKException.Code.NoSuchEntryException; + } + + fail(firstError); + } + } + + @Override + BookieSocketAddress maybeSendSpeculativeRead(Set<BookieSocketAddress> heardFromHosts) { + // no speculative read + return null; + } + } + class SequenceReadRequest extends LedgerEntryRequest { final static int NOT_FOUND = -1; int nextReplicaIndexToReadFrom = 0; @@ -280,7 +344,7 @@ class PendingReadOp implements Enumeration<LedgerEntry>, ReadEntryCallback { firstError = BKException.Code.NoSuchEntryException; } - submitCallback(firstError); + fail(firstError); return null; } @@ -296,7 +360,7 @@ class PendingReadOp implements Enumeration<LedgerEntry>, ReadEntryCallback { } catch (InterruptedException ie) { LOG.error("Interrupted reading entry " + this, ie); Thread.currentThread().interrupt(); - submitCallback(BKException.Code.ReadException); + fail(BKException.Code.InterruptedException); return null; } } @@ -347,12 +411,17 @@ class PendingReadOp implements Enumeration<LedgerEntry>, ReadEntryCallback { } } + PendingReadOp parallelRead(boolean enabled) { + this.parallelRead = enabled; + return this; + } + public void initiate() throws InterruptedException { long nextEnsembleChange = startEntryId, i = startEntryId; this.requestTimeNanos = MathUtils.nowInNano(); ArrayList<BookieSocketAddress> ensemble = null; - if (speculativeReadTimeout > 0) { + if (speculativeReadTimeout > 0 && !parallelRead) { Runnable readTask = new Runnable() { public void run() { int x = 0; @@ -393,7 +462,12 @@ class PendingReadOp implements Enumeration<LedgerEntry>, ReadEntryCallback { ensemble = getLedgerMetadata().getEnsemble(i); nextEnsembleChange = getLedgerMetadata().getNextEnsembleChange(i); } - LedgerEntryRequest entry = new SequenceReadRequest(ensemble, lh.ledgerId, i); + LedgerEntryRequest entry; + if (parallelRead) { + entry = new ParallelReadRequest(ensemble, lh.ledgerId, i); + } else { + entry = new SequenceReadRequest(ensemble, lh.ledgerId, i); + } seq.add(entry); i++; @@ -449,6 +523,11 @@ class PendingReadOp implements Enumeration<LedgerEntry>, ReadEntryCallback { return; } + // ensure callback once + if (!complete.compareAndSet(false, true)) { + return; + } + long latencyNanos = MathUtils.elapsedNanos(requestTimeNanos); if (code != BKException.Code.OK) { long firstUnread = LedgerHandle.INVALID_ENTRY_ID; http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/b3b958c2/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ClientConfiguration.java ---------------------------------------------------------------------- diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ClientConfiguration.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ClientConfiguration.java index ad025f2..5d4bf03 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ClientConfiguration.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ClientConfiguration.java @@ -62,6 +62,7 @@ public class ClientConfiguration extends AbstractConfiguration { // Read Parameters protected final static String READ_TIMEOUT = "readTimeout"; protected final static String SPECULATIVE_READ_TIMEOUT = "speculativeReadTimeout"; + protected final static String ENABLE_PARALLEL_RECOVERY_READ = "enableParallelRecoveryRead"; // Timeout Setting protected final static String ADD_ENTRY_TIMEOUT_SEC = "addEntryTimeoutSec"; protected final static String ADD_ENTRY_QUORUM_TIMEOUT_SEC = "addEntryQuorumTimeoutSec"; @@ -793,6 +794,27 @@ public class ClientConfiguration extends AbstractConfiguration { } /** + * Whether to enable parallel reading in recovery read. + * + * @return true if enable parallel reading in recovery read. otherwise, return false. + */ + public boolean getEnableParallelRecoveryRead() { + return getBoolean(ENABLE_PARALLEL_RECOVERY_READ, false); + } + + /** + * Enable/Disable parallel reading in recovery read. + * + * @param enabled + * flag to enable/disable parallel reading in recovery read. + * @return client configuration. + */ + public ClientConfiguration setEnableParallelRecoveryRead(boolean enabled) { + setProperty(ENABLE_PARALLEL_RECOVERY_READ, enabled); + return this; + } + + /** * Get Ensemble Placement Policy Class. * * @return ensemble placement policy class. @@ -1096,5 +1118,4 @@ public class ClientConfiguration extends AbstractConfiguration { public String getClientRole() { return getString(CLIENT_ROLE, CLIENT_ROLE_STANDARD); } - } http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/b3b958c2/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestParallelRead.java ---------------------------------------------------------------------- diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestParallelRead.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestParallelRead.java new file mode 100644 index 0000000..3b1ca19 --- /dev/null +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestParallelRead.java @@ -0,0 +1,276 @@ +/* + * + * 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.apache.bookkeeper.client; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import org.apache.bookkeeper.client.AsyncCallback.ReadCallback; +import org.apache.bookkeeper.client.BookKeeper.DigestType; +import org.apache.bookkeeper.conf.ClientConfiguration; +import org.apache.bookkeeper.net.BookieSocketAddress; +import org.apache.bookkeeper.test.BookKeeperClusterTestCase; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Enumeration; +import java.util.concurrent.CountDownLatch; + +/** + * Unit tests for parallel reading + */ +public class TestParallelRead extends BookKeeperClusterTestCase { + + static Logger LOG = LoggerFactory.getLogger(TestParallelRead.class); + + final DigestType digestType; + final byte[] passwd = "parallel-read".getBytes(); + + public TestParallelRead() { + super(6); + this.digestType = DigestType.CRC32; + } + + long getLedgerToRead(int ensemble, int writeQuorum, int ackQuorum, int numEntries) + throws Exception { + LedgerHandle lh = bkc.createLedger(ensemble, writeQuorum, ackQuorum, digestType, passwd); + for (int i = 0; i < numEntries; i++) { + lh.addEntry(("" + i).getBytes()); + } + lh.close(); + return lh.getId(); + } + + static class LatchCallback implements ReadCallback { + + final CountDownLatch l = new CountDownLatch(1); + int rc = -0x1314; + Enumeration<LedgerEntry> entries; + + Enumeration<LedgerEntry> getEntries() { + return entries; + } + + int getRc() { + return rc; + } + + @Override + public void readComplete(int rc, LedgerHandle lh, Enumeration<LedgerEntry> seq, Object ctx) { + this.rc = rc; + this.entries = seq; + l.countDown(); + } + + void expectSuccess() throws Exception { + l.await(); + assertTrue(BKException.Code.OK == rc); + } + + void expectFail() throws Exception { + l.await(); + assertFalse(BKException.Code.OK == rc); + } + + } + + @Test(timeout = 60000) + public void testNormalParallelRead() throws Exception { + int numEntries = 10; + + long id = getLedgerToRead(5, 2, 2, numEntries); + LedgerHandle lh = bkc.openLedger(id, digestType, passwd); + + // read single entry + for (int i = 0; i < numEntries; i++) { + LatchCallback latch = new LatchCallback(); + PendingReadOp readOp = + new PendingReadOp(lh, lh.bk.scheduler, i, i, latch, null); + readOp.parallelRead(true).initiate(); + latch.expectSuccess(); + Enumeration<LedgerEntry> entries = latch.getEntries(); + assertNotNull(entries); + assertTrue(entries.hasMoreElements()); + LedgerEntry entry = entries.nextElement(); + assertNotNull(entry); + assertEquals(i, Integer.parseInt(new String(entry.getEntry()))); + assertFalse(entries.hasMoreElements()); + } + + // read multiple entries + LatchCallback latch = new LatchCallback(); + PendingReadOp readOp = + new PendingReadOp(lh, lh.bk.scheduler, 0, numEntries - 1, latch, null); + readOp.parallelRead(true).initiate(); + latch.expectSuccess(); + Enumeration<LedgerEntry> entries = latch.getEntries(); + assertNotNull(entries); + + int numReads = 0; + while (entries.hasMoreElements()) { + LedgerEntry entry = entries.nextElement(); + assertNotNull(entry); + assertEquals(numReads, Integer.parseInt(new String(entry.getEntry()))); + ++numReads; + } + assertEquals(numEntries, numReads); + + lh.close(); + } + + @Test(timeout = 60000) + public void testParallelReadMissingEntries() throws Exception { + int numEntries = 10; + + long id = getLedgerToRead(5, 2, 2, numEntries); + LedgerHandle lh = bkc.openLedger(id, digestType, passwd); + + // read single entry + LatchCallback latch = new LatchCallback(); + PendingReadOp readOp = + new PendingReadOp(lh, lh.bk.scheduler, 11, 11, latch, null); + readOp.parallelRead(true).initiate(); + latch.expectFail(); + assertEquals(BKException.Code.NoSuchEntryException, latch.getRc()); + + // read multiple entries + latch = new LatchCallback(); + readOp = new PendingReadOp(lh, lh.bk.scheduler, 8, 11, latch, null); + readOp.parallelRead(true).initiate(); + latch.expectFail(); + assertEquals(BKException.Code.NoSuchEntryException, latch.getRc()); + + lh.close(); + } + + @Test(timeout = 60000) + public void testFailParallelReadMissingEntryImmediately() throws Exception { + int numEntries = 1; + + long id = getLedgerToRead(5, 5, 3, numEntries); + + ClientConfiguration newConf = new ClientConfiguration() + .setReadEntryTimeout(30000); + newConf.setZkServers(zkUtil.getZooKeeperConnectString()); + BookKeeper newBk = new BookKeeper(newConf); + + LedgerHandle lh = bkc.openLedger(id, digestType, passwd); + + ArrayList<BookieSocketAddress> ensemble = + lh.getLedgerMetadata().getEnsemble(10); + CountDownLatch latch1 = new CountDownLatch(1); + CountDownLatch latch2 = new CountDownLatch(1); + // sleep two bookie + sleepBookie(ensemble.get(0), latch1); + sleepBookie(ensemble.get(1), latch2); + + LatchCallback latchCallback = new LatchCallback(); + PendingReadOp readOp = + new PendingReadOp(lh, lh.bk.scheduler, 10, 10, latchCallback, null); + readOp.parallelRead(true).initiate(); + // would fail immediately if found missing entries don't cover ack quorum + latchCallback.expectFail(); + assertEquals(BKException.Code.NoSuchEntryException, latchCallback.getRc()); + latch1.countDown(); + latch2.countDown(); + + lh.close(); + newBk.close(); + } + + @Test(timeout = 60000) + public void testParallelReadWithFailedBookies() throws Exception { + int numEntries = 10; + + long id = getLedgerToRead(5, 3, 3, numEntries); + + ClientConfiguration newConf = new ClientConfiguration() + .setReadEntryTimeout(30000); + newConf.setZkServers(zkUtil.getZooKeeperConnectString()); + BookKeeper newBk = new BookKeeper(newConf); + + LedgerHandle lh = bkc.openLedger(id, digestType, passwd); + + ArrayList<BookieSocketAddress> ensemble = + lh.getLedgerMetadata().getEnsemble(5); + // kill two bookies + killBookie(ensemble.get(0)); + killBookie(ensemble.get(1)); + + // read multiple entries + LatchCallback latch = new LatchCallback(); + PendingReadOp readOp = + new PendingReadOp(lh, lh.bk.scheduler, 0, numEntries - 1, latch, null); + readOp.parallelRead(true).initiate(); + latch.expectSuccess(); + Enumeration<LedgerEntry> entries = latch.getEntries(); + assertNotNull(entries); + + int numReads = 0; + while (entries.hasMoreElements()) { + LedgerEntry entry = entries.nextElement(); + assertNotNull(entry); + assertEquals(numReads, Integer.parseInt(new String(entry.getEntry()))); + ++numReads; + } + assertEquals(numEntries, numReads); + + lh.close(); + newBk.close(); + } + + @Test(timeout = 60000) + public void testParallelReadFailureWithFailedBookies() throws Exception { + int numEntries = 10; + + long id = getLedgerToRead(5, 3, 3, numEntries); + + ClientConfiguration newConf = new ClientConfiguration() + .setReadEntryTimeout(30000); + newConf.setZkServers(zkUtil.getZooKeeperConnectString()); + BookKeeper newBk = new BookKeeper(newConf); + + LedgerHandle lh = bkc.openLedger(id, digestType, passwd); + + ArrayList<BookieSocketAddress> ensemble = + lh.getLedgerMetadata().getEnsemble(5); + // kill two bookies + killBookie(ensemble.get(0)); + killBookie(ensemble.get(1)); + killBookie(ensemble.get(2)); + + // read multiple entries + LatchCallback latch = new LatchCallback(); + PendingReadOp readOp = + new PendingReadOp(lh, lh.bk.scheduler, 0, numEntries - 1, latch, null); + readOp.parallelRead(true).initiate(); + latch.expectFail(); + assertEquals(BKException.Code.BookieHandleNotAvailableException, latch.getRc()); + + lh.close(); + newBk.close(); + } + +}
