Repository: incubator-distributedlog Updated Branches: refs/heads/master 50f6bd40a -> ebaf14580
Fix deadlock on BKSyncLogReaderDLSN Change the lastSeenDLSN to volatile to remove the synchronization block to avoid deadlock with sharedLock Author: Khurrum Nasim <khurrumnas...@gmail.com> Reviewers: Leigh Stewart <lstew...@apache.org> Closes #42 from khurrumnasimm/kn/fix_deadlock Project: http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/commit/ebaf1458 Tree: http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/tree/ebaf1458 Diff: http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/diff/ebaf1458 Branch: refs/heads/master Commit: ebaf145807813c2731f3afbcd30fb526a1d677da Parents: 50f6bd4 Author: Khurrum Nasim <khurrumnas...@gmail.com> Authored: Tue Nov 29 11:50:17 2016 -0800 Committer: Sijie Guo <sij...@twitter.com> Committed: Tue Nov 29 11:50:17 2016 -0800 ---------------------------------------------------------------------- .../distributedlog/BKSyncLogReaderDLSN.java | 30 +++++++------------- 1 file changed, 11 insertions(+), 19 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/ebaf1458/distributedlog-core/src/main/java/com/twitter/distributedlog/BKSyncLogReaderDLSN.java ---------------------------------------------------------------------- diff --git a/distributedlog-core/src/main/java/com/twitter/distributedlog/BKSyncLogReaderDLSN.java b/distributedlog-core/src/main/java/com/twitter/distributedlog/BKSyncLogReaderDLSN.java index cef5ddb..ac670c2 100644 --- a/distributedlog-core/src/main/java/com/twitter/distributedlog/BKSyncLogReaderDLSN.java +++ b/distributedlog-core/src/main/java/com/twitter/distributedlog/BKSyncLogReaderDLSN.java @@ -51,7 +51,7 @@ class BKSyncLogReaderDLSN implements LogReader, Runnable, FutureEventListener<Lo private Promise<Void> closeFuture; private final Optional<Long> startTransactionId; private final DLSN startDLSN; - private DLSN lastSeenDLSN = DLSN.InvalidDLSN; + private volatile DLSN lastSeenDLSN = DLSN.InvalidDLSN; // lock on variables that would be accessed by both background threads and foreground threads private final Object sharedLock = new Object(); @@ -101,12 +101,6 @@ class BKSyncLogReaderDLSN implements LogReader, Runnable, FutureEventListener<Lo } } - private synchronized void setLastSeenDLSN(DLSN dlsn) { - synchronized (sharedLock) { - this.lastSeenDLSN = dlsn; - } - } - // Background Read Future Listener @Override @@ -116,7 +110,7 @@ class BKSyncLogReaderDLSN implements LogReader, Runnable, FutureEventListener<Lo @Override public void onSuccess(LogRecordWithDLSN record) { - setLastSeenDLSN(record.getDlsn()); + this.lastSeenDLSN = record.getDlsn(); if (!startTransactionId.isPresent() || record.getTransactionId() >= startTransactionId.get()) { readAheadRecords.add(record); } @@ -173,17 +167,15 @@ class BKSyncLogReaderDLSN implements LogReader, Runnable, FutureEventListener<Lo if (null != record) { break; } - synchronized (sharedLock) { - DLSN lastDLSNSeenByReadAhead = - reader.bkLedgerManager.readAheadCache.getLastReadAheadUserDLSN(); - - // if last seen DLSN by reader is same as the one seen by ReadAhead - // that means that reader is caught up with ReadAhead and ReadAhead - // is caught up with stream - shallWait = DLSN.InitialDLSN != lastDLSNSeenByReadAhead - && lastSeenDLSN.compareTo(lastDLSNSeenByReadAhead) < 0 - && startDLSN.compareTo(lastDLSNSeenByReadAhead) <= 0; - } + DLSN lastDLSNSeenByReadAhead = + reader.bkLedgerManager.readAheadCache.getLastReadAheadUserDLSN(); + + // if last seen DLSN by reader is same as the one seen by ReadAhead + // that means that reader is caught up with ReadAhead and ReadAhead + // is caught up with stream + shallWait = DLSN.InitialDLSN != lastDLSNSeenByReadAhead + && lastSeenDLSN.compareTo(lastDLSNSeenByReadAhead) < 0 + && startDLSN.compareTo(lastDLSNSeenByReadAhead) <= 0; } } catch (InterruptedException e) { throw new DLInterruptedException("Interrupted on waiting next available log record for stream "