From 7628a4dfe0736bfc2e616afc2d3704330e686286 Mon Sep 17 00:00:00 2001
From: Calle Wilund <calle@scylladb.com>
Date: Mon, 25 Jan 2016 12:57:59 +0000
Subject: [PATCH] commitlog: Add some feedback/measurement methods

Suitable to derive "back pressure" from.
---
 db/commitlog/commitlog.cc | 52 ++++++++++++++++++++++++++++++++++++++-
 db/commitlog/commitlog.hh | 33 +++++++++++++++++++++++++
 2 files changed, 84 insertions(+), 1 deletion(-)

diff --git a/db/commitlog/commitlog.cc b/db/commitlog/commitlog.cc
index 603027f0d6..5b3de64987 100644
--- a/db/commitlog/commitlog.cc
+++ b/db/commitlog/commitlog.cc
@@ -214,6 +214,18 @@ public:
         if (cfg.commit_log_location.empty()) {
             cfg.commit_log_location = "/var/lib/scylla/commitlog";
         }
+
+        if (cfg.max_active_writes == 0) {
+            cfg.max_active_writes = // TODO: call someone to get an idea...
+                            25 * smp::count;
+        }
+        cfg.max_active_writes = std::max(uint64_t(1), cfg.max_active_writes / smp::count);
+        if (cfg.max_active_flushes == 0) {
+            cfg.max_active_flushes = // TODO: call someone to get an idea...
+                            5 * smp::count;
+        }
+        cfg.max_active_flushes = std::max(uint64_t(1), cfg.max_active_flushes / smp::count);
+
         logger.trace("Commitlog {} maximum disk size: {} MB / cpu ({} cpus)",
                 cfg.commit_log_location, max_disk_size / (1024 * 1024),
                 smp::count);
@@ -250,6 +262,8 @@ public:
     }
 
     std::vector<sstring> get_active_names() const;
+    uint64_t get_num_dirty_segments() const;
+    uint64_t get_num_active_segments() const;
 
     using buffer_type = temporary_buffer<char>;
 
@@ -1117,6 +1131,19 @@ std::vector<sstring> db::commitlog::segment_manager::get_active_names() const {
     return res;
 }
 
+uint64_t db::commitlog::segment_manager::get_num_dirty_segments() const {
+    return std::count_if(_segments.begin(), _segments.end(), [](sseg_ptr s) {
+        return !s->is_still_allocating() && !s->is_clean();
+    });
+}
+
+uint64_t db::commitlog::segment_manager::get_num_active_segments() const {
+    return std::count_if(_segments.begin(), _segments.end(), [](sseg_ptr s) {
+        return s->is_still_allocating();
+    });
+}
+
+
 db::commitlog::segment_manager::buffer_type db::commitlog::segment_manager::acquire_buffer(size_t s) {
     auto i = _temp_buffers.begin();
     auto e = _temp_buffers.end();
@@ -1267,11 +1294,18 @@ future<> db::commitlog::shutdown() {
     return _segment_manager->shutdown();
 }
 
-
 size_t db::commitlog::max_record_size() const {
     return _segment_manager->max_mutation_size - segment::entry_overhead_size;
 }
 
+uint64_t db::commitlog::max_active_writes() const {
+    return _segment_manager->cfg.max_active_writes;
+}
+
+uint64_t db::commitlog::max_active_flushes() const {
+    return _segment_manager->cfg.max_active_flushes;
+}
+
 future<> db::commitlog::clear() {
     return _segment_manager->clear();
 }
@@ -1526,6 +1560,14 @@ uint64_t db::commitlog::get_pending_tasks() const {
                     + _segment_manager->totals.pending_flushes;
 }
 
+uint64_t db::commitlog::get_pending_writes() const {
+    return _segment_manager->totals.pending_writes;
+}
+
+uint64_t db::commitlog::get_pending_flushes() const {
+    return _segment_manager->totals.pending_flushes;
+}
+
 uint64_t db::commitlog::get_num_segments_created() const {
     return _segment_manager->totals.segments_created;
 }
@@ -1534,6 +1576,14 @@ uint64_t db::commitlog::get_num_segments_destroyed() const {
     return _segment_manager->totals.segments_destroyed;
 }
 
+uint64_t db::commitlog::get_num_dirty_segments() const {
+    return _segment_manager->get_num_dirty_segments();
+}
+
+uint64_t db::commitlog::get_num_active_segments() const {
+    return _segment_manager->get_num_active_segments();
+}
+
 future<std::vector<db::commitlog::descriptor>> db::commitlog::list_existing_descriptors() const {
     return list_existing_descriptors(active_config().commit_log_location);
 }
diff --git a/db/commitlog/commitlog.hh b/db/commitlog/commitlog.hh
index 0ed98c6b7d..6e090546ec 100644
--- a/db/commitlog/commitlog.hh
+++ b/db/commitlog/commitlog.hh
@@ -115,6 +115,10 @@ public:
         // Max number of segments to keep in pre-alloc reserve.
         // Not (yet) configurable from scylla.conf.
         uint64_t max_reserve_segments = 12;
+        // Max active writes/flushes. Default value
+        // zero means try to figure it out ourselves
+        uint64_t max_active_writes = 0;
+        uint64_t max_active_flushes = 0;
 
         sync_mode mode = sync_mode::PERIODIC;
     };
@@ -241,14 +245,43 @@ public:
     uint64_t get_completed_tasks() const;
     uint64_t get_flush_count() const;
     uint64_t get_pending_tasks() const;
+    uint64_t get_pending_writes() const;
+    uint64_t get_pending_flushes() const;
     uint64_t get_num_segments_created() const;
     uint64_t get_num_segments_destroyed() const;
+    /**
+     * Get number of inactive (finished), segments lingering
+     * due to still being dirty
+     */
+    uint64_t get_num_dirty_segments() const;
+    /**
+     * Get number of active segments, i.e. still being allocated to
+     */
+    uint64_t get_num_active_segments() const;
 
     /**
      * Returns the largest amount of data that can be written in a single "mutation".
      */
     size_t max_record_size() const;
 
+    /**
+     * Return max allowed pending writes (per this shard)
+     */
+    uint64_t max_active_writes() const;
+    /**
+     * Set max allowed pending writes (per this shard)
+     */
+    void max_active_writes(uint64_t);
+
+    /**
+     * Return max allowed pending flushes (per this shard)
+     */
+    uint64_t max_active_flushes() const;
+    /**
+     * Set max allowed pending flushes (per this shard)
+     */
+    void max_active_flushes(uint64_t);
+
     future<> clear();
 
     const config& active_config() const;