Commitlog: Add sync_all_segments()

For #293 - allows explicit flush to disk (not close!) of all active segments

db/commitlog/commitlog.cc

@@ -184,6 +184,7 @@ public:
     future<sseg_ptr> new_segment();
     future<sseg_ptr> active_segment();
     future<> clear();
+    future<> sync_all_segments();
 
     scollectd::registrations create_counters();
 
@@ -824,17 +825,22 @@ void db::commitlog::segment_manager::discard_unused_segments() {
     }
 }
 
+future<> db::commitlog::segment_manager::sync_all_segments() {
+    return parallel_for_each(_segments, [this](sseg_ptr s) {
+        return s->sync().then([](sseg_ptr) {});
+    });
+}
+
 /*
  * Sync all segments, then clear them out. To ensure all ops are done.
  * (Assumes you have barriered adding ops!)
+ * Only use from tests.
  */
 future<> db::commitlog::segment_manager::clear() {
     logger.debug("Clearing all segments");
     flush_segments(true);
-    return do_until([this]() {return _segments.empty();}, [this]() {
-        auto s = _segments.front();
-        _segments.erase(_segments.begin());
-        return s->sync().then([](sseg_ptr) {});
+    return sync_all_segments().then([this] {
+       _segments.clear();
     });
 }
 /**
@@ -964,6 +970,10 @@ void db::commitlog::discard_completed_segments(const cf_id_type& id,
     _segment_manager->discard_completed_segments(id, pos);
 }
 
+future<> db::commitlog::sync_all_segments() {
+    return _segment_manager->sync_all_segments();
+}
+
 size_t db::commitlog::max_record_size() const {
     return _segment_manager->max_mutation_size - segment::entry_overhead_size;
 }