oak-store-document/src/main/java/org/apache/jackrabbit/oak/plugins/document/MissingBcSweeper2.java - jackrabbit-oak - Git at Google

 /*
  * 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 static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.collect.Iterables.filter;
 import static com.google.common.collect.Iterables.partition;
 import static com.google.common.collect.Iterables.transform;
 import static com.google.common.collect.Maps.immutableEntry;
 import static com.google.common.collect.Maps.newHashMap;
 import static org.apache.jackrabbit.oak.plugins.document.util.Utils.COMMITROOT_OR_REVISIONS;

 import java.util.Collections;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;

 import org.apache.jackrabbit.oak.commons.TimeDurationFormatter;
 import org.apache.jackrabbit.oak.plugins.document.util.Utils;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import com.google.common.base.Function;
 import com.google.common.base.Predicate;

 /**
  * The {@code MissingBcSweeper2} is used for the so-called sweep2, which is
  * a repository traversal updating documents that have missing branch commit ("_bc")
  * properties (details see OAK-9176).
  * This class is similar to NodeDocumentSweeper as it is based on the same principles,
  * with a few notable exceptions (like it only looks at _commitRoot and _revisions etc).
  * And due to these exceptions the class is forked rather than modified/subclasses
  * (also to enable later refactoring of the NodeDocumentSweeper itself).
  * <p>
  * This class is not thread-safe.
  */
 final class MissingBcSweeper2 {

     private static final Logger LOG = LoggerFactory.getLogger(MissingBcSweeper2.class);

     private static final int YIELD_SIZE = 500;

     private static final int INVALIDATE_BATCH_SIZE = 100;

     private static final long LOGINTERVALMS = TimeUnit.MINUTES.toMillis(1);

     private final RevisionContext context;

     private final CommitValueResolver commitValueResolver;

     private final int executingClusterId;

     private final List<Integer> includedClusterIds;

     private final RevisionVector headRevision;

     private final AtomicBoolean isDisposed;

     private long totalCount;
     private long lastCount;
     private long startOfScan;
     private long lastLog;

     /**
      * Creates a new sweeper v2 for the given context..
      *
      * @param context the revision context.
      */
     MissingBcSweeper2(RevisionContext context,
                     CommitValueResolver commitValueResolver,
                     List<Integer> includedClusterIds,
                     AtomicBoolean isDisposed) {
         this.context = checkNotNull(context);
         this.commitValueResolver = checkNotNull(commitValueResolver);
         this.executingClusterId = context.getClusterId();
         this.includedClusterIds = includedClusterIds == null ? new LinkedList<>() : Collections.unmodifiableList(includedClusterIds);
         this.headRevision= context.getHeadRevision();
         this.isDisposed = isDisposed;
     }

     /**
      * Performs a sweep2 and reports the required updates to the given sweep
      * listener.
      *
      * @param documents the documents to sweep
      * @param listener the listener to receive required sweep update operations.
      * @throws DocumentStoreException if reading from the store or writing to
      *          the store failed.
      */
     void sweep2(@NotNull Iterable<NodeDocument> documents,
                    @NotNull NodeDocumentSweepListener listener)
             throws DocumentStoreException {
         performSweep2(documents, checkNotNull(listener));
     }

     //----------------------------< internal >----------------------------------

     private void performSweep2(Iterable<NodeDocument> documents,
                                   NodeDocumentSweepListener listener)
             throws DocumentStoreException {
         totalCount = 0;
         lastCount = 0;
         startOfScan = context.getClock().getTime();
         lastLog = startOfScan;

         Iterable<Map.Entry<Path, UpdateOp>> ops = sweepOperations(documents);
         for (List<Map.Entry<Path, UpdateOp>> batch : partition(ops, INVALIDATE_BATCH_SIZE)) {
             Map<Path, UpdateOp> updates = newHashMap();
             for (Map.Entry<Path, UpdateOp> entry : batch) {
                 updates.put(entry.getKey(), entry.getValue());
             }
             listener.sweepUpdate(updates);
             if (isDisposed.get()) {
                 throw new DocumentStoreException("sweep2 interrupted by shutdown");
             }
         }
         LOG.debug("Document sweep2 finished");
     }

     private Iterable<Map.Entry<Path, UpdateOp>> sweepOperations(
             final Iterable<NodeDocument> docs) {
         return filter(transform(docs,
                 new Function<NodeDocument, Map.Entry<Path, UpdateOp>>() {

             int yieldCnt = 0;
             long lastYield = context.getClock().getTime();

             @Override
             public Map.Entry<Path, UpdateOp> apply(NodeDocument doc) {
                 if (++yieldCnt >= YIELD_SIZE) {
                     try {
                         final long now = context.getClock().getTime();
                         final long timeSinceLastYield = now - lastYield;
                         // wait the same amount of time that passed since last yield
                         // that corresponds to roughly 50% throttle (ignoring the min 1ms sleep)
                         final long waitUntil = now + Math.max(1, timeSinceLastYield);
                         context.getClock().waitUntil(waitUntil);
                     } catch (InterruptedException e) {
                         // ignore
                     }
                     lastYield = context.getClock().getTime();
                     yieldCnt = 0;
                 }
                 return immutableEntry(doc.getPath(), sweepOne(doc));
             }
         }), new Predicate<Map.Entry<Path, UpdateOp>>() {
             @Override
             public boolean apply(Map.Entry<Path, UpdateOp> input) {
                 return input.getValue() != null;
             }
         });
     }

     private UpdateOp sweepOne(NodeDocument doc) throws DocumentStoreException {
         UpdateOp op = null;
         // the blueprint NodeDocumentSweeper.sweepOne goes through
         // PROPERTY_OR_DELETED_OR_COMMITROOT here, as that's the new full sweep1.
         // the sweep2 though, only goes through COMMITROOT_OR_REVISIONS,
         // as that's what was left out by the original sweep1:
         // - COMMITROOT : for new child (parent)
         // - REVISIONS : for commit roots (root for branch commits)
         for (String property : filter(doc.keySet(), COMMITROOT_OR_REVISIONS)) {
             Map<Revision, String> valueMap = doc.getLocalMap(property);
             for (Map.Entry<Revision, String> entry : valueMap.entrySet()) {
                 Revision rev = entry.getKey();

                 if (!includedClusterIds.isEmpty() && !includedClusterIds.contains(rev.getClusterId())) {
                     // sweep2 is only done for those clusterIds that went through sweep1 (before 1.8).
                     // includedClusterIds contains the latter list of clusterIds.
                     // For testing, the code also supports going through all clusterIds.
                     // (in this case the includedClusterIds list is empty).
                     // Note that sweep2 is never triggered if "_sweepRev" is empty,
                     // so the empty case here never applies for production.
                     continue;
                 }

                 Revision cRev = getCommitRevision(doc, rev);
                 if (cRev == null) {
                     // originally the uncommitted case
                     // however, sweep2 doesn't touch anything uncommitted
                 } else if (cRev.equals(rev)) {
                     // originally the committed case
                     // however, that was only logging and sweep2 is not about that,
                     // hence nothing to be done here
                 } else {
                     // this is what sweep2 is about : adding _bc for _commitRoot and _revisions
                     if (op == null) {
                         op = createUpdateOp(doc);
                     }
                     committedBranch(doc, property, rev, cRev, op);
                 }
             }
         }

         totalCount++;
         lastCount++;
         long now = context.getClock().getTime();
         long lastElapsed = now - lastLog;

         if (lastElapsed >= LOGINTERVALMS) {
             TimeDurationFormatter df = TimeDurationFormatter.forLogging();

             long totalElapsed = now - startOfScan;
             long totalRateMin = (totalCount * TimeUnit.MINUTES.toMillis(1)) / totalElapsed;
             long lastRateMin = (lastCount * TimeUnit.MINUTES.toMillis(1)) / lastElapsed;

             String restrictionMsg;
             if (includedClusterIds.isEmpty()) {
                 restrictionMsg = "unrestricted, ie for all clusterIds";
             } else {
                 restrictionMsg = "restricted to clusterIds " + includedClusterIds;
             }
             String message = String.format(
                     "Sweep2 executed by cluster node [%d] (%s): %d nodes scanned in %s (~%d/m) - last interval %d nodes in %s (~%d/m)",
                     executingClusterId, restrictionMsg, totalCount, df.format(totalElapsed, TimeUnit.MILLISECONDS), totalRateMin, lastCount,
                     df.format(lastElapsed, TimeUnit.MILLISECONDS), lastRateMin);

             LOG.info(message);
             lastLog = now;
             lastCount = 0;
         }

         return op == null ? null : op.hasChanges() ? op : null;
     }

     /**
      * Returns {@code true} if the given revision is marked as a branch commit
      * on the document. This method only checks local branch commit information
      * available on the document ({@link NodeDocument#getLocalBranchCommits()}).
      * If the given revision is related to a branch commit that was created
      * prior to Oak 1.8, the method will return {@code false}.
      *
      * @param rev a revision.
      * @param doc the document to check.
      * @return {@code true} if the revision is marked as a branch commit;
      *          {@code false} otherwise.
      */
     private boolean isV18BranchCommit(Revision rev, NodeDocument doc) {
         return doc.getLocalBranchCommits().contains(rev);
     }

     private void committedBranch(NodeDocument doc,
                                  String property,
                                  Revision rev,
                                  Revision cRev,
                                  UpdateOp op) {
         boolean newerThanHead = headRevision.isRevisionNewer(cRev);
         if (LOG.isDebugEnabled()) {
             String msg = newerThanHead ? " (newer than head)" : "";
             LOG.debug("Committed branch change on {}, {} @ {}/{}{}",
                     op.getId(), property, rev, cRev, msg);
         }
         if (!isV18BranchCommit(rev, doc)) {
             NodeDocument.setBranchCommit(op, rev);
         }
     }

     private static UpdateOp createUpdateOp(NodeDocument doc) {
         return new UpdateOp(doc.getId(), false);
     }

     private String getCommitValue(@NotNull Revision changeRevision,
             @NotNull NodeDocument doc) {
         return commitValueResolver.resolve(changeRevision, doc);
     }

     @Nullable
     private Revision getCommitRevision(final NodeDocument doc,
                                        final Revision rev)
             throws DocumentStoreException {
         String cv = /*context.*/getCommitValue(rev, doc);
         if (cv == null) {
             return null;
         }
         return Utils.resolveCommitRevision(rev, cv);
     }
 }
	/*
	* 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 static com.google.common.base.Preconditions.checkNotNull;
	import static com.google.common.collect.Iterables.filter;
	import static com.google.common.collect.Iterables.partition;
	import static com.google.common.collect.Iterables.transform;
	import static com.google.common.collect.Maps.immutableEntry;
	import static com.google.common.collect.Maps.newHashMap;
	import static org.apache.jackrabbit.oak.plugins.document.util.Utils.COMMITROOT_OR_REVISIONS;

	import java.util.Collections;
	import java.util.LinkedList;
	import java.util.List;
	import java.util.Map;
	import java.util.concurrent.TimeUnit;
	import java.util.concurrent.atomic.AtomicBoolean;

	import org.apache.jackrabbit.oak.commons.TimeDurationFormatter;
	import org.apache.jackrabbit.oak.plugins.document.util.Utils;
	import org.jetbrains.annotations.NotNull;
	import org.jetbrains.annotations.Nullable;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import com.google.common.base.Function;
	import com.google.common.base.Predicate;

	/**
	* The {@code MissingBcSweeper2} is used for the so-called sweep2, which is
	* a repository traversal updating documents that have missing branch commit ("_bc")
	* properties (details see OAK-9176).
	* This class is similar to NodeDocumentSweeper as it is based on the same principles,
	* with a few notable exceptions (like it only looks at _commitRoot and _revisions etc).
	* And due to these exceptions the class is forked rather than modified/subclasses
	* (also to enable later refactoring of the NodeDocumentSweeper itself).
	* <p>
	* This class is not thread-safe.
	*/
	final class MissingBcSweeper2 {

	private static final Logger LOG = LoggerFactory.getLogger(MissingBcSweeper2.class);

	private static final int YIELD_SIZE = 500;

	private static final int INVALIDATE_BATCH_SIZE = 100;

	private static final long LOGINTERVALMS = TimeUnit.MINUTES.toMillis(1);

	private final RevisionContext context;

	private final CommitValueResolver commitValueResolver;

	private final int executingClusterId;

	private final List<Integer> includedClusterIds;

	private final RevisionVector headRevision;

	private final AtomicBoolean isDisposed;

	private long totalCount;
	private long lastCount;
	private long startOfScan;
	private long lastLog;

	/**
	* Creates a new sweeper v2 for the given context..
	*
	* @param context the revision context.
	*/
	MissingBcSweeper2(RevisionContext context,
	CommitValueResolver commitValueResolver,
	List<Integer> includedClusterIds,
	AtomicBoolean isDisposed) {
	this.context = checkNotNull(context);
	this.commitValueResolver = checkNotNull(commitValueResolver);
	this.executingClusterId = context.getClusterId();
	this.includedClusterIds = includedClusterIds == null ? new LinkedList<>() : Collections.unmodifiableList(includedClusterIds);
	this.headRevision= context.getHeadRevision();
	this.isDisposed = isDisposed;
	}

	/**
	* Performs a sweep2 and reports the required updates to the given sweep
	* listener.
	*
	* @param documents the documents to sweep
	* @param listener the listener to receive required sweep update operations.
	* @throws DocumentStoreException if reading from the store or writing to
	* the store failed.
	*/
	void sweep2(@NotNull Iterable<NodeDocument> documents,
	@NotNull NodeDocumentSweepListener listener)
	throws DocumentStoreException {
	performSweep2(documents, checkNotNull(listener));
	}

	//----------------------------< internal >----------------------------------

	private void performSweep2(Iterable<NodeDocument> documents,
	NodeDocumentSweepListener listener)
	throws DocumentStoreException {
	totalCount = 0;
	lastCount = 0;
	startOfScan = context.getClock().getTime();
	lastLog = startOfScan;

	Iterable<Map.Entry<Path, UpdateOp>> ops = sweepOperations(documents);
	for (List<Map.Entry<Path, UpdateOp>> batch : partition(ops, INVALIDATE_BATCH_SIZE)) {
	Map<Path, UpdateOp> updates = newHashMap();
	for (Map.Entry<Path, UpdateOp> entry : batch) {
	updates.put(entry.getKey(), entry.getValue());
	}
	listener.sweepUpdate(updates);
	if (isDisposed.get()) {
	throw new DocumentStoreException("sweep2 interrupted by shutdown");
	}
	}
	LOG.debug("Document sweep2 finished");
	}

	private Iterable<Map.Entry<Path, UpdateOp>> sweepOperations(
	final Iterable<NodeDocument> docs) {
	return filter(transform(docs,
	new Function<NodeDocument, Map.Entry<Path, UpdateOp>>() {

	int yieldCnt = 0;
	long lastYield = context.getClock().getTime();

	@Override
	public Map.Entry<Path, UpdateOp> apply(NodeDocument doc) {
	if (++yieldCnt >= YIELD_SIZE) {
	try {
	final long now = context.getClock().getTime();
	final long timeSinceLastYield = now - lastYield;
	// wait the same amount of time that passed since last yield
	// that corresponds to roughly 50% throttle (ignoring the min 1ms sleep)
	final long waitUntil = now + Math.max(1, timeSinceLastYield);
	context.getClock().waitUntil(waitUntil);
	} catch (InterruptedException e) {
	// ignore
	}
	lastYield = context.getClock().getTime();
	yieldCnt = 0;
	}
	return immutableEntry(doc.getPath(), sweepOne(doc));
	}
	}), new Predicate<Map.Entry<Path, UpdateOp>>() {
	@Override
	public boolean apply(Map.Entry<Path, UpdateOp> input) {
	return input.getValue() != null;
	}
	});
	}

	private UpdateOp sweepOne(NodeDocument doc) throws DocumentStoreException {
	UpdateOp op = null;
	// the blueprint NodeDocumentSweeper.sweepOne goes through
	// PROPERTY_OR_DELETED_OR_COMMITROOT here, as that's the new full sweep1.
	// the sweep2 though, only goes through COMMITROOT_OR_REVISIONS,
	// as that's what was left out by the original sweep1:
	// - COMMITROOT : for new child (parent)
	// - REVISIONS : for commit roots (root for branch commits)
	for (String property : filter(doc.keySet(), COMMITROOT_OR_REVISIONS)) {
	Map<Revision, String> valueMap = doc.getLocalMap(property);
	for (Map.Entry<Revision, String> entry : valueMap.entrySet()) {
	Revision rev = entry.getKey();

	if (!includedClusterIds.isEmpty() && !includedClusterIds.contains(rev.getClusterId())) {
	// sweep2 is only done for those clusterIds that went through sweep1 (before 1.8).
	// includedClusterIds contains the latter list of clusterIds.
	// For testing, the code also supports going through all clusterIds.
	// (in this case the includedClusterIds list is empty).
	// Note that sweep2 is never triggered if "_sweepRev" is empty,
	// so the empty case here never applies for production.
	continue;
	}

	Revision cRev = getCommitRevision(doc, rev);
	if (cRev == null) {
	// originally the uncommitted case
	// however, sweep2 doesn't touch anything uncommitted
	} else if (cRev.equals(rev)) {
	// originally the committed case
	// however, that was only logging and sweep2 is not about that,
	// hence nothing to be done here
	} else {
	// this is what sweep2 is about : adding _bc for _commitRoot and _revisions
	if (op == null) {
	op = createUpdateOp(doc);
	}
	committedBranch(doc, property, rev, cRev, op);
	}
	}
	}

	totalCount++;
	lastCount++;
	long now = context.getClock().getTime();
	long lastElapsed = now - lastLog;

	if (lastElapsed >= LOGINTERVALMS) {
	TimeDurationFormatter df = TimeDurationFormatter.forLogging();

	long totalElapsed = now - startOfScan;
	long totalRateMin = (totalCount * TimeUnit.MINUTES.toMillis(1)) / totalElapsed;
	long lastRateMin = (lastCount * TimeUnit.MINUTES.toMillis(1)) / lastElapsed;

	String restrictionMsg;
	if (includedClusterIds.isEmpty()) {
	restrictionMsg = "unrestricted, ie for all clusterIds";
	} else {
	restrictionMsg = "restricted to clusterIds " + includedClusterIds;
	}
	String message = String.format(
	"Sweep2 executed by cluster node [%d] (%s): %d nodes scanned in %s (~%d/m) - last interval %d nodes in %s (~%d/m)",
	executingClusterId, restrictionMsg, totalCount, df.format(totalElapsed, TimeUnit.MILLISECONDS), totalRateMin, lastCount,
	df.format(lastElapsed, TimeUnit.MILLISECONDS), lastRateMin);

	LOG.info(message);
	lastLog = now;
	lastCount = 0;
	}

	return op == null ? null : op.hasChanges() ? op : null;
	}

	/**
	* Returns {@code true} if the given revision is marked as a branch commit
	* on the document. This method only checks local branch commit information
	* available on the document ({@link NodeDocument#getLocalBranchCommits()}).
	* If the given revision is related to a branch commit that was created
	* prior to Oak 1.8, the method will return {@code false}.
	*
	* @param rev a revision.
	* @param doc the document to check.
	* @return {@code true} if the revision is marked as a branch commit;
	* {@code false} otherwise.
	*/
	private boolean isV18BranchCommit(Revision rev, NodeDocument doc) {
	return doc.getLocalBranchCommits().contains(rev);
	}

	private void committedBranch(NodeDocument doc,
	String property,
	Revision rev,
	Revision cRev,
	UpdateOp op) {
	boolean newerThanHead = headRevision.isRevisionNewer(cRev);
	if (LOG.isDebugEnabled()) {
	String msg = newerThanHead ? " (newer than head)" : "";
	LOG.debug("Committed branch change on {}, {} @ {}/{}{}",
	op.getId(), property, rev, cRev, msg);
	}
	if (!isV18BranchCommit(rev, doc)) {
	NodeDocument.setBranchCommit(op, rev);
	}
	}

	private static UpdateOp createUpdateOp(NodeDocument doc) {
	return new UpdateOp(doc.getId(), false);
	}

	private String getCommitValue(@NotNull Revision changeRevision,
	@NotNull NodeDocument doc) {
	return commitValueResolver.resolve(changeRevision, doc);
	}

	@Nullable
	private Revision getCommitRevision(final NodeDocument doc,
	final Revision rev)
	throws DocumentStoreException {
	String cv = /context./getCommitValue(rev, doc);
	if (cv == null) {
	return null;
	}
	return Utils.resolveCommitRevision(rev, cv);
	}
	}