You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zookeeper.apache.org by iv...@apache.org on 2014/05/30 17:41:39 UTC
svn commit: r1598644 - in /zookeeper/bookkeeper/branches/branch-4.2:
CHANGES.txt
bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
Author: ivank
Date: Fri May 30 15:41:38 2014
New Revision: 1598644
URL: http://svn.apache.org/r1598644
Log:
BOOKKEEPER-751: Ensure all the bookkeeper callbacks not run under ledger handle lock (sijie via ivank)
Modified:
zookeeper/bookkeeper/branches/branch-4.2/CHANGES.txt
zookeeper/bookkeeper/branches/branch-4.2/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
Modified: zookeeper/bookkeeper/branches/branch-4.2/CHANGES.txt
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/branches/branch-4.2/CHANGES.txt?rev=1598644&r1=1598643&r2=1598644&view=diff
==============================================================================
--- zookeeper/bookkeeper/branches/branch-4.2/CHANGES.txt (original)
+++ zookeeper/bookkeeper/branches/branch-4.2/CHANGES.txt Fri May 30 15:41:38 2014
@@ -32,6 +32,8 @@ Release 4.2.3 - 2013-12-04
BOOKKEEPER-673: Ledger length can be inaccurate in failure case (sijie via ivank)
+ BOOKKEEPER-751: Ensure all the bookkeeper callbacks not run under ledger handle lock (sijie via ivank)
+
IMPROVEMENT:
BOOKKEEPER-747: Implement register/unregister LedgerMetadataListener in MSLedgerManagerFactory (fpj via sijie)
Modified: zookeeper/bookkeeper/branches/branch-4.2/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/branches/branch-4.2/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java?rev=1598644&r1=1598643&r2=1598644&view=diff
==============================================================================
--- zookeeper/bookkeeper/branches/branch-4.2/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java (original)
+++ zookeeper/bookkeeper/branches/branch-4.2/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java Fri May 30 15:41:38 2014
@@ -26,6 +26,7 @@ import java.util.concurrent.ConcurrentLi
import java.util.Arrays;
import java.util.ArrayList;
import java.util.Enumeration;
+import java.util.List;
import java.util.Queue;
import java.util.concurrent.atomic.AtomicInteger;
@@ -250,6 +251,7 @@ public class LedgerHandle {
final long prevLastEntryId;
final long prevLength;
final State prevState;
+ List<PendingAddOp> pendingAdds;
synchronized(LedgerHandle.this) {
// if the metadata is already closed, we don't need to proceed the process
@@ -262,8 +264,8 @@ public class LedgerHandle {
prevLastEntryId = metadata.getLastEntryId();
prevLength = metadata.getLength();
- // error out pending adds first
- errorOutPendingAdds(rc);
+ // drain pending adds first
+ pendingAdds = drainPendingAddsToErrorOut();
// synchronized on LedgerHandle.this to ensure that
// lastAddPushed can not be updated after the metadata
@@ -273,6 +275,10 @@ public class LedgerHandle {
lastAddPushed = lastAddConfirmed;
}
+ // error out all pending adds during closing, the callbacks shouldn't be
+ // running under any bk locks.
+ errorOutPendingAdds(rc, pendingAdds);
+
if (LOG.isDebugEnabled()) {
LOG.debug("Closing ledger: " + ledgerId + " at entryId: "
+ metadata.getLastEntryId() + " with this many bytes: " + metadata.getLength());
@@ -492,21 +498,28 @@ public class LedgerHandle {
final long entryId;
final long currentLength;
+ boolean wasClosed = false;
synchronized(this) {
// synchronized on this to ensure that
// the ledger isn't closed between checking and
// updating lastAddPushed
if (metadata.isClosed()) {
- LOG.warn("Attempt to add to closed ledger: " + ledgerId);
- cb.addComplete(BKException.Code.LedgerClosedException,
- LedgerHandle.this, INVALID_ENTRY_ID, ctx);
- return;
+ wasClosed = true;
+ entryId = -1;
+ currentLength = 0;
+ } else {
+ entryId = ++lastAddPushed;
+ currentLength = addToLength(length);
+ op.setEntryId(entryId);
+ pendingAddOps.add(op);
}
+ }
- entryId = ++lastAddPushed;
- currentLength = addToLength(length);
- op.setEntryId(entryId);
- pendingAddOps.add(op);
+ if (wasClosed) {
+ LOG.warn("Attempt to add to closed ledger: {}", ledgerId);
+ cb.addComplete(BKException.Code.LedgerClosedException,
+ LedgerHandle.this, INVALID_ENTRY_ID, ctx);
+ return;
}
try {
@@ -644,10 +657,22 @@ public class LedgerHandle {
}
void errorOutPendingAdds(int rc) {
+ errorOutPendingAdds(rc, drainPendingAddsToErrorOut());
+ }
+
+ synchronized List<PendingAddOp> drainPendingAddsToErrorOut() {
PendingAddOp pendingAddOp;
+ List<PendingAddOp> opsDrained = new ArrayList<PendingAddOp>(pendingAddOps.size());
while ((pendingAddOp = pendingAddOps.poll()) != null) {
addToLength(-pendingAddOp.entryLength);
- pendingAddOp.submitCallback(rc);
+ opsDrained.add(pendingAddOp);
+ }
+ return opsDrained;
+ }
+
+ void errorOutPendingAdds(int rc, List<PendingAddOp> ops) {
+ for (PendingAddOp op : ops) {
+ op.submitCallback(rc);
}
}
@@ -896,25 +921,39 @@ public class LedgerHandle {
bk.getLedgerManager().readLedgerMetadata(ledgerId, cb);
}
- synchronized void recover(final GenericCallback<Void> cb) {
- if (metadata.isClosed()) {
- lastAddConfirmed = lastAddPushed = metadata.getLastEntryId();
- length = metadata.getLength();
+ void recover(final GenericCallback<Void> cb) {
+ boolean wasClosed = false;
+ boolean wasInRecovery = false;
+ synchronized (this) {
+ if (metadata.isClosed()) {
+ lastAddConfirmed = lastAddPushed = metadata.getLastEntryId();
+ length = metadata.getLength();
+ wasClosed = true;
+ } else {
+ wasClosed = false;
+ if (metadata.isInRecovery()) {
+ wasInRecovery = true;
+ } else {
+ wasInRecovery = false;
+ metadata.markLedgerInRecovery();
+ }
+ }
+ }
+
+ if (wasClosed) {
// We are already closed, nothing to do
cb.operationComplete(BKException.Code.OK, null);
return;
}
- // if metadata is already in recover, dont try to write again,
- // just do the recovery from the starting point
- if (metadata.isInRecovery()) {
+ 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();
return;
}
- metadata.markLedgerInRecovery();
-
writeLedgerConfig(new OrderedSafeGenericCallback<Void>(bk.mainWorkerPool, ledgerId) {
@Override
public void safeOperationComplete(final int rc, Void result) {