os/LevelDBStore: merge adjacent ranges in compactionqueue

If we get behind and multiple adjacent ranges end up in the queue, merge
them so that we fire off compaction on larger ranges.

Signed-off-by: Sage Weil <sage@inktank.com>
(cherry picked from commit f628dd0e4a5ace079568773edfab29d9f764d4f0)

diff --git a/src/os/LevelDBStore.cc b/src/os/LevelDBStore.cc
index 166a1553578f0a3233f84146aa87d803191ee928..435a8aa2f65557814b2c0b151d0a54a7243bfc13 100644 (file)
--- a/src/os/LevelDBStore.cc
+++ b/src/os/LevelDBStore.cc
@@ -157,3 +157,36 @@ void LevelDBStore::compact_thread_entry()
   }
   compact_queue_lock.Unlock();
 }
+
+void compact_range_async(const string& start, const string& end)
+{
+  Mutex::Locker l(compact_queue_lock);
+
+  // try to merge adjacent ranges.  this is O(n), but the queue should
+  // be short.
+  list< pair<string,string> >::iterator p = compact_queue.begin();
+  while (p != compact_queue.end(); ++p) {
+    if (p->first == start && p->second == end) {
+      // dup; no-op
+      return;
+    } else if (p->first <= end && p->first > start) {
+      // merge with existing range to the right
+      compact_queue.push_back(make_pair(start, p->second));
+      break;
+    } else if (p->second >= start && p->second < end) {
+      // merge with existing range to the left
+      compact_queue.push_back(make_pair(p->first, end));
+      break;
+    } else {
+      ++p;
+    }
+  }
+  if (p == compact_queue.end()) {
+    // no merge, new entry.
+    compact_queue.push_back(make_pair(start, end));
+  }
+  compact_queue_cond.Signal();
+  if (!compact_thread.is_started()) {
+    compact_thread.create();
+  }
+}

diff --git a/src/os/LevelDBStore.h b/src/os/LevelDBStore.h
index e4e84107fdf0381110a10cb932d4760befffd3b2..556ba0388ee8652530ac9119e46398cd01b73235 100644 (file)
--- a/src/os/LevelDBStore.h
+++ b/src/os/LevelDBStore.h
@@ -56,15 +56,7 @@ class LevelDBStore : public KeyValueDB {
     leveldb::Slice cend(end);
     db->CompactRange(&cstart, &cend);
   }
-  void compact_range_async(const string& start, const string& end) {
-    Mutex::Locker l(compact_queue_lock);
-    compact_queue.remove(make_pair(start, end));    // prevent unbounded dups
-    compact_queue.push_back(make_pair(start, end));
-    compact_queue_cond.Signal();
-    if (!compact_thread.is_started()) {
-      compact_thread.create();
-    }
-  }
+  void compact_range_async(const string& start, const string& end);
 
 public:
   /// compact the underlying leveldb store