crimson/os/seastore/segment_cleaner: track projected usage for in progress operations

We're going to want to permit multiple transactions to be writing
concurrently.  Replace await_hard_limits() with a mechanism that
remembers bytes that will be used by in-progress operations.

Signed-off-by: Samuel Just <sjust@redhat.com>

diff --git a/src/crimson/os/seastore/segment_cleaner.h b/src/crimson/os/seastore/segment_cleaner.h
index de668d270d61991cde9e7c8f52058a29135e36e6..3099fa2a84abb7533e956948bd8ddb512e03129e 100644 (file)
--- a/src/crimson/os/seastore/segment_cleaner.h
+++ b/src/crimson/os/seastore/segment_cleaner.h
@@ -391,6 +391,14 @@ private:
   uint64_t used_bytes = 0;
   bool init_complete = false;
 
+  /**
+   * projected_used_bytes
+   *
+   * Sum of projected bytes used by each transaction between throttle
+   * acquisition and commit completion.  See await_throttle()
+   */
+  uint64_t projected_used_bytes = 0;
+
   struct {
     uint64_t segments_released = 0;
   } stats;
@@ -762,6 +770,11 @@ private:
       get_bytes_available_current_segment() +
       get_bytes_scanned_current_segment();
   }
+  size_t get_projected_available_bytes() const {
+    return (get_available_bytes() > projected_used_bytes) ?
+      get_available_bytes() - projected_used_bytes:
+      0;
+  }
 
   /// Returns total space available
   size_t get_total_bytes() const {
@@ -772,11 +785,19 @@ private:
   size_t get_unavailable_bytes() const {
     return get_total_bytes() - get_available_bytes();
   }
+  size_t get_projected_unavailable_bytes() const {
+    return (get_total_bytes() > get_projected_available_bytes()) ?
+      (get_total_bytes() - get_projected_available_bytes()) :
+      0;
+  }
 
   /// Returns bytes currently occupied by live extents (not journal)
   size_t get_used_bytes() const {
     return used_bytes;
   }
+  size_t get_projected_used_bytes() const {
+    return used_bytes + projected_used_bytes;
+  }
 
   /// Return bytes contained in segments in journal
   size_t get_journal_segment_bytes() const {
@@ -798,6 +819,13 @@ private:
     else
       return 0;
   }
+  size_t get_projected_reclaimable_bytes() const {
+    auto ret = get_projected_unavailable_bytes() - get_projected_used_bytes();
+    if (ret > get_journal_segment_bytes())
+      return ret - get_journal_segment_bytes();
+    else
+      return 0;
+  }
 
   /**
    * get_reclaim_ratio
@@ -809,6 +837,11 @@ private:
     if (get_unavailable_bytes() == 0) return 0;
     return (double)get_reclaimable_bytes() / (double)get_unavailable_bytes();
   }
+  double get_projected_reclaim_ratio() const {
+    if (get_projected_unavailable_bytes() == 0) return 0;
+    return (double)get_reclaimable_bytes() /
+      (double)get_projected_unavailable_bytes();
+  }
 
   /**
    * get_available_ratio
@@ -818,6 +851,10 @@ private:
   double get_available_ratio() const {
     return (double)get_available_bytes() / (double)get_total_bytes();
   }
+  double get_projected_available_ratio() const {
+    return (double)get_projected_available_bytes() /
+      (double)get_total_bytes();
+  }
 
   /**
    * should_block_on_gc
@@ -825,11 +862,13 @@ private:
    * Encapsulates whether block pending gc.
    */
   bool should_block_on_gc() const {
-    auto aratio = get_available_ratio();
+    // TODO: probably worth projecting journal usage as well
+    auto aratio = get_projected_available_ratio();
     return (
       ((aratio < config.available_ratio_gc_max) &&
-       (get_reclaim_ratio() > config.reclaim_ratio_hard_limit ||
-       aratio < config.available_ratio_hard_limit)) ||
+       ((get_projected_reclaim_ratio() >
+        config.reclaim_ratio_hard_limit) ||
+       (aratio < config.available_ratio_hard_limit))) ||
       (get_dirty_tail_limit() > journal_tail_target)
     );
   }
@@ -875,7 +914,7 @@ private:
   }
 
 public:
-  seastar::future<> await_hard_limits() {
+  seastar::future<> reserve_projected_usage(size_t projected_usage) {
     // The pipeline configuration prevents another IO from entering
     // prepare until the prior one exits and clears this.
     ceph_assert(!blocked_io_wake);
@@ -887,7 +926,17 @@ public:
       [this] {
        blocked_io_wake = seastar::promise<>();
        return blocked_io_wake->get_future();
-      });
+      }
+    ).then([this, projected_usage] {
+      ceph_assert(!blocked_io_wake);
+      projected_used_bytes += projected_usage;
+    });
+  }
+
+  void release_projected_usage(size_t projected_usage) {
+    ceph_assert(projected_used_bytes >= projected_usage);
+    projected_used_bytes -= projected_usage;
+    return maybe_wake_gc_blocked_io();
   }
 private:
   void maybe_wake_gc_blocked_io() {

diff --git a/src/crimson/os/seastore/transaction.h b/src/crimson/os/seastore/transaction.h
index bed0f32e1c7f4520986ea4c7f31c6cfff13c33b8..7d83c0a98917668ddaea6f4a5fc09dacd237b239 100644 (file)
--- a/src/crimson/os/seastore/transaction.h
+++ b/src/crimson/os/seastore/transaction.h
@@ -189,6 +189,12 @@ public:
     return inline_block_list.size() + ool_block_list.size();
   }
 
+  size_t get_allocation_size() const {
+    size_t ret = 0;
+    for_each_fresh_block([&ret](auto &e) { ret += e->get_length(); });
+    return ret;
+  }
+
   enum class src_t : uint8_t {
     MUTATE = 0,
     READ, // including weak and non-weak read transactions

diff --git a/src/crimson/os/seastore/transaction_manager.cc b/src/crimson/os/seastore/transaction_manager.cc
index 393407d5dbfb24948d364be2ac191030db07796a..05df59ae0e1eadde39a3492caf28febbba00e24b 100644 (file)
--- a/src/crimson/os/seastore/transaction_manager.cc
+++ b/src/crimson/os/seastore/transaction_manager.cc
@@ -224,10 +224,15 @@ TransactionManager::submit_transaction(
   Transaction &t)
 {
   LOG_PREFIX(TransactionManager::submit_transaction);
-  DEBUGT("about to await throttle", t);
-  return trans_intr::make_interruptible(segment_cleaner->await_hard_limits()
-  ).then_interruptible([this, &t]() {
+  size_t projected_usage = t.get_allocation_size();
+  DEBUGT("waiting for projected_usage: {}", t, projected_usage);
+  return trans_intr::make_interruptible(
+    segment_cleaner->reserve_projected_usage(projected_usage)
+  ).then_interruptible([this, &t] {
     return submit_transaction_direct(t);
+  }).finally([this, FNAME, projected_usage, &t] {
+    DEBUGT("releasing projected_usage: {}", t, projected_usage);
+    segment_cleaner->release_projected_usage(projected_usage);
   });
 }