database: Fix use and assumptions about pending compations

Fixes #934 - faulty assert in discard_sstables

run_with_compaction_disabled clears out a CF from compaction
mananger queue. discard_sstables wants to assert on this, but looks
at the wrong counters.

pending_compactions is an indicator on how much interested parties
want a CF compacted (again and again). It should not be considered
an indicator of compactions actually being done.

This modifies the usage slightly so that:
1.) The counter is always incremented, even if compaction is disallowed.
    The counters value on end of run_with_compaction_disabled is then
    instead used as an indicator as to whether a compaction should be
    re-triggered. (If compactions finished, it will be zero)
2.) Document the use and purpose of the pending counter, and add
    method to re-add CF to compaction for r_w_c_d above.
3.) discard_sstables now asserts on the right things.

Message-Id: <1456332824-23349-1-git-send-email-calle@scylladb.com>

database.cc

@@ -861,14 +861,25 @@ void column_family::start_compaction() {
 
 void column_family::trigger_compaction() {
     // Submitting compaction job to compaction manager.
+    // #934 - always inc the pending counter, to help
+    // indicate the want for compaction.
+    _stats.pending_compactions++;
+    do_trigger_compaction(); // see below
+}
+
+void column_family::do_trigger_compaction() {
+    // But only submit if we're not locked out
     if (!_compaction_disabled) {
-        _stats.pending_compactions++;
         _compaction_manager.submit(this);
     }
 }
 
 future<> column_family::run_compaction(sstables::compaction_descriptor descriptor) {
+    assert(_stats.pending_compactions > 0);
     return compact_sstables(std::move(descriptor)).then([this] {
+        // only do this on success. (no exceptions)
+        // in that case, we rely on it being still set
+        // for reqeueuing
         _stats.pending_compactions--;
     });
 }
@@ -2282,7 +2293,8 @@ void column_family::clear() {
 // NOTE: does not need to be futurized, but might eventually, depending on
 // if we implement notifications, whatnot.
 future<db::replay_position> column_family::discard_sstables(db_clock::time_point truncated_at) {
-    assert(_stats.pending_compactions == 0);
+    assert(_compaction_disabled > 0);
+    assert(!compaction_manager_queued());
 
     return with_lock(_sstables_lock.for_read(), [this, truncated_at] {
         db::replay_position rp;

database.hh

@@ -203,6 +203,7 @@ private:
     key_source sstables_as_key_source() const;
     partition_presence_checker make_partition_presence_checker(lw_shared_ptr<sstable_list> old_sstables);
     std::chrono::steady_clock::time_point _sstable_writes_disabled_at;
+    void do_trigger_compaction();
 public:
     // Creates a mutation reader which covers all data sources for this column family.
     // Caller needs to ensure that column_family remains live (FIXME: relax this).
@@ -358,8 +359,12 @@ public:
     Result run_with_compaction_disabled(Func && func) {
         ++_compaction_disabled;
         return _compaction_manager.remove(this).then(std::forward<Func>(func)).finally([this] {
-            if (--_compaction_disabled == 0) {
-                trigger_compaction();
+            // #934. The pending counter is actually a great indicator into whether we
+            // actually need to trigger a compaction again.
+            if (--_compaction_disabled == 0 && _stats.pending_compactions > 0) {
+                // we're turning if on again, use function that does not increment
+                // the counter further.
+                do_trigger_compaction();
             }
         });
     }