You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by be...@apache.org on 2015/07/02 20:21:11 UTC
[11/12] cassandra git commit: Merge branch 'cassandra-2.1' into
cassandra-2.2
Merge branch 'cassandra-2.1' into cassandra-2.2
Conflicts:
CHANGES.txt
src/java/org/apache/cassandra/db/DataTracker.java
src/java/org/apache/cassandra/io/sstable/SSTableDeletingTask.java
src/java/org/apache/cassandra/io/sstable/SSTableRewriter.java
src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/2357589d
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/2357589d
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/2357589d
Branch: refs/heads/cassandra-2.2
Commit: 2357589dd393be1614ed5156ce05674afc4d0921
Parents: 4aa56ec 2595e74
Author: Benedict Elliott Smith <be...@apache.org>
Authored: Thu Jul 2 19:14:39 2015 +0100
Committer: Benedict Elliott Smith <be...@apache.org>
Committed: Thu Jul 2 19:14:39 2015 +0100
----------------------------------------------------------------------
CHANGES.txt | 6 +
.../apache/cassandra/db/lifecycle/Helpers.java | 23 +---
.../db/lifecycle/LifecycleTransaction.java | 7 +-
.../apache/cassandra/db/lifecycle/Tracker.java | 11 +-
.../io/sstable/SSTableDeletingTask.java | 36 ++----
.../cassandra/io/sstable/SSTableRewriter.java | 33 +++--
.../io/sstable/format/SSTableReader.java | 128 ++++++++++++-------
.../cassandra/tools/StandaloneScrubber.java | 2 +-
.../cassandra/tools/StandaloneSplitter.java | 2 +-
.../cassandra/tools/StandaloneUpgrader.java | 2 +-
.../cassandra/db/lifecycle/HelpersTest.java | 4 +-
.../cassandra/db/lifecycle/TrackerTest.java | 9 +-
.../io/sstable/SSTableRewriterTest.java | 2 +-
13 files changed, 135 insertions(+), 130 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/2357589d/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index a734a4b,c37ef02..bc4b57c
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,23 -1,6 +1,29 @@@
++<<<<<<< HEAD
+2.2.0-rc2
+ * (cqlsh) Allow setting the initial connection timeout (CASSANDRA-9601)
+ * BulkLoader has --transport-factory option but does not use it (CASSANDRA-9675)
+ * Allow JMX over SSL directly from nodetool (CASSANDRA-9090)
+ * Update cqlsh for UDFs (CASSANDRA-7556)
+ * Change Windows kernel default timer resolution (CASSANDRA-9634)
+ * Deprected sstable2json and json2sstable (CASSANDRA-9618)
+ * Allow native functions in user-defined aggregates (CASSANDRA-9542)
+ * Don't repair system_distributed by default (CASSANDRA-9621)
+ * Fix mixing min, max, and count aggregates for blob type (CASSANRA-9622)
+ * Rename class for DATE type in Java driver (CASSANDRA-9563)
+ * Duplicate compilation of UDFs on coordinator (CASSANDRA-9475)
+ * Fix connection leak in CqlRecordWriter (CASSANDRA-9576)
+ * Mlockall before opening system sstables & remove boot_without_jna option (CASSANDRA-9573)
+ * Add functions to convert timeuuid to date or time, deprecate dateOf and unixTimestampOf (CASSANDRA-9229)
+ * Make sure we cancel non-compacting sstables from LifecycleTransaction (CASSANDRA-9566)
+ * Fix deprecated repair JMX API (CASSANDRA-9570)
+ * Add logback metrics (CASSANDRA-9378)
+ * Update and refactor ant test/test-compression to run the tests in parallel (CASSANDRA-9583)
+Merged from 2.1:
++=======
+ 2.1.8
+ * Eliminate strong self-reference chains in sstable ref tidiers (CASSANDRA-9656)
+ * Ensure StreamSession uses canonical sstable reader instances (CASSANDRA-9700)
++>>>>>>> cassandra-2.1
* Ensure memtable book keeping is not corrupted in the event we shrink usage (CASSANDRA-9681)
* Update internal python driver for cqlsh (CASSANDRA-9064)
* Fix IndexOutOfBoundsException when inserting tuple with too many
http://git-wip-us.apache.org/repos/asf/cassandra/blob/2357589d/src/java/org/apache/cassandra/db/lifecycle/Helpers.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/lifecycle/Helpers.java
index 05f7531,0000000..e14c339
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/db/lifecycle/Helpers.java
+++ b/src/java/org/apache/cassandra/db/lifecycle/Helpers.java
@@@ -1,241 -1,0 +1,222 @@@
+/*
+ * 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.cassandra.db.lifecycle;
+
+import java.util.*;
+
+import com.google.common.base.Predicate;
+import com.google.common.collect.*;
+
+import org.apache.cassandra.io.sstable.format.SSTableReader;
+
+import static com.google.common.base.Predicates.*;
+import static com.google.common.collect.Iterables.any;
+import static com.google.common.collect.Iterables.concat;
+import static com.google.common.collect.Iterables.filter;
+import static com.google.common.collect.Iterables.getFirst;
+import static org.apache.cassandra.utils.Throwables.merge;
+
+class Helpers
+{
+ /**
+ * update the contents of a set with the provided sets, ensuring that the items to remove are
+ * really present, and that the items to add are not (unless we're also removing them)
+ * @return a new set with the contents of the provided one modified
+ */
+ static <T> Set<T> replace(Set<T> original, Set<T> remove, Iterable<T> add)
+ {
+ return ImmutableSet.copyOf(replace(identityMap(original), remove, add).keySet());
+ }
+
+ /**
+ * update the contents of an "identity map" with the provided sets, ensuring that the items to remove are
+ * really present, and that the items to add are not (unless we're also removing them)
+ * @return a new identity map with the contents of the provided one modified
+ */
+ static <T> Map<T, T> replace(Map<T, T> original, Set<T> remove, Iterable<T> add)
+ {
+ // ensure the ones being removed are the exact same ones present
+ for (T reader : remove)
+ assert original.get(reader) == reader;
+
+ // ensure we don't already contain any we're adding, that we aren't also removing
+ assert !any(add, and(not(in(remove)), in(original.keySet()))) : String.format("original:%s remove:%s add:%s", original.keySet(), remove, add);
+
+ Map<T, T> result =
+ identityMap(concat(add, filter(original.keySet(), not(in(remove)))));
+
+ assert result.size() == original.size() - remove.size() + Iterables.size(add) :
+ String.format("Expecting new size of %d, got %d while replacing %s by %s in %s",
+ original.size() - remove.size() + Iterables.size(add), result.size(), remove, add, original.keySet());
+ return result;
+ }
+
+ /**
+ * A convenience method for encapsulating this action over multiple SSTableReader with exception-safety
+ * @return accumulate if not null (with any thrown exception attached), or any thrown exception otherwise
+ */
- static Throwable setupDeleteNotification(Iterable<SSTableReader> readers, Tracker tracker, Throwable accumulate)
- {
- try
- {
- for (SSTableReader reader : readers)
- reader.setupDeleteNotification(tracker);
- }
- catch (Throwable t)
- {
- // shouldn't be possible, but in case the contract changes in future and we miss it...
- accumulate = merge(accumulate, t);
- }
- return accumulate;
- }
-
- /**
- * A convenience method for encapsulating this action over multiple SSTableReader with exception-safety
- * @return accumulate if not null (with any thrown exception attached), or any thrown exception otherwise
- */
+ static Throwable setReplaced(Iterable<SSTableReader> readers, Throwable accumulate)
+ {
+ for (SSTableReader reader : readers)
+ {
+ try
+ {
+ reader.setReplaced();
+ }
+ catch (Throwable t)
+ {
+ accumulate = merge(accumulate, t);
+ }
+ }
+ return accumulate;
+ }
+
+ /**
+ * assert that none of these readers have been replaced
+ */
+ static void checkNotReplaced(Iterable<SSTableReader> readers)
+ {
+ for (SSTableReader reader : readers)
+ assert !reader.isReplaced();
+ }
+
+ /**
+ * A convenience method for encapsulating this action over multiple SSTableReader with exception-safety
+ * @return accumulate if not null (with any thrown exception attached), or any thrown exception otherwise
+ */
- static Throwable markObsolete(Iterable<SSTableReader> readers, Throwable accumulate)
++ static Throwable markObsolete(Tracker tracker, Iterable<SSTableReader> readers, Throwable accumulate)
+ {
+ for (SSTableReader reader : readers)
+ {
+ try
+ {
- boolean firstToCompact = reader.markObsolete();
++ boolean firstToCompact = reader.markObsolete(tracker);
+ assert firstToCompact : reader + " was already marked compacted";
+ }
+ catch (Throwable t)
+ {
+ accumulate = merge(accumulate, t);
+ }
+ }
+ return accumulate;
+ }
+
+ /**
+ * @return the identity function, as a Map, with domain of the provided values
+ */
+ static <T> Map<T, T> identityMap(Iterable<T> values)
+ {
+ ImmutableMap.Builder<T, T> builder = ImmutableMap.<T, T>builder();
+ for (T t : values)
+ builder.put(t, t);
+ return builder.build();
+ }
+
+ /**
+ * @return an Iterable of the union if the sets, with duplicates being represented by their first encountered instance
+ * (as defined by the order of set provision)
+ */
+ static <T> Iterable<T> concatUniq(Set<T>... sets)
+ {
+ List<Predicate<T>> notIn = new ArrayList<>(sets.length);
+ for (Set<T> set : sets)
+ notIn.add(not(in(set)));
+ List<Iterable<T>> results = new ArrayList<>(sets.length);
+ for (int i = 0 ; i < sets.length ; i++)
+ results.add(filter(sets[i], and(notIn.subList(0, i))));
+ return concat(results);
+ }
+
+ /**
+ * @return a Predicate yielding true for an item present in NONE of the provided sets
+ */
+ static <T> Predicate<T> notIn(Set<T>... sets)
+ {
+ return not(orIn(sets));
+ }
+
+ /**
+ * @return a Predicate yielding true for an item present in ANY of the provided sets
+ */
+ static <T> Predicate<T> orIn(Collection<T>... sets)
+ {
+ Predicate<T>[] orIn = new Predicate[sets.length];
+ for (int i = 0 ; i < orIn.length ; i++)
+ orIn[i] = in(sets[i]);
+ return or(orIn);
+ }
+
+ /**
+ * filter out (i.e. remove) matching elements
+ * @return filter, filtered to only those elements that *are not* present in *any* of the provided sets (are present in none)
+ */
+ static <T> Iterable<T> filterOut(Iterable<T> filter, Set<T>... inNone)
+ {
+ return filter(filter, notIn(inNone));
+ }
+
+ /**
+ * filter in (i.e. retain)
+ *
+ * @return filter, filtered to only those elements that *are* present in *any* of the provided sets
+ */
+ static <T> Iterable<T> filterIn(Iterable<T> filter, Set<T>... inAny)
+ {
+ return filter(filter, orIn(inAny));
+ }
+
+ static Set<SSTableReader> emptySet()
+ {
+ return Collections.emptySet();
+ }
+
+ static <T> T select(T t, Collection<T> col)
+ {
+ if (col instanceof Set && !col.contains(t))
+ return null;
+ return getFirst(filter(col, equalTo(t)), null);
+ }
+
+ static <T> T selectFirst(T t, Collection<T> ... sets)
+ {
+ for (Collection<T> set : sets)
+ {
+ T select = select(t, set);
+ if (select != null)
+ return select;
+ }
+ return null;
+ }
+
+ static <T> Predicate<T> idIn(Set<T> set)
+ {
+ return idIn(identityMap(set));
+ }
+
+ static <T> Predicate<T> idIn(final Map<T, T> identityMap)
+ {
+ return new Predicate<T>()
+ {
+ public boolean apply(T t)
+ {
+ return identityMap.get(t) == t;
+ }
+ };
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/2357589d/src/java/org/apache/cassandra/db/lifecycle/LifecycleTransaction.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/lifecycle/LifecycleTransaction.java
index acc9747,0000000..f3091d5
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/db/lifecycle/LifecycleTransaction.java
+++ b/src/java/org/apache/cassandra/db/lifecycle/LifecycleTransaction.java
@@@ -1,511 -1,0 +1,512 @@@
+/*
+ * 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.cassandra.db.lifecycle;
+
+import java.util.*;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Function;
+import com.google.common.base.Predicate;
+import com.google.common.collect.*;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.db.compaction.OperationType;
+import org.apache.cassandra.io.sstable.format.SSTableReader;
+import org.apache.cassandra.io.sstable.format.SSTableReader.UniqueIdentifier;
+import org.apache.cassandra.utils.concurrent.Transactional;
+
+import static com.google.common.base.Functions.compose;
+import static com.google.common.base.Predicates.*;
+import static com.google.common.collect.ImmutableSet.copyOf;
+import static com.google.common.collect.Iterables.*;
+import static java.util.Collections.singleton;
+import static org.apache.cassandra.db.lifecycle.Helpers.*;
+import static org.apache.cassandra.db.lifecycle.View.updateCompacting;
+import static org.apache.cassandra.db.lifecycle.View.updateLiveSet;
+import static org.apache.cassandra.utils.Throwables.maybeFail;
+import static org.apache.cassandra.utils.concurrent.Refs.release;
+import static org.apache.cassandra.utils.concurrent.Refs.selfRefs;
+
+public class LifecycleTransaction extends Transactional.AbstractTransactional
+{
+ private static final Logger logger = LoggerFactory.getLogger(LifecycleTransaction.class);
+
+ /**
+ * a class that represents accumulated modifications to the Tracker.
+ * has two instances, one containing modifications that are "staged" (i.e. invisible)
+ * and one containing those "logged" that have been made visible through a call to checkpoint()
+ */
+ private static class State
+ {
+ // readers that are either brand new, update a previous new reader, or update one of the original readers
+ final Set<SSTableReader> update = new HashSet<>();
+ // disjoint from update, represents a subset of originals that is no longer needed
+ final Set<SSTableReader> obsolete = new HashSet<>();
+
+ void log(State staged)
+ {
+ update.removeAll(staged.obsolete);
+ update.removeAll(staged.update);
+ update.addAll(staged.update);
+ obsolete.addAll(staged.obsolete);
+ }
+
+ boolean contains(SSTableReader reader)
+ {
+ return update.contains(reader) || obsolete.contains(reader);
+ }
+
+ boolean isEmpty()
+ {
+ return update.isEmpty() && obsolete.isEmpty();
+ }
+
+ void clear()
+ {
+ update.clear();
+ obsolete.clear();
+ }
+ }
+
+ public final Tracker tracker;
+ private final OperationType operationType;
+ // the original readers this transaction was opened over, and that it guards
+ // (no other transactions may operate over these readers concurrently)
+ private final Set<SSTableReader> originals = new HashSet<>();
+ // the set of readers we've marked as compacting (only updated on creation and in checkpoint())
+ private final Set<SSTableReader> marked = new HashSet<>();
+ // the identity set of readers we've ever encountered; used to ensure we don't accidentally revisit the
+ // same version of a reader. potentially a dangerous property if there are reference counting bugs
+ // as they won't be caught until the transaction's lifespan is over.
+ private final Set<UniqueIdentifier> identities = Collections.newSetFromMap(new IdentityHashMap<UniqueIdentifier, Boolean>());
+
+ // changes that have been made visible
+ private final State logged = new State();
+ // changes that are pending
+ private final State staged = new State();
+
+ /**
+ * construct a Transaction for use in an offline operation
+ */
+ public static LifecycleTransaction offline(OperationType operationType, SSTableReader reader)
+ {
+ return offline(operationType, singleton(reader));
+ }
+
+ /**
+ * construct a Transaction for use in an offline operation
+ */
+ public static LifecycleTransaction offline(OperationType operationType, Iterable<SSTableReader> readers)
+ {
+ // if offline, for simplicity we just use a dummy tracker
+ Tracker dummy = new Tracker(null, false);
+ dummy.addInitialSSTables(readers);
+ dummy.apply(updateCompacting(emptySet(), readers));
+ return new LifecycleTransaction(dummy, operationType, readers);
+ }
+
+ LifecycleTransaction(Tracker tracker, OperationType operationType, Iterable<SSTableReader> readers)
+ {
+ this.tracker = tracker;
+ this.operationType = operationType;
+ for (SSTableReader reader : readers)
+ {
+ originals.add(reader);
+ marked.add(reader);
+ identities.add(reader.instanceId);
+ }
+ }
+
+ public void doPrepare()
+ {
+ // note for future: in anticompaction two different operations use the same Transaction, and both prepareToCommit()
+ // separately: the second prepareToCommit is ignored as a "redundant" transition. since it is only a checkpoint
+ // (and these happen anyway) this is fine but if more logic gets inserted here than is performed in a checkpoint,
+ // it may break this use case, and care is needed
+ checkpoint();
+ }
+
+ /**
+ * point of no return: commit all changes, but leave all readers marked as compacting
+ */
+ public Throwable doCommit(Throwable accumulate)
+ {
+ assert staged.isEmpty() : "must be no actions introduced between prepareToCommit and a commit";
+
+ logger.debug("Committing update:{}, obsolete:{}", staged.update, staged.obsolete);
+
+ // this is now the point of no return; we cannot safely rollback, so we ignore exceptions until we're done
+ // we restore state by obsoleting our obsolete files, releasing our references to them, and updating our size
+ // and notification status for the obsolete and new files
- accumulate = setupDeleteNotification(logged.update, tracker, accumulate);
- accumulate = markObsolete(logged.obsolete, accumulate);
++ accumulate = markObsolete(tracker, logged.obsolete, accumulate);
+ accumulate = tracker.updateSizeTracking(logged.obsolete, logged.update, accumulate);
+ accumulate = release(selfRefs(logged.obsolete), accumulate);
+ accumulate = tracker.notifySSTablesChanged(originals, logged.update, operationType, accumulate);
+ return accumulate;
+ }
+
+ /**
+ * undo all of the changes made by this transaction, resetting the state to its original form
+ */
+ public Throwable doAbort(Throwable accumulate)
+ {
+ if (logger.isDebugEnabled())
+ logger.debug("Aborting transaction over {}, with ({},{}) logged and ({},{}) staged", originals, logged.update, logged.obsolete, staged.update, staged.obsolete);
+
+ if (logged.isEmpty() && staged.isEmpty())
+ return accumulate;
+
+ // mark obsolete all readers that are not versions of those present in the original set
+ Iterable<SSTableReader> obsolete = filterOut(concatUniq(staged.update, logged.update), originals);
+ logger.debug("Obsoleting {}", obsolete);
- accumulate = markObsolete(obsolete, accumulate);
++ // we don't pass the tracker in for the obsoletion, since these readers have never been notified externally
++ // nor had their size accounting affected
++ accumulate = markObsolete(null, obsolete, accumulate);
+
+ // replace all updated readers with a version restored to its original state
+ accumulate = tracker.apply(updateLiveSet(logged.update, restoreUpdatedOriginals()), accumulate);
+ // setReplaced immediately preceding versions that have not been obsoleted
+ accumulate = setReplaced(logged.update, accumulate);
+ // we have replaced all of logged.update and never made visible staged.update,
+ // and the files we have logged as obsolete we clone fresh versions of, so they are no longer needed either
+ // any _staged_ obsoletes should either be in staged.update already, and dealt with there,
+ // or is still in its original form (so left as is); in either case no extra action is needed
+ accumulate = release(selfRefs(concat(staged.update, logged.update, logged.obsolete)), accumulate);
+ logged.clear();
+ staged.clear();
+ return accumulate;
+ }
+
+ @Override
+ protected Throwable doPostCleanup(Throwable accumulate)
+ {
+ return unmarkCompacting(marked, accumulate);
+ }
+
+ public void permitRedundantTransitions()
+ {
+ super.permitRedundantTransitions();
+ }
+
+ /**
+ * call when a consistent batch of changes is ready to be made atomically visible
+ * these will be exposed in the Tracker atomically, or an exception will be thrown; in this case
+ * the transaction should be rolled back
+ */
+ public void checkpoint()
+ {
+ maybeFail(checkpoint(null));
+ }
+ private Throwable checkpoint(Throwable accumulate)
+ {
+ if (logger.isDebugEnabled())
+ logger.debug("Checkpointing update:{}, obsolete:{}", staged.update, staged.obsolete);
+
+ if (staged.isEmpty())
+ return accumulate;
+
+ Set<SSTableReader> toUpdate = toUpdate();
+ Set<SSTableReader> fresh = copyOf(fresh());
+
+ // check the current versions of the readers we're replacing haven't somehow been replaced by someone else
+ checkNotReplaced(filterIn(toUpdate, staged.update));
+
+ // ensure any new readers are in the compacting set, since we aren't done with them yet
+ // and don't want anyone else messing with them
+ // apply atomically along with updating the live set of readers
+ tracker.apply(compose(updateCompacting(emptySet(), fresh),
+ updateLiveSet(toUpdate, staged.update)));
+
+ // log the staged changes and our newly marked readers
+ marked.addAll(fresh);
+ logged.log(staged);
+
+ // setup our tracker, and mark our prior versions replaced, also releasing our references to them
+ // we do not replace/release obsoleted readers, since we may need to restore them on rollback
+ accumulate = setReplaced(filterOut(toUpdate, staged.obsolete), accumulate);
+ accumulate = release(selfRefs(filterOut(toUpdate, staged.obsolete)), accumulate);
+
+ staged.clear();
+ return accumulate;
+ }
+
+ /**
+ * update a reader: if !original, this is a reader that is being introduced by this transaction;
+ * otherwise it must be in the originals() set, i.e. a reader guarded by this transaction
+ */
+ public void update(SSTableReader reader, boolean original)
+ {
+ assert !staged.update.contains(reader) : "each reader may only be updated once per checkpoint: " + reader;
+ assert !identities.contains(reader.instanceId) : "each reader instance may only be provided as an update once: " + reader;
+ // check it isn't obsolete, and that it matches the original flag
+ assert !(logged.obsolete.contains(reader) || staged.obsolete.contains(reader)) : "may not update a reader that has been obsoleted";
+ assert original == originals.contains(reader) : String.format("the 'original' indicator was incorrect (%s provided): %s", original, reader);
+ staged.update.add(reader);
+ identities.add(reader.instanceId);
+ reader.setupKeyCache();
+ }
+
+ /**
+ * mark this reader as for obsoletion. this does not actually obsolete the reader until commit() is called,
+ * but on checkpoint() the reader will be removed from the live set
+ */
+ public void obsolete(SSTableReader reader)
+ {
+ logger.debug("Staging for obsolescence {}", reader);
+ // check this is: a reader guarded by the transaction, an instance we have already worked with
+ // and that we haven't already obsoleted it, nor do we have other changes staged for it
+ assert identities.contains(reader.instanceId) : "only reader instances that have previously been provided may be obsoleted: " + reader;
+ assert originals.contains(reader) : "only readers in the 'original' set may be obsoleted: " + reader + " vs " + originals;
+ assert !(logged.obsolete.contains(reader) || staged.obsolete.contains(reader)) : "may not obsolete a reader that has already been obsoleted: " + reader;
+ assert !staged.update.contains(reader) : "may not obsolete a reader that has a staged update (must checkpoint first): " + reader;
+ assert current(reader) == reader : "may only obsolete the latest version of the reader: " + reader;
+ staged.obsolete.add(reader);
+ }
+
+ /**
+ * obsolete every file in the original transaction
+ */
+ public void obsoleteOriginals()
+ {
+ logger.debug("Staging for obsolescence {}", originals);
+ // if we're obsoleting, we should have no staged updates for the original files
+ assert Iterables.isEmpty(filterIn(staged.update, originals)) : staged.update;
+
+ // stage obsoletes for any currently visible versions of any original readers
+ Iterables.addAll(staged.obsolete, filterIn(current(), originals));
+ }
+
+ /**
+ * return the readers we're replacing in checkpoint(), i.e. the currently visible version of those in staged
+ */
+ private Set<SSTableReader> toUpdate()
+ {
+ return copyOf(filterIn(current(), staged.obsolete, staged.update));
+ }
+
+ /**
+ * new readers that haven't appeared previously (either in the original set or the logged updates)
+ */
+ private Iterable<SSTableReader> fresh()
+ {
+ return filterOut(staged.update,
+ originals, logged.update);
+ }
+
+ /**
+ * returns the currently visible readers managed by this transaction
+ */
+ public Iterable<SSTableReader> current()
+ {
+ // i.e., those that are updates that have been logged (made visible),
+ // and any original readers that have neither been obsoleted nor updated
+ return concat(logged.update, filterOut(originals, logged.update, logged.obsolete));
+ }
+
+ /**
+ * update the current replacement of any original reader back to its original start
+ */
+ private List<SSTableReader> restoreUpdatedOriginals()
+ {
+ Iterable<SSTableReader> torestore = filterIn(originals, logged.update, logged.obsolete);
+ return ImmutableList.copyOf(transform(torestore,
+ new Function<SSTableReader, SSTableReader>()
+ {
+ public SSTableReader apply(SSTableReader reader)
+ {
+ return current(reader).cloneWithNewStart(reader.first, null);
+ }
+ }));
+ }
+
+ /**
+ * the set of readers guarded by this transaction _in their original instance/state_
+ * call current(SSTableReader) on any reader in this set to get the latest instance
+ */
+ public Set<SSTableReader> originals()
+ {
+ return Collections.unmodifiableSet(originals);
+ }
+
+ /**
+ * indicates if the reader has been marked for obsoletion
+ */
+ public boolean isObsolete(SSTableReader reader)
+ {
+ return logged.obsolete.contains(reader) || staged.obsolete.contains(reader);
+ }
+
+ /**
+ * return the current version of the provided reader, whether or not it is visible or staged;
+ * i.e. returns the first version present by testing staged, logged and originals in order.
+ */
+ public SSTableReader current(SSTableReader reader)
+ {
+ Set<SSTableReader> container;
+ if (staged.contains(reader))
+ container = staged.update.contains(reader) ? staged.update : staged.obsolete;
+ else if (logged.contains(reader))
+ container = logged.update.contains(reader) ? logged.update : logged.obsolete;
+ else if (originals.contains(reader))
+ container = originals;
+ else throw new AssertionError();
+ return select(reader, container);
+ }
+
+ /**
+ * remove the reader from the set we're modifying
+ */
+ public void cancel(SSTableReader cancel)
+ {
+ logger.debug("Cancelling {} from transaction", cancel);
+ assert originals.contains(cancel) : "may only cancel a reader in the 'original' set: " + cancel + " vs " + originals;
+ assert !(staged.contains(cancel) || logged.contains(cancel)) : "may only cancel a reader that has not been updated or obsoleted in this transaction: " + cancel;
+ originals.remove(cancel);
+ marked.remove(cancel);
+ maybeFail(unmarkCompacting(singleton(cancel), null));
+ }
+
+ /**
+ * remove the readers from the set we're modifying
+ */
+ public void cancel(Iterable<SSTableReader> cancels)
+ {
+ for (SSTableReader cancel : cancels)
+ cancel(cancel);
+ }
+
+ /**
+ * remove the provided readers from this Transaction, and return a new Transaction to manage them
+ * only permitted to be called if the current Transaction has never been used
+ */
+ public LifecycleTransaction split(Collection<SSTableReader> readers)
+ {
+ logger.debug("Splitting {} into new transaction", readers);
+ checkUnused();
+ for (SSTableReader reader : readers)
+ assert identities.contains(reader.instanceId) : "may only split the same reader instance the transaction was opened with: " + reader;
+
+ for (SSTableReader reader : readers)
+ {
+ identities.remove(reader.instanceId);
+ originals.remove(reader);
+ marked.remove(reader);
+ }
+ return new LifecycleTransaction(tracker, operationType, readers);
+ }
+
+ /**
+ * check this transaction has never been used
+ */
+ private void checkUnused()
+ {
+ assert logged.isEmpty();
+ assert staged.isEmpty();
+ assert identities.size() == originals.size();
+ assert originals.size() == marked.size();
+ }
+
+ private Throwable unmarkCompacting(Set<SSTableReader> unmark, Throwable accumulate)
+ {
+ accumulate = tracker.apply(updateCompacting(unmark, emptySet()), accumulate);
+ // when the CFS is invalidated, it will call unreferenceSSTables(). However, unreferenceSSTables only deals
+ // with sstables that aren't currently being compacted. If there are ongoing compactions that finish or are
+ // interrupted after the CFS is invalidated, those sstables need to be unreferenced as well, so we do that here.
+ accumulate = tracker.dropSSTablesIfInvalid(accumulate);
+ return accumulate;
+ }
+
+ // convenience method for callers that know only one sstable is involved in the transaction
+ public SSTableReader onlyOne()
+ {
+ assert originals.size() == 1;
+ return getFirst(originals, null);
+ }
+
+ // a class representing the current state of the reader within this transaction, encoding the actions both logged
+ // and pending, and the reader instances that are visible now, and will be after the next checkpoint (with null
+ // indicating either obsolescence, or that the reader does not occur in the transaction; which is defined
+ // by the corresponding Action)
+ @VisibleForTesting
+ public static class ReaderState
+ {
+ public static enum Action
+ {
+ UPDATED, OBSOLETED, NONE;
+ public static Action get(boolean updated, boolean obsoleted)
+ {
+ assert !(updated && obsoleted);
+ return updated ? UPDATED : obsoleted ? OBSOLETED : NONE;
+ }
+ }
+
+ final Action staged;
+ final Action logged;
+ final SSTableReader nextVisible;
+ final SSTableReader currentlyVisible;
+ final boolean original;
+
+ public ReaderState(Action logged, Action staged, SSTableReader currentlyVisible, SSTableReader nextVisible, boolean original)
+ {
+ this.staged = staged;
+ this.logged = logged;
+ this.currentlyVisible = currentlyVisible;
+ this.nextVisible = nextVisible;
+ this.original = original;
+ }
+
+ public boolean equals(Object that)
+ {
+ return that instanceof ReaderState && equals((ReaderState) that);
+ }
+
+ public boolean equals(ReaderState that)
+ {
+ return this.staged == that.staged && this.logged == that.logged && this.original == that.original
+ && this.currentlyVisible == that.currentlyVisible && this.nextVisible == that.nextVisible;
+ }
+
+ public String toString()
+ {
+ return String.format("[logged=%s staged=%s original=%s]", logged, staged, original);
+ }
+
+ public static SSTableReader visible(SSTableReader reader, Predicate<SSTableReader> obsolete, Collection<SSTableReader> ... selectFrom)
+ {
+ return obsolete.apply(reader) ? null : selectFirst(reader, selectFrom);
+ }
+ }
+
+ @VisibleForTesting
+ public ReaderState state(SSTableReader reader)
+ {
+ SSTableReader currentlyVisible = ReaderState.visible(reader, in(logged.obsolete), logged.update, originals);
+ SSTableReader nextVisible = ReaderState.visible(reader, orIn(staged.obsolete, logged.obsolete), staged.update, logged.update, originals);
+ return new ReaderState(ReaderState.Action.get(logged.update.contains(reader), logged.obsolete.contains(reader)),
+ ReaderState.Action.get(staged.update.contains(reader), staged.obsolete.contains(reader)),
+ currentlyVisible, nextVisible, originals.contains(reader)
+ );
+ }
+
+ public String toString()
+ {
+ return originals.toString();
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/2357589d/src/java/org/apache/cassandra/db/lifecycle/Tracker.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/lifecycle/Tracker.java
index 50f567f,0000000..a32dd87
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/db/lifecycle/Tracker.java
+++ b/src/java/org/apache/cassandra/db/lifecycle/Tracker.java
@@@ -1,468 -1,0 +1,459 @@@
+/*
+ * 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.cassandra.db.lifecycle;
+
+import java.io.File;
+import java.util.*;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicReference;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Function;
+import com.google.common.base.Predicate;
+import com.google.common.base.Predicates;
+import com.google.common.collect.*;
+
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.Directories;
+import org.apache.cassandra.db.Memtable;
+import org.apache.cassandra.db.commitlog.ReplayPosition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.compaction.OperationType;
+import org.apache.cassandra.io.sstable.format.SSTableReader;
+import org.apache.cassandra.io.util.FileUtils;
+import org.apache.cassandra.metrics.StorageMetrics;
+import org.apache.cassandra.notifications.*;
+import org.apache.cassandra.utils.Pair;
+import org.apache.cassandra.utils.concurrent.OpOrder;
+
+import static com.google.common.base.Predicates.and;
+import static com.google.common.collect.ImmutableSet.copyOf;
+import static com.google.common.collect.Iterables.filter;
+import static java.util.Collections.singleton;
+import static org.apache.cassandra.db.lifecycle.Helpers.*;
+import static org.apache.cassandra.db.lifecycle.View.permitCompacting;
+import static org.apache.cassandra.db.lifecycle.View.updateCompacting;
+import static org.apache.cassandra.db.lifecycle.View.updateLiveSet;
+import static org.apache.cassandra.utils.Throwables.maybeFail;
+import static org.apache.cassandra.utils.Throwables.merge;
+import static org.apache.cassandra.utils.concurrent.Refs.release;
+import static org.apache.cassandra.utils.concurrent.Refs.selfRefs;
+
+public class Tracker
+{
+ private static final Logger logger = LoggerFactory.getLogger(Tracker.class);
+
+ public final Collection<INotificationConsumer> subscribers = new CopyOnWriteArrayList<>();
+ public final ColumnFamilyStore cfstore;
+ final AtomicReference<View> view;
+ public final boolean loadsstables;
+
+ public Tracker(ColumnFamilyStore cfstore, boolean loadsstables)
+ {
+ this.cfstore = cfstore;
+ this.view = new AtomicReference<>();
+ this.loadsstables = loadsstables;
+ this.reset();
+ }
+
+ public LifecycleTransaction tryModify(SSTableReader sstable, OperationType operationType)
+ {
+ return tryModify(singleton(sstable), operationType);
+ }
+
+ /**
+ * @return a Transaction over the provided sstables if we are able to mark the given @param sstables as compacted, before anyone else
+ */
+ public LifecycleTransaction tryModify(Iterable<SSTableReader> sstables, OperationType operationType)
+ {
+ if (Iterables.isEmpty(sstables))
+ return new LifecycleTransaction(this, operationType, sstables);
+ if (null == apply(permitCompacting(sstables), updateCompacting(emptySet(), sstables)))
+ return null;
+ return new LifecycleTransaction(this, operationType, sstables);
+ }
+
+
+ // METHODS FOR ATOMICALLY MODIFYING THE VIEW
+
+ Pair<View, View> apply(Function<View, View> function)
+ {
+ return apply(Predicates.<View>alwaysTrue(), function);
+ }
+
+ Throwable apply(Function<View, View> function, Throwable accumulate)
+ {
+ try
+ {
+ apply(function);
+ }
+ catch (Throwable t)
+ {
+ accumulate = merge(accumulate, t);
+ }
+ return accumulate;
+ }
+
+ /**
+ * atomically tests permit against the view and applies function to it, if permit yields true, returning the original;
+ * otherwise the method aborts, returning null
+ */
+ Pair<View, View> apply(Predicate<View> permit, Function<View, View> function)
+ {
+ while (true)
+ {
+ View cur = view.get();
+ if (!permit.apply(cur))
+ return null;
+ View updated = function.apply(cur);
+ if (view.compareAndSet(cur, updated))
+ return Pair.create(cur, updated);
+ }
+ }
+
+ Throwable updateSizeTracking(Iterable<SSTableReader> oldSSTables, Iterable<SSTableReader> newSSTables, Throwable accumulate)
+ {
+ if (cfstore == null)
+ return accumulate;
+
+ long add = 0;
+ for (SSTableReader sstable : newSSTables)
+ {
+ if (logger.isDebugEnabled())
+ logger.debug("adding {} to list of files tracked for {}.{}", sstable.descriptor, cfstore.keyspace.getName(), cfstore.name);
+ try
+ {
+ add += sstable.bytesOnDisk();
+ }
+ catch (Throwable t)
+ {
+ accumulate = merge(accumulate, t);
+ }
+ }
+ long subtract = 0;
+ for (SSTableReader sstable : oldSSTables)
+ {
+ if (logger.isDebugEnabled())
+ logger.debug("removing {} from list of files tracked for {}.{}", sstable.descriptor, cfstore.keyspace.getName(), cfstore.name);
+ try
+ {
+ subtract += sstable.bytesOnDisk();
+ }
+ catch (Throwable t)
+ {
+ accumulate = merge(accumulate, t);
+ }
+ }
+ StorageMetrics.load.inc(add - subtract);
+ cfstore.metric.liveDiskSpaceUsed.inc(add - subtract);
+ // we don't subtract from total until the sstable is deleted
+ cfstore.metric.totalDiskSpaceUsed.inc(add);
+ return accumulate;
+ }
+
+ // SETUP / CLEANUP
+
+ public void addInitialSSTables(Iterable<SSTableReader> sstables)
+ {
- maybeFail(setupDeleteNotification(sstables, this, null));
+ apply(updateLiveSet(emptySet(), sstables));
+ maybeFail(updateSizeTracking(emptySet(), sstables, null));
+ // no notifications or backup necessary
+ }
+
+ public void addSSTables(Iterable<SSTableReader> sstables)
+ {
+ addInitialSSTables(sstables);
+ for (SSTableReader sstable : sstables)
+ {
+ maybeIncrementallyBackup(sstable);
+ notifyAdded(sstable);
+ }
+ }
+
+ /** (Re)initializes the tracker, purging all references. */
+ @VisibleForTesting
+ public void reset()
+ {
+ view.set(new View(
+ cfstore != null ? ImmutableList.of(new Memtable(cfstore)) : Collections.<Memtable>emptyList(),
+ ImmutableList.<Memtable>of(),
+ Collections.<SSTableReader, SSTableReader>emptyMap(),
+ Collections.<SSTableReader>emptySet(),
+ SSTableIntervalTree.empty()));
+ }
+
+ public Throwable dropSSTablesIfInvalid(Throwable accumulate)
+ {
+ if (cfstore != null && !cfstore.isValid())
+ accumulate = dropSSTables(accumulate);
+ return accumulate;
+ }
+
+ public void dropSSTables()
+ {
+ maybeFail(dropSSTables(null));
+ }
+
+ public Throwable dropSSTables(Throwable accumulate)
+ {
+ return dropSSTables(Predicates.<SSTableReader>alwaysTrue(), OperationType.UNKNOWN, accumulate);
+ }
+
+ /**
+ * removes all sstables that are not busy compacting.
+ */
+ public Throwable dropSSTables(final Predicate<SSTableReader> remove, OperationType operationType, Throwable accumulate)
+ {
+ Pair<View, View> result = apply(new Function<View, View>()
+ {
+ public View apply(View view)
+ {
+ Set<SSTableReader> toremove = copyOf(filter(view.sstables, and(remove, notIn(view.compacting))));
+ return updateLiveSet(toremove, emptySet()).apply(view);
+ }
+ });
+
+ Set<SSTableReader> removed = Sets.difference(result.left.sstables, result.right.sstables);
+ assert Iterables.all(removed, remove);
+
+ if (!removed.isEmpty())
+ {
+ // notifySSTablesChanged -> LeveledManifest.promote doesn't like a no-op "promotion"
+ accumulate = notifySSTablesChanged(removed, Collections.<SSTableReader>emptySet(), operationType, accumulate);
+ accumulate = updateSizeTracking(removed, emptySet(), accumulate);
- accumulate = markObsolete(removed, accumulate);
++ accumulate = markObsolete(this, removed, accumulate);
+ accumulate = release(selfRefs(removed), accumulate);
+ }
+ return accumulate;
+ }
+
+ /**
+ * Removes every SSTable in the directory from the Tracker's view.
+ * @param directory the unreadable directory, possibly with SSTables in it, but not necessarily.
+ */
+ public void removeUnreadableSSTables(final File directory)
+ {
+ maybeFail(dropSSTables(new Predicate<SSTableReader>()
+ {
+ public boolean apply(SSTableReader reader)
+ {
+ return reader.descriptor.directory.equals(directory);
+ }
+ }, OperationType.UNKNOWN, null));
+ }
+
+
+
+ // FLUSHING
+
+ /**
+ * get the Memtable that the ordered writeOp should be directed to
+ */
+ public Memtable getMemtableFor(OpOrder.Group opGroup, ReplayPosition replayPosition)
+ {
+ // since any new memtables appended to the list after we fetch it will be for operations started
+ // after us, we can safely assume that we will always find the memtable that 'accepts' us;
+ // if the barrier for any memtable is set whilst we are reading the list, it must accept us.
+
+ // there may be multiple memtables in the list that would 'accept' us, however we only ever choose
+ // the oldest such memtable, as accepts() only prevents us falling behind (i.e. ensures we don't
+ // assign operations to a memtable that was retired/queued before we started)
+ for (Memtable memtable : view.get().liveMemtables)
+ {
+ if (memtable.accepts(opGroup, replayPosition))
+ return memtable;
+ }
+ throw new AssertionError(view.get().liveMemtables.toString());
+ }
+
+ /**
+ * Switch the current memtable. This atomically appends a new memtable to the end of the list of active memtables,
+ * returning the previously last memtable. It leaves the previous Memtable in the list of live memtables until
+ * discarding(memtable) is called. These two methods must be synchronized/paired, i.e. m = switchMemtable
+ * must be followed by discarding(m), they cannot be interleaved.
+ *
+ * @return the previously active memtable
+ */
+ public Memtable switchMemtable(boolean truncating)
+ {
+ Memtable newMemtable = new Memtable(cfstore);
+ Pair<View, View> result = apply(View.switchMemtable(newMemtable));
+ if (truncating)
+ notifyRenewed(newMemtable);
+
+ return result.left.getCurrentMemtable();
+ }
+
+ public void markFlushing(Memtable memtable)
+ {
+ apply(View.markFlushing(memtable));
+ }
+
+ public void replaceFlushed(Memtable memtable, SSTableReader sstable)
+ {
+ if (sstable == null)
+ {
+ // sstable may be null if we flushed batchlog and nothing needed to be retained
+ // if it's null, we don't care what state the cfstore is in, we just replace it and continue
+ apply(View.replaceFlushed(memtable, null));
+ return;
+ }
+
- sstable.setupDeleteNotification(this);
+ sstable.setupKeyCache();
+ // back up before creating a new Snapshot (which makes the new one eligible for compaction)
+ maybeIncrementallyBackup(sstable);
+
+ apply(View.replaceFlushed(memtable, sstable));
+
+ Throwable fail;
+ fail = updateSizeTracking(emptySet(), singleton(sstable), null);
+ // TODO: if we're invalidated, should we notifyadded AND removed, or just skip both?
+ fail = notifyAdded(sstable, fail);
+
+ if (cfstore != null && !cfstore.isValid())
+ dropSSTables();
+
+ maybeFail(fail);
+ }
+
+
+
+ // MISCELLANEOUS public utility calls
+
+ public Set<SSTableReader> getSSTables()
+ {
+ return view.get().sstables;
+ }
+
+ public Set<SSTableReader> getCompacting()
+ {
+ return view.get().compacting;
+ }
+
+ public Set<SSTableReader> getUncompacting()
+ {
+ return view.get().nonCompactingSStables();
+ }
+
+ public Iterable<SSTableReader> getUncompacting(Iterable<SSTableReader> candidates)
+ {
+ return view.get().getUncompacting(candidates);
+ }
+
+ public void maybeIncrementallyBackup(final SSTableReader sstable)
+ {
+ if (!DatabaseDescriptor.isIncrementalBackupsEnabled())
+ return;
+
+ File backupsDir = Directories.getBackupsDirectory(sstable.descriptor);
+ sstable.createLinks(FileUtils.getCanonicalPath(backupsDir));
+ }
+
- public void spaceReclaimed(long size)
- {
- if (cfstore != null)
- cfstore.metric.totalDiskSpaceUsed.dec(size);
- }
-
-
+
+ // NOTIFICATION
+
+ Throwable notifySSTablesChanged(Collection<SSTableReader> removed, Collection<SSTableReader> added, OperationType compactionType, Throwable accumulate)
+ {
+ INotification notification = new SSTableListChangedNotification(added, removed, compactionType);
+ for (INotificationConsumer subscriber : subscribers)
+ {
+ try
+ {
+ subscriber.handleNotification(notification, this);
+ }
+ catch (Throwable t)
+ {
+ accumulate = merge(accumulate, t);
+ }
+ }
+ return accumulate;
+ }
+
+ Throwable notifyAdded(SSTableReader added, Throwable accumulate)
+ {
+ INotification notification = new SSTableAddedNotification(added);
+ for (INotificationConsumer subscriber : subscribers)
+ {
+ try
+ {
+ subscriber.handleNotification(notification, this);
+ }
+ catch (Throwable t)
+ {
+ accumulate = merge(accumulate, t);
+ }
+ }
+ return accumulate;
+ }
+
+ public void notifyAdded(SSTableReader added)
+ {
+ maybeFail(notifyAdded(added, null));
+ }
+
+ public void notifySSTableRepairedStatusChanged(Collection<SSTableReader> repairStatusesChanged)
+ {
+ INotification notification = new SSTableRepairStatusChanged(repairStatusesChanged);
+ for (INotificationConsumer subscriber : subscribers)
+ subscriber.handleNotification(notification, this);
+ }
+
+ public void notifyDeleting(SSTableReader deleting)
+ {
+ INotification notification = new SSTableDeletingNotification(deleting);
+ for (INotificationConsumer subscriber : subscribers)
+ subscriber.handleNotification(notification, this);
+ }
+
+ public void notifyRenewed(Memtable renewed)
+ {
+ INotification notification = new MemtableRenewedNotification(renewed);
+ for (INotificationConsumer subscriber : subscribers)
+ subscriber.handleNotification(notification, this);
+ }
+
+ public void notifyTruncated(long truncatedAt)
+ {
+ INotification notification = new TruncationNotification(truncatedAt);
+ for (INotificationConsumer subscriber : subscribers)
+ subscriber.handleNotification(notification, this);
+ }
+
+ public void subscribe(INotificationConsumer consumer)
+ {
+ subscribers.add(consumer);
+ }
+
+ public void unsubscribe(INotificationConsumer consumer)
+ {
+ subscribers.remove(consumer);
+ }
+
+ private static Set<SSTableReader> emptySet()
+ {
+ return Collections.emptySet();
+ }
+
+ public View getView()
+ {
+ return view.get();
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/2357589d/src/java/org/apache/cassandra/io/sstable/SSTableDeletingTask.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/io/sstable/SSTableDeletingTask.java
index cc837ba,8703494..ec6439c
--- a/src/java/org/apache/cassandra/io/sstable/SSTableDeletingTask.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableDeletingTask.java
@@@ -23,16 -23,15 +23,16 @@@ import java.util.Set
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.TimeUnit;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Sets;
- import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.yammer.metrics.core.Counter;
++import com.codahale.metrics.Counter;
import org.apache.cassandra.concurrent.ScheduledExecutors;
-import org.apache.cassandra.db.DataTracker;
-import org.apache.cassandra.db.SystemKeyspace;
+import org.apache.cassandra.db.lifecycle.Tracker;
import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.concurrent.Blocker;
public class SSTableDeletingTask implements Runnable
{
@@@ -43,12 -42,11 +43,12 @@@
// Additionally, we need to make sure to delete the data file first, so on restart the others
// will be recognized as GCable.
private static final Set<SSTableDeletingTask> failedTasks = new CopyOnWriteArraySet<>();
+ private static final Blocker blocker = new Blocker();
- private final SSTableReader referent;
private final Descriptor desc;
private final Set<Component> components;
- private Tracker tracker;
+ private final long bytesOnDisk;
+ private final Counter totalDiskSpaceUsed;
/**
* realDescriptor is the actual descriptor for the sstable, the descriptor inside
@@@ -93,12 -78,6 +80,7 @@@
public void run()
{
+ blocker.ask();
- long size = referent.bytesOnDisk();
-
- if (tracker != null)
- tracker.notifyDeleting(referent);
-
// If we can't successfully delete the DATA component, set the task to be retried later: see above
File datafile = new File(desc.filenameFor(Component.DATA));
if (!datafile.delete())
http://git-wip-us.apache.org/repos/asf/cassandra/blob/2357589d/src/java/org/apache/cassandra/io/sstable/SSTableRewriter.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/io/sstable/SSTableRewriter.java
index 011c7d9,92f74d7..b4b6c72
--- a/src/java/org/apache/cassandra/io/sstable/SSTableRewriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableRewriter.java
@@@ -20,9 -20,14 +20,11 @@@ package org.apache.cassandra.io.sstable
import java.util.*;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Throwables;
-import com.google.common.collect.ImmutableList;
+ import org.apache.cassandra.cache.InstrumentingCache;
+ import org.apache.cassandra.cache.KeyCacheKey;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
-import org.apache.cassandra.db.DataTracker;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.RowIndexEntry;
import org.apache.cassandra.db.compaction.AbstractCompactedRow;
@@@ -225,66 -300,109 +227,75 @@@ public class SSTableRewriter extends Tr
{
// we call getCurrentReplacement() to support multiple rewriters operating over the same source readers at once.
// note: only one such writer should be written to at any moment
- final SSTableReader latest = dataTracker.getCurrentVersion(sstable);
- SSTableReader replacement;
- if (reset)
- {
- DecoratedKey newStart = originalStarts.get(sstable.descriptor);
- replacement = latest.cloneWithNewStart(newStart, null);
- }
- else
+ final SSTableReader latest = transaction.current(sstable);
+
+ // skip any sstables that we know to already be shadowed
+ if (latest.first.compareTo(lowerbound) > 0)
+ continue;
+
- final Runnable runOnClose = new Runnable()
- {
- public void run()
- {
- // this is somewhat racey, in that we could theoretically be closing this old reader
- // when an even older reader is still in use, but it's not likely to have any major impact
- for (DecoratedKey key : invalidateKeys)
- latest.invalidateCacheKey(key);
- }
- };
-
++ Runnable runOnClose = new InvalidateKeys(latest, invalidateKeys);
+ if (lowerbound.compareTo(latest.last) >= 0)
{
- // skip any sstables that we know to already be shadowed
- if (latest.openReason == SSTableReader.OpenReason.SHADOWED)
- continue;
- if (latest.first.compareTo(lowerbound) > 0)
- continue;
-
- final Runnable runOnClose = new InvalidateKeys(latest, invalidateKeys);
- if (lowerbound.compareTo(latest.last) >= 0)
- {
- replacement = latest.cloneAsShadowed(runOnClose);
- }
- else
+ if (!transaction.isObsolete(latest))
{
- DecoratedKey newStart = latest.firstKeyBeyond(lowerbound);
- assert newStart != null;
- replacement = latest.cloneWithNewStart(newStart, runOnClose);
+ latest.runOnClose(runOnClose);
+ transaction.obsolete(latest);
}
+ continue;
}
- toReplace.add(latest);
- replaceWith.add(replacement);
- rewriting.remove(sstable);
- rewriting.add(replacement);
+ DecoratedKey newStart = latest.firstKeyBeyond(lowerbound);
+ assert newStart != null;
+ SSTableReader replacement = latest.cloneWithNewStart(newStart, runOnClose);
+ transaction.update(replacement, true);
}
- cfs.getDataTracker().replaceWithNewInstances(toReplace, replaceWith);
+ }
+
+ private static final class InvalidateKeys implements Runnable
+ {
+ final List<KeyCacheKey> cacheKeys = new ArrayList<>();
+ final InstrumentingCache<KeyCacheKey, ?> cache;
+
+ private InvalidateKeys(SSTableReader reader, Collection<DecoratedKey> invalidate)
+ {
+ this.cache = reader.getKeyCache();
+ for (DecoratedKey key : invalidate)
+ cacheKeys.add(reader.getCacheKey(key));
+ }
+
+ public void run()
+ {
+ for (KeyCacheKey key : cacheKeys)
+ cache.remove(key);
+ }
}
- private void replaceEarlyOpenedFile(SSTableReader toReplace, SSTableReader replaceWith)
- {
- if (isOffline)
- return;
- Set<SSTableReader> toReplaceSet;
- if (toReplace != null)
- {
- toReplace.setReplacedBy(replaceWith);
- toReplaceSet = Collections.singleton(toReplace);
- }
- else
- {
- dataTracker.markCompacting(Collections.singleton(replaceWith), true, isOffline);
- toReplaceSet = Collections.emptySet();
- }
- dataTracker.replaceEarlyOpenedFiles(toReplaceSet, Collections.singleton(replaceWith));
- }
-
public void switchWriter(SSTableWriter newWriter)
{
- if (writer == null)
- {
- writer = newWriter;
- return;
- }
+ if (newWriter != null)
+ writers.add(newWriter.setMaxDataAge(maxAge));
- if (writer.getFilePointer() != 0)
+ if (writer == null || writer.getFilePointer() == 0)
{
- // If early re-open is disabled, simply finalize the writer and store it
- if (preemptiveOpenInterval == Long.MAX_VALUE)
+ if (writer != null)
{
- SSTableReader reader = writer.finish(SSTableWriter.FinishType.NORMAL, maxAge, -1);
- finishedReaders.add(reader);
- }
- else
- {
- // we leave it as a tmp file, but we open it and add it to the dataTracker
- SSTableReader reader = writer.finish(SSTableWriter.FinishType.EARLY, maxAge, -1);
- replaceEarlyOpenedFile(currentlyOpenedEarly, reader);
- moveStarts(reader, reader.last, false);
- finishedEarly.add(new Finished(writer, reader));
+ writer.abort();
+ writers.remove(writer);
}
+ writer = newWriter;
+ return;
}
- else
+
+ SSTableReader reader = null;
+ if (preemptiveOpenInterval != Long.MAX_VALUE)
{
- writer.abort();
+ // we leave it as a tmp file, but we open it and add it to the Tracker
+ reader = writer.setMaxDataAge(maxAge).openFinalEarly();
+ transaction.update(reader, false);
+ moveStarts(reader, reader.last);
+ transaction.checkpoint();
}
- currentlyOpenedEarly = null;
+
currentlyOpenedEarlyAt = 0;
writer = newWriter;
}