You are viewing a plain text version of this content. The canonical link for it is here.
Posted to oak-commits@jackrabbit.apache.org by mr...@apache.org on 2015/01/21 10:31:46 UTC
svn commit: r1653463 - in /jackrabbit/oak/trunk/oak-core/src:
main/java/org/apache/jackrabbit/oak/plugins/document/
test/java/org/apache/jackrabbit/oak/plugins/document/
Author: mreutegg
Date: Wed Jan 21 09:31:46 2015
New Revision: 1653463
URL: http://svn.apache.org/r1653463
Log:
OAK-2421: Clean up orphaned branches
Added:
jackrabbit/oak/trunk/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/OrphanedBranchTest.java (with props)
Modified:
jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/Branch.java
jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/Commit.java
jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentMK.java
jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentNodeStore.java
jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentNodeStoreBranch.java
jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/MergeCommit.java
jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/RevisionContext.java
jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/SplitOperations.java
jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/UnmergedBranches.java
jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/UnsavedModifications.java
jackrabbit/oak/trunk/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/CommitQueueTest.java
jackrabbit/oak/trunk/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/DocumentNodeStoreTest.java
jackrabbit/oak/trunk/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/DummyRevisionContext.java
Modified: jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/Branch.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/Branch.java?rev=1653463&r1=1653462&r2=1653463&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/Branch.java (original)
+++ jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/Branch.java Wed Jan 21 09:31:46 2015
@@ -20,6 +20,8 @@ import static com.google.common.base.Pre
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.collect.Iterables.transform;
+import java.lang.ref.ReferenceQueue;
+import java.lang.ref.WeakReference;
import java.util.NavigableMap;
import java.util.Set;
import java.util.SortedSet;
@@ -27,6 +29,7 @@ import java.util.concurrent.ConcurrentSk
import javax.annotation.CheckForNull;
import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
import com.google.common.base.Function;
import com.google.common.collect.Iterables;
@@ -49,15 +52,29 @@ class Branch {
private final Revision base;
/**
+ * The branch reference.
+ */
+ private final BranchReference ref;
+
+ /**
* Create a new branch instance with an initial set of commits and a given
- * base revision.
+ * base revision. The life time of this branch can be controlled with
+ * the {@code guard} parameter. Once the {@code guard} object becomes weakly
+ * reachable, the {@link BranchReference} for this branch is appended to
+ * the passed {@code queue}. No {@link BranchReference} is appended if the
+ * passed {@code guard} is {@code null}.
*
* @param commits the initial branch commits.
* @param base the base commit.
+ * @param queue a {@link BranchReference} to this branch will be appended to
+ * this queue when {@code guard} becomes weakly reachable.
+ * @param guard controls the life time of this branch.
* @throws IllegalArgumentException if base is a branch revision.
*/
Branch(@Nonnull SortedSet<Revision> commits,
- @Nonnull Revision base) {
+ @Nonnull Revision base,
+ @Nonnull ReferenceQueue<Object> queue,
+ @Nullable Object guard) {
checkArgument(!checkNotNull(base).isBranch(), "base is not a trunk revision: %s", base);
this.base = base;
this.commits = new ConcurrentSkipListMap<Revision, BranchCommit>(commits.comparator());
@@ -65,6 +82,11 @@ class Branch {
this.commits.put(r.asBranchRevision(),
new BranchCommitImpl(base, r.asBranchRevision()));
}
+ if (guard != null) {
+ this.ref = new BranchReference(queue, this, guard);
+ } else {
+ this.ref = null;
+ }
}
/**
@@ -161,6 +183,15 @@ class Branch {
}
/**
+ * @return the branch reference or {@code null} if no guard object was
+ * passed to the constructor of this branch.
+ */
+ @CheckForNull
+ BranchReference getRef() {
+ return ref;
+ }
+
+ /**
* Removes the commit with the given revision <code>r</code>. Does nothing
* if there is no such commit.
*
@@ -330,4 +361,20 @@ class Branch {
throw new UnsupportedOperationException("RebaseCommit is read-only");
}
}
+
+ final static class BranchReference extends WeakReference<Object> {
+
+ private final Branch branch;
+
+ private BranchReference(@Nonnull ReferenceQueue<Object> queue,
+ @Nonnull Branch branch,
+ @Nonnull Object referent) {
+ super(checkNotNull(referent), queue);
+ this.branch = checkNotNull(branch);
+ }
+
+ Branch getBranch() {
+ return branch;
+ }
+ }
}
Modified: jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/Commit.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/Commit.java?rev=1653463&r1=1653462&r2=1653463&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/Commit.java (original)
+++ jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/Commit.java Wed Jan 21 09:31:46 2015
@@ -37,6 +37,7 @@ import org.apache.jackrabbit.oak.commons
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.collect.Iterables.filter;
import static com.google.common.collect.Iterables.transform;
import static org.apache.jackrabbit.oak.commons.PathUtils.denotesRoot;
@@ -52,6 +53,7 @@ public class Commit {
private static final Logger LOG = LoggerFactory.getLogger(Commit.class);
private final DocumentNodeStore nodeStore;
+ private final DocumentNodeStoreBranch branch;
private final Revision baseRevision;
private final Revision revision;
private HashMap<String, UpdateOp> operations = new LinkedHashMap<String, UpdateOp>();
@@ -70,10 +72,25 @@ public class Commit {
/** Set of all nodes which have binary properties. **/
private HashSet<String> nodesWithBinaries = Sets.newHashSet();
- Commit(DocumentNodeStore nodeStore, Revision baseRevision, Revision revision) {
+ /**
+ * Create a new Commit.
+ *
+ * @param nodeStore the node store.
+ * @param revision the revision for this commit.
+ * @param baseRevision the base revision for this commit or {@code null} if
+ * there is none.
+ * @param branch the branch associated with this commit or {@code null} if
+ * there is none.
+ *
+ */
+ Commit(@Nonnull DocumentNodeStore nodeStore,
+ @Nonnull Revision revision,
+ @Nullable Revision baseRevision,
+ @Nullable DocumentNodeStoreBranch branch) {
+ this.nodeStore = checkNotNull(nodeStore);
+ this.revision = checkNotNull(revision);
this.baseRevision = baseRevision;
- this.revision = revision;
- this.nodeStore = nodeStore;
+ this.branch = branch;
}
UpdateOp getUpdateOperationForNode(String path) {
@@ -162,7 +179,8 @@ public class Commit {
Branch b = nodeStore.getBranches().getBranch(baseRev);
if (b == null) {
// baseRev is marker for new branch
- b = nodeStore.getBranches().create(baseRev.asTrunkRevision(), rev);
+ b = nodeStore.getBranches().create(
+ baseRev.asTrunkRevision(), rev, branch);
} else {
b.addCommit(rev);
}
Modified: jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentMK.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentMK.java?rev=1653463&r1=1653462&r2=1653463&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentMK.java (original)
+++ jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentMK.java Wed Jan 21 09:31:46 2015
@@ -269,7 +269,7 @@ public class DocumentMK implements Micro
boolean success = false;
boolean isBranch = false;
Revision rev;
- Commit commit = nodeStore.newCommit(baseRevId != null ? Revision.fromString(baseRevId) : null);
+ Commit commit = nodeStore.newCommit(baseRevId != null ? Revision.fromString(baseRevId) : null, null);
try {
Revision baseRev = commit.getBaseRevision();
isBranch = baseRev != null && baseRev.isBranch();
@@ -340,7 +340,7 @@ public class DocumentMK implements Micro
throw new MicroKernelException("Not a branch revision: " + ancestorRevisionId);
}
try {
- return nodeStore.reset(branch, ancestor).toString();
+ return nodeStore.reset(branch, ancestor, null).toString();
} catch (DocumentStoreException e) {
throw new MicroKernelException(e);
}
Modified: jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentNodeStore.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentNodeStore.java?rev=1653463&r1=1653462&r2=1653463&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentNodeStore.java (original)
+++ jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentNodeStore.java Wed Jan 21 09:31:46 2015
@@ -195,8 +195,6 @@ public final class DocumentNodeStore
/**
* Unmerged branches of this DocumentNodeStore instance.
*/
- // TODO at some point, open (unmerged) branches
- // need to be garbage collected (in-memory and on disk)
private final UnmergedBranches branches;
/**
@@ -398,7 +396,7 @@ public final class DocumentNodeStore
if (store.find(Collection.NODES, Utils.getIdFromPath("/")) == null) {
// root node is missing: repository is not initialized
Revision head = newRevision();
- Commit commit = new Commit(this, null, head);
+ Commit commit = new Commit(this, head, null, null);
DocumentNodeState n = new DocumentNodeState(this, "/", head);
commit.addNode(n);
commit.applyToDocumentStore();
@@ -488,11 +486,6 @@ public final class DocumentNodeStore
}
}
- @Nonnull
- Revision getHeadRevision() {
- return headRevision;
- }
-
Revision setHeadRevision(@Nonnull Revision newHead) {
checkArgument(!newHead.isBranch());
Revision previous = headRevision;
@@ -528,10 +521,16 @@ public final class DocumentNodeStore
*
* @param base the base revision for the commit or <code>null</code> if the
* commit should use the current head revision as base.
+ * @param branch the branch instance if this is a branch commit. The life
+ * time of this branch commit is controlled by the
+ * reachability of this parameter. Once {@code branch} is
+ * weakly reachable, the document store implementation is
+ * free to remove the commits associated with the branch.
* @return a new commit.
*/
@Nonnull
- Commit newCommit(@Nullable Revision base) {
+ Commit newCommit(@Nullable Revision base,
+ @Nullable DocumentNodeStoreBranch branch) {
if (base == null) {
base = headRevision;
}
@@ -539,7 +538,7 @@ public final class DocumentNodeStore
boolean success = false;
Commit c;
try {
- c = new Commit(this, base, commitQueue.createRevision());
+ c = new Commit(this, commitQueue.createRevision(), base, branch);
success = true;
} finally {
if (!success) {
@@ -1099,7 +1098,9 @@ public final class DocumentNodeStore
}
@Nonnull
- Revision reset(@Nonnull Revision branchHead, @Nonnull Revision ancestor) {
+ Revision reset(@Nonnull Revision branchHead,
+ @Nonnull Revision ancestor,
+ @Nullable DocumentNodeStoreBranch branch) {
checkNotNull(branchHead);
checkNotNull(ancestor);
Branch b = getBranches().getBranch(branchHead);
@@ -1119,7 +1120,7 @@ public final class DocumentNodeStore
return branchHead;
}
boolean success = false;
- Commit commit = newCommit(branchHead);
+ Commit commit = newCommit(branchHead, branch);
try {
Iterator<Revision> it = b.getCommits().tailSet(ancestor).iterator();
// first revision is the ancestor (tailSet is inclusive)
@@ -1443,6 +1444,11 @@ public final class DocumentNodeStore
return clusterId;
}
+ @Nonnull
+ public Revision getHeadRevision() {
+ return headRevision;
+ }
+
//----------------------< background operations >---------------------------
public synchronized void runBackgroundOperations() {
@@ -1456,6 +1462,11 @@ public final class DocumentNodeStore
try {
long start = clock.getTime();
long time = start;
+ // clean orphaned branches and collisions
+ cleanOrphanedBranches();
+ cleanCollisions();
+ long cleanTime = clock.getTime() - time;
+ time = clock.getTime();
// split documents (does not create new revisions)
backgroundSplit();
long splitTime = clock.getTime() - time;
@@ -1467,12 +1478,12 @@ public final class DocumentNodeStore
// pull in changes from other cluster nodes
backgroundRead(true);
long readTime = clock.getTime() - time;
- String msg = "Background operations stats (split:{}, write:{}, read:{})";
+ String msg = "Background operations stats (clean:{}, split:{}, write:{}, read:{})";
if (clock.getTime() - start > TimeUnit.SECONDS.toMillis(10)) {
// log as info if it took more than 10 seconds
- LOG.info(msg, splitTime, writeTime, readTime);
+ LOG.info(msg, cleanTime, splitTime, writeTime, readTime);
} else {
- LOG.debug(msg, splitTime, writeTime, readTime);
+ LOG.debug(msg, cleanTime, splitTime, writeTime, readTime);
}
} catch (RuntimeException e) {
if (isDisposed.get()) {
@@ -1602,6 +1613,47 @@ public final class DocumentNodeStore
return Revision.getCurrentTimestamp() - REMEMBER_REVISION_ORDER_MILLIS;
}
+ private void cleanOrphanedBranches() {
+ Branch b;
+ while ((b = branches.pollOrphanedBranch()) != null) {
+ LOG.debug("Cleaning up orphaned branch with base revision: {}, " +
+ "commits: {}", b.getBase(), b.getCommits());
+ UpdateOp op = new UpdateOp(Utils.getIdFromPath("/"), false);
+ for (Revision r : b.getCommits()) {
+ r = r.asTrunkRevision();
+ NodeDocument.removeRevision(op, r);
+ }
+ store.findAndUpdate(NODES, op);
+ }
+ }
+
+ private void cleanCollisions() {
+ String id = Utils.getIdFromPath("/");
+ NodeDocument root = store.find(NODES, id);
+ if (root == null) {
+ return;
+ }
+ Revision head = getHeadRevision();
+ Map<Revision, String> map = root.getLocalMap(NodeDocument.COLLISIONS);
+ UpdateOp op = new UpdateOp(id, false);
+ for (Revision r : map.keySet()) {
+ if (r.getClusterId() == clusterId) {
+ // remove collision if there is no active branch with
+ // this revision and the revision is before the current
+ // head. That is, the collision cannot be related to commit
+ // which is progress.
+ if (branches.getBranchCommit(r) == null
+ && isRevisionNewer(head, r)) {
+ NodeDocument.removeCollision(op, r);
+ }
+ }
+ }
+ if (op.hasChanges()) {
+ LOG.debug("Removing collisions {}", op.getChanges().keySet());
+ store.findAndUpdate(NODES, op);
+ }
+ }
+
private void backgroundSplit() {
for (Iterator<String> it = splitCandidates.keySet().iterator(); it.hasNext();) {
String id = it.next();
Modified: jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentNodeStoreBranch.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentNodeStoreBranch.java?rev=1653463&r1=1653462&r2=1653463&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentNodeStoreBranch.java (original)
+++ jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentNodeStoreBranch.java Wed Jan 21 09:31:46 2015
@@ -296,7 +296,7 @@ class DocumentNodeStoreBranch implements
DocumentNodeState base,
CommitInfo info) {
boolean success = false;
- Commit c = store.newCommit(base.getRevision());
+ Commit c = store.newCommit(base.getRevision(), this);
Revision rev;
try {
op.with(c);
@@ -655,7 +655,10 @@ class DocumentNodeStoreBranch implements
private void resetBranch(DocumentNodeState branchHead, DocumentNodeState ancestor) {
try {
- head = store.getRoot(store.reset(branchHead.getRevision(), ancestor.getRevision()));
+ head = store.getRoot(
+ store.reset(branchHead.getRevision(),
+ ancestor.getRevision(),
+ DocumentNodeStoreBranch.this));
} catch (Exception e) {
CommitFailedException ex = new CommitFailedException(
OAK, 100, "Branch reset failed", e);
Modified: jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/MergeCommit.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/MergeCommit.java?rev=1653463&r1=1653462&r2=1653463&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/MergeCommit.java (original)
+++ jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/MergeCommit.java Wed Jan 21 09:31:46 2015
@@ -29,7 +29,7 @@ class MergeCommit extends Commit {
MergeCommit(DocumentNodeStore nodeStore,
Revision baseRevision,
SortedSet<Revision> revisions) {
- super(nodeStore, baseRevision, revisions.last());
+ super(nodeStore, revisions.last(), baseRevision, null);
this.mergeRevs = revisions;
}
Modified: jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/RevisionContext.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/RevisionContext.java?rev=1653463&r1=1653462&r2=1653463&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/RevisionContext.java (original)
+++ jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/RevisionContext.java Wed Jan 21 09:31:46 2015
@@ -18,6 +18,8 @@ package org.apache.jackrabbit.oak.plugin
import java.util.Comparator;
+import javax.annotation.Nonnull;
+
/**
* Provides revision related context.
*/
@@ -44,4 +46,9 @@ public interface RevisionContext {
*/
int getClusterId();
+ /**
+ * @return the current head revision.
+ */
+ @Nonnull
+ Revision getHeadRevision();
}
Modified: jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/SplitOperations.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/SplitOperations.java?rev=1653463&r1=1653462&r2=1653463&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/SplitOperations.java (original)
+++ jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/SplitOperations.java Wed Jan 21 09:31:46 2015
@@ -20,6 +20,7 @@ package org.apache.jackrabbit.oak.plugin
import java.util.ArrayList;
import java.util.Collections;
+import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -74,6 +75,7 @@ class SplitOperations {
private Revision low;
private int numValues;
private Map<String, NavigableMap<Revision, String>> committedChanges;
+ private Map<String, Set<Revision>> garbage;
private Set<Revision> mostRecentRevs;
private Set<Revision> splitRevs;
private List<UpdateOp> splitOps;
@@ -116,6 +118,7 @@ class SplitOperations {
splitOps = Lists.newArrayList();
mostRecentRevs = Sets.newHashSet();
splitRevs = Sets.newHashSet();
+ garbage = Maps.newHashMap();
committedChanges = getCommittedLocalChanges();
// revisions of the most recent committed changes on this document
@@ -134,6 +137,9 @@ class SplitOperations {
// remove stale references to previous docs
disconnectStalePrevDocs();
+
+ // remove garbage
+ removeGarbage();
// main document must be updated last
if (main != null) {
@@ -366,11 +372,33 @@ class SplitOperations {
}
if (doc.isCommitted(rev)) {
splitMap.put(rev, entry.getValue());
+ } else if (isGarbage(rev)) {
+ addGarbage(rev, property);
}
}
}
return committedLocally;
}
+
+ private boolean isGarbage(Revision rev) {
+ Revision head = context.getHeadRevision();
+ Comparator<Revision> comp = context.getRevisionComparator();
+ if (comp.compare(head, rev) <= 0) {
+ // this may be an in-progress commit
+ return false;
+ }
+ // garbage if not part of an active branch
+ return context.getBranches().getBranchCommit(rev) == null;
+ }
+
+ private void addGarbage(Revision rev, String property) {
+ Set<Revision> revisions = garbage.get(property);
+ if (revisions == null) {
+ revisions = Sets.newHashSet();
+ garbage.put(property, revisions);
+ }
+ revisions.add(rev);
+ }
private void disconnectStalePrevDocs() {
NavigableMap<Revision, Range> ranges = doc.getPreviousRanges(true);
@@ -406,6 +434,21 @@ class SplitOperations {
}
}
+
+ private void removeGarbage() {
+ if (garbage.isEmpty()) {
+ return;
+ } else if (main == null) {
+ main = new UpdateOp(id, false);
+ }
+ for (Map.Entry<String, Set<Revision>> entry : garbage.entrySet()) {
+ for (Revision r : entry.getValue()) {
+ main.removeMapEntry(entry.getKey(), r);
+ NodeDocument.removeCommitRoot(main, r);
+ NodeDocument.removeRevision(main, r);
+ }
+ }
+ }
private void trackHigh(Revision r) {
if (high == null || isRevisionNewer(context, r, high)) {
Modified: jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/UnmergedBranches.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/UnmergedBranches.java?rev=1653463&r1=1653462&r2=1653463&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/UnmergedBranches.java (original)
+++ jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/UnmergedBranches.java Wed Jan 21 09:31:46 2015
@@ -16,6 +16,7 @@
*/
package org.apache.jackrabbit.oak.plugins.document;
+import java.lang.ref.ReferenceQueue;
import java.util.Comparator;
import java.util.List;
import java.util.SortedSet;
@@ -25,8 +26,10 @@ import java.util.concurrent.atomic.Atomi
import javax.annotation.CheckForNull;
import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
import org.apache.jackrabbit.oak.plugins.document.Branch.BranchCommit;
+import org.apache.jackrabbit.oak.plugins.document.Branch.BranchReference;
import org.apache.jackrabbit.oak.plugins.document.util.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -43,11 +46,19 @@ class UnmergedBranches {
private final Logger log = LoggerFactory.getLogger(getClass());
/**
- * Map of branches with the head of the branch as key.
+ * List of active branches.
*/
private final List<Branch> branches = new CopyOnWriteArrayList<Branch>();
/**
+ * Queue for {@link BranchReference} when the
+ * {@link DocumentNodeStoreBranch} associated with a {@link Branch} is only
+ * weakly reachable.
+ */
+ private final ReferenceQueue<Object> queue =
+ new ReferenceQueue<Object>();
+
+ /**
* Set to <code>true</code> once initialized.
*/
private final AtomicBoolean initialized = new AtomicBoolean(false);
@@ -89,20 +100,28 @@ class UnmergedBranches {
/**
* Create a branch with an initial commit revision.
*
- * @param base the base revision of the branch.
- * @param initial the initial commit to the branch.
+ * @param base the base revision of the branch (must be a non-branch revision).
+ * @param initial the initial commit to the branch (must be a branch revision).
+ * @param guard an optional guard object controlling the life time of this
+ * branch. When {@code guard} becomes weakly reachable, the
+ * branch will at some point later be considered orphaned.
+ * Orphaned branches can be retrieved with
+ * {@link #pollOrphanedBranch()}.
* @return the branch.
- * @throws IllegalArgumentException if
+ * @throws IllegalArgumentException if {@code base} is a branch revision
+ * or {@code initial} is not a branch revision.
*/
@Nonnull
- Branch create(@Nonnull Revision base, @Nonnull Revision initial) {
+ Branch create(@Nonnull Revision base,
+ @Nonnull Revision initial,
+ @Nullable Object guard) {
checkArgument(!checkNotNull(base).isBranch(),
"base is not a trunk revision: %s", base);
checkArgument(checkNotNull(initial).isBranch(),
"initial is not a branch revision: %s", initial);
SortedSet<Revision> commits = new TreeSet<Revision>(comparator);
commits.add(initial);
- Branch b = new Branch(commits, base);
+ Branch b = new Branch(commits, base, queue, guard);
branches.add(b);
return b;
}
@@ -156,4 +175,20 @@ class UnmergedBranches {
int size(){
return branches.size();
}
+
+ /**
+ * Polls for an orphaned branch. The implementation will remove the branch
+ * from the internal list of unmerged branches.
+ *
+ * @return an orphaned branch or {@code null} if there is none at this time.
+ */
+ Branch pollOrphanedBranch() {
+ BranchReference ref = (BranchReference) queue.poll();
+ if (ref != null) {
+ if (branches.remove(ref.getBranch())) {
+ return ref.getBranch();
+ }
+ }
+ return null;
+ }
}
Modified: jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/UnsavedModifications.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/UnsavedModifications.java?rev=1653463&r1=1653462&r2=1653463&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/UnsavedModifications.java (original)
+++ jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/UnsavedModifications.java Wed Jan 21 09:31:46 2015
@@ -162,7 +162,7 @@ class UnsavedModifications {
int size = pathList.size();
if (updateOp == null) {
// create UpdateOp
- Commit commit = new Commit(store, null, r);
+ Commit commit = new Commit(store, r, null, null);
updateOp = commit.getUpdateOperationForNode(p);
NodeDocument.setLastRev(updateOp, r);
lastRev = r;
Modified: jackrabbit/oak/trunk/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/CommitQueueTest.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/CommitQueueTest.java?rev=1653463&r1=1653462&r2=1653463&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/CommitQueueTest.java (original)
+++ jackrabbit/oak/trunk/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/CommitQueueTest.java Wed Jan 21 09:31:46 2015
@@ -72,7 +72,7 @@ public class CommitQueueTest {
public void run() {
try {
for (int i = 0; i < COMMITS_PER_WRITER; i++) {
- Commit commit = store.newCommit(null);
+ Commit commit = store.newCommit(null, null);
try {
Thread.sleep(0, random.nextInt(1000));
} catch (InterruptedException e) {
Modified: jackrabbit/oak/trunk/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/DocumentNodeStoreTest.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/DocumentNodeStoreTest.java?rev=1653463&r1=1653462&r2=1653463&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/DocumentNodeStoreTest.java (original)
+++ jackrabbit/oak/trunk/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/DocumentNodeStoreTest.java Wed Jan 21 09:31:46 2015
@@ -229,7 +229,7 @@ public class DocumentNodeStoreTest {
public void run() {
try {
Revision r = store.newRevision();
- Commit c = new Commit(store, Revision.fromString(head), r);
+ Commit c = new Commit(store, r, Revision.fromString(head), null);
c.addNode(new DocumentNodeState(store, "/foo/node", r));
c.addNode(new DocumentNodeState(store, "/bar/node", r));
c.apply();
@@ -246,7 +246,7 @@ public class DocumentNodeStoreTest {
created.acquireUninterruptibly();
// commit will succeed and add collision marker to writer commit
Revision r = store.newRevision();
- Commit c = new Commit(store, Revision.fromString(head), r);
+ Commit c = new Commit(store, r, Revision.fromString(head), null);
c.addNode(new DocumentNodeState(store, "/foo/node", r));
c.addNode(new DocumentNodeState(store, "/bar/node", r));
c.apply();
Modified: jackrabbit/oak/trunk/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/DummyRevisionContext.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/DummyRevisionContext.java?rev=1653463&r1=1653462&r2=1653463&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/DummyRevisionContext.java (original)
+++ jackrabbit/oak/trunk/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/DummyRevisionContext.java Wed Jan 21 09:31:46 2015
@@ -18,6 +18,8 @@ package org.apache.jackrabbit.oak.plugin
import java.util.Comparator;
+import javax.annotation.Nonnull;
+
/**
* A revision context for tests.
*/
@@ -47,4 +49,10 @@ public class DummyRevisionContext implem
public int getClusterId() {
return 1;
}
+
+ @Nonnull
+ @Override
+ public Revision getHeadRevision() {
+ return Revision.newRevision(1);
+ }
}
Added: jackrabbit/oak/trunk/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/OrphanedBranchTest.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/OrphanedBranchTest.java?rev=1653463&view=auto
==============================================================================
--- jackrabbit/oak/trunk/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/OrphanedBranchTest.java (added)
+++ jackrabbit/oak/trunk/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/OrphanedBranchTest.java Wed Jan 21 09:31:46 2015
@@ -0,0 +1,149 @@
+/*
+ * 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.jackrabbit.oak.plugins.document;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.collect.Lists;
+
+import org.apache.jackrabbit.oak.plugins.document.util.Utils;
+import org.apache.jackrabbit.oak.spi.commit.CommitInfo;
+import org.apache.jackrabbit.oak.spi.commit.EmptyHook;
+import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.jackrabbit.oak.plugins.document.Collection.NODES;
+import static org.apache.jackrabbit.oak.plugins.document.NodeDocument.COLLISIONS;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Test for OAK-2421.
+ */
+@RunWith(Parameterized.class)
+public class OrphanedBranchTest {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OrphanedBranchTest.class);
+ private DocumentStoreFixture fixture;
+ private DocumentNodeStore store;
+ private VersionGarbageCollector gc;
+
+ public OrphanedBranchTest(DocumentStoreFixture fixture) {
+ this.fixture = fixture;
+ }
+
+ @Parameterized.Parameters
+ public static java.util.Collection<Object[]> fixtures() throws IOException {
+ List<Object[]> fixtures = Lists.newArrayList();
+ fixtures.add(new Object[] {new DocumentStoreFixture.MemoryFixture()});
+
+ DocumentStoreFixture mongo = new DocumentStoreFixture.MongoFixture();
+ if(mongo.isAvailable()){
+ fixtures.add(new Object[] {mongo});
+ }
+ return fixtures;
+ }
+
+ @Before
+ public void setUp() throws InterruptedException {
+ store = new DocumentMK.Builder()
+ .setDocumentStore(fixture.createDocumentStore())
+ .setAsyncDelay(0)
+ .getNodeStore();
+ gc = store.getVersionGarbageCollector();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ store.dispose();
+ }
+
+ @Test
+ public void orphanedBranches() throws Exception {
+ int numCreated = 0;
+ for (;;) {
+ NodeBuilder builder = store.getRoot().builder();
+ NodeBuilder child = builder.child("foo");
+ int numBranches = store.getBranches().size();
+
+ int count = 0;
+ // perform changes until a branch is created
+ while (store.getBranches().size() == numBranches) {
+ child.setProperty("prop", count++);
+ }
+ // but do not merge!
+ numCreated++;
+
+ // commit a change to get new head revision.
+ // collisions will only be cleaned up if the head
+ // revision is newer.
+ builder = store.getRoot().builder();
+ builder.child("bar").setProperty("prop", numCreated);
+ store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
+
+ numBranches = store.getBranches().size();
+ store.runBackgroundOperations();
+ // after background ops we must not see more collisions
+ // than active branches
+ String id = Utils.getIdFromPath("/");
+ NodeDocument doc = store.getDocumentStore().find(NODES, id);
+ assertNotNull(doc);
+ Map<Revision, String> collisions = doc.getLocalMap(COLLISIONS);
+ assertTrue("too many collisions: " + collisions.size(),
+ collisions.size() <= numBranches);
+ // split ops must remove orphaned changes
+ // limit to check is number of branches considered active
+ // plus NodeDocument.NUM_REVS_THRESHOLD
+ int limit = numBranches + NodeDocument.NUM_REVS_THRESHOLD;
+ id = Utils.getIdFromPath("/foo");
+ doc = store.getDocumentStore().find(NODES, id);
+ assertNotNull(doc);
+ Map<Revision, String> map = doc.getLocalMap("prop");
+ assertTrue("too many orphaned changes: " + map.size() + " > " + limit,
+ map.size() <= limit);
+ map = doc.getLocalCommitRoot();
+ assertTrue("too many orphaned commit root entries: " + map.size() + " > " + limit,
+ map.size() <= limit);
+
+ // run garbage collector once in a while for changes on /bar
+ if (numCreated % NodeDocument.NUM_REVS_THRESHOLD == 0) {
+ gc.gc(1, TimeUnit.SECONDS);
+ }
+
+ LOG.info("created {}, still considered active: {}",
+ numCreated, store.getBranches().size());
+
+ // create orphaned branches, until we were able to
+ // collect 500 of them (or 100 for MongoDB and other fixtures)
+ int collect = fixture instanceof DocumentStoreFixture.MemoryFixture ? 500 : 100;
+ if (numCreated - store.getBranches().size() >= collect) {
+ break;
+ }
+ }
+ }
+}
Propchange: jackrabbit/oak/trunk/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/document/OrphanedBranchTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Re: svn commit: r1653463 - in /jackrabbit/oak/trunk/oak-core/src:
main/java/org/apache/jackrabbit/oak/plugins/document/ test/java/org/apache/jackrabbit/oak/plugins/document/
Posted by Chetan Mehrotra <ch...@gmail.com>.
On Wed, Jan 21, 2015 at 3:01 PM, <mr...@apache.org> wrote:
> OAK-2421: Clean up orphaned branches
Thats neat solution for this problem !!
Chetan Mehrotra