ivankelly commented on a change in pull request #1225: Issue #570: getting rid
of unnecessary synchronization in InterleavedLedgerStorage
URL: https://github.com/apache/bookkeeper/pull/1225#discussion_r172492072
##########
File path:
bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/EntryLogTest.java
##########
@@ -391,4 +402,174 @@ public void testGetEntryLogsSet() throws Exception {
assertEquals(Sets.newHashSet(0L, 1L, 2L, 3L),
logger.getEntryLogsSet());
}
+
+ static class LedgerStorageWriteTask implements Callable<Boolean> {
+ long ledgerId;
+ int entryId;
+ LedgerStorage ledgerStorage;
+
+ LedgerStorageWriteTask(long ledgerId, int entryId, LedgerStorage
ledgerStorage) {
+ this.ledgerId = ledgerId;
+ this.entryId = entryId;
+ this.ledgerStorage = ledgerStorage;
+ }
+
+ @Override
+ public Boolean call() throws IOException {
+ try {
+ ledgerStorage.addEntry(generateEntry(ledgerId, entryId));
+ } catch (IOException e) {
+ LOG.error("Got Exception for AddEntry call. LedgerId: " +
ledgerId + " entryId: " + entryId, e);
+ throw new IOException("Got Exception for AddEntry call.
LedgerId: " + ledgerId + " entryId: " + entryId,
+ e);
+ }
+ return true;
+ }
+ }
+
+ static class LedgerStorageFlushTask implements Callable<Boolean> {
+ LedgerStorage ledgerStorage;
+
+ LedgerStorageFlushTask(LedgerStorage ledgerStorage) {
+ this.ledgerStorage = ledgerStorage;
+ }
+
+ @Override
+ public Boolean call() throws IOException {
+ try {
+ ledgerStorage.flush();
+ } catch (IOException e) {
+ LOG.error("Got Exception for flush call", e);
+ throw new IOException("Got Exception for Flush call", e);
+ }
+ return true;
+ }
+ }
+
+ static class LedgerStorageReadTask implements Callable<Boolean> {
+ long ledgerId;
+ int entryId;
+ LedgerStorage ledgerStorage;
+
+ LedgerStorageReadTask(long ledgerId, int entryId, LedgerStorage
ledgerStorage) {
+ this.ledgerId = ledgerId;
+ this.entryId = entryId;
+ this.ledgerStorage = ledgerStorage;
+ }
+
+ @Override
+ public Boolean call() throws IOException {
+ try {
+ ByteBuf expectedByteBuf = generateEntry(ledgerId, entryId);
+ ByteBuf actualByteBuf = ledgerStorage.getEntry(ledgerId,
entryId);
+ if (!expectedByteBuf.equals(actualByteBuf)) {
+ LOG.error("Expected Entry: {} Actual Entry: {}",
expectedByteBuf.toString(Charset.defaultCharset()),
+ actualByteBuf.toString(Charset.defaultCharset()));
+ throw new IOException("Expected Entry: " +
expectedByteBuf.toString(Charset.defaultCharset())
+ + " Actual Entry: " +
actualByteBuf.toString(Charset.defaultCharset()));
+ }
+ } catch (IOException e) {
+ LOG.error("Got Exception for GetEntry call. LedgerId: " +
ledgerId + " entryId: " + entryId, e);
+ throw new IOException("Got Exception for GetEntry call.
LedgerId: " + ledgerId + " entryId: " + entryId,
+ e);
+ }
+ return true;
+ }
+ }
+
+ /**
+ * test concurrent write operations and then concurrent read
+ * operations using InterleavedLedgerStorage.
+ */
+ @Test
+ public void testConcurrentWriteAndReadCallsOfInterleavedLedgerStorage()
throws Exception {
+ File ledgerDir = createTempDir("bkTest", ".dir");
+ ServerConfiguration conf =
TestBKConfiguration.newServerConfiguration();
+ conf.setJournalDirName(ledgerDir.toString());
+ conf.setLedgerDirNames(new String[] { ledgerDir.getAbsolutePath()});
+ conf.setLedgerStorageClass(InterleavedLedgerStorage.class.getName());
+ Bookie bookie = new Bookie(conf);
+ InterleavedLedgerStorage ledgerStorage = ((InterleavedLedgerStorage)
bookie.ledgerStorage);
+ Random rand = new Random();
+
+ int numOfLedgers = 70;
+ int numEntries = 2000;
+ // Create ledgers
+ for (int i = 0; i < numOfLedgers; i++) {
+ ledgerStorage.setMasterKey(i, "key".getBytes());
+ }
+
+ ExecutorService executor = Executors.newFixedThreadPool(10);
+ List<Callable<Boolean>> writeAndFlushTasks = new
ArrayList<Callable<Boolean>>();
+ for (int j = 0; j < numEntries; j++) {
+ for (int i = 0; i < numOfLedgers; i++) {
+ writeAndFlushTasks.add(new LedgerStorageWriteTask(i, j,
ledgerStorage));
+ }
+ }
+
+ /*
+ * add some flush tasks to the list of writetasks list.
+ */
+ for (int i = 0; i < (numOfLedgers * numEntries) / 500; i++) {
+ writeAndFlushTasks.add(rand.nextInt(writeAndFlushTasks.size()),
new LedgerStorageFlushTask(ledgerStorage));
+ }
+
+ // invoke all those write tasks all at once concurrently and set
timeout
+ // 6 seconds for them to complete
+ List<Future<Boolean>> writeTasksFutures =
executor.invokeAll(writeAndFlushTasks, 6, TimeUnit.SECONDS);
Review comment:
actually, looking at the code again, the read path hasn't changed and was
never synchronized anyhow, so this only serves to verify that we can read what
the right task has written. Since this is the case, we should read back in a
straight loop, and not in tasks.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
