crimson/os/seastore/transaction_manager: remap_pin into coroutines

Signed-off-by: Matan Breizman <mbreizma@redhat.com>

diff --git a/src/crimson/os/seastore/transaction_manager.h b/src/crimson/os/seastore/transaction_manager.h
index 487bebe8939e27a7f9d34d2d99eb8493ee5d5f7c..74b0cae5260a7544c30a596050aeca9652d4b6d9 100644 (file)
--- a/src/crimson/os/seastore/transaction_manager.h
+++ b/src/crimson/os/seastore/transaction_manager.h
@@ -1240,7 +1240,7 @@ private:
   template <typename T, std::size_t N>
   remap_pin_ret remap_pin(
     Transaction &t,
-    LBAMapping &&pin,
+    LBAMapping pin,
     std::array<remap_entry_t, N> remaps) {
     static_assert(std::is_base_of_v<LogicalChildNode, T>);
     // data extents don't need maybe_init yet, currently,
@@ -1276,122 +1276,100 @@ private:
     }
 #endif
 
-    return seastar::do_with(
-      std::move(pin),
-      std::move(remaps),
-      [FNAME, &t, this](auto &pin, auto &remaps) {
-      // The according extent might be stable or pending.
-      auto fut = base_iertr::now();
-      if (pin.is_indirect()) {
-       SUBDEBUGT(seastore_tm, "{} into {} remaps ...",
-         t, pin, remaps.size());
-       fut = pin.refresh().si_then([this, &pin, &t](auto mapping) {
-         return lba_manager->complete_indirect_lba_mapping(
-           t, std::move(mapping)
-         ).si_then([&pin](auto mapping) {
-           pin = std::move(mapping);
-         });
-       });
+    if (pin.is_indirect()) {
+      SUBDEBUGT(seastore_tm, "{} into {} remaps ...",
+        t, pin, remaps.size());
+      pin = co_await pin.refresh();
+      pin = co_await lba_manager->complete_indirect_lba_mapping(t, pin);
+    } else {
+      laddr_t original_laddr = pin.get_key();
+      extent_len_t original_len = pin.get_length();
+      paddr_t original_paddr = pin.get_val();
+      SUBDEBUGT(seastore_tm, "{}~0x{:x} {} into {} remaps ... {}",
+        t, original_laddr, original_len, original_paddr, remaps.size(), pin);
+      ceph_assert(!pin.is_clone());
+
+      TCachedExtentRef<T> extent;
+      pin = co_await pin.refresh();
+      if (full_extent_integrity_check) {
+        // read the entire extent from disk (See: pin_to_extent)
+        auto maybe_indirect_extent = co_await read_pin<T>(t, pin);
+        assert(!maybe_indirect_extent.is_indirect());
+        assert(!maybe_indirect_extent.is_clone);
+        extent = maybe_indirect_extent.extent;
       } else {
-       laddr_t original_laddr = pin.get_key();
-       extent_len_t original_len = pin.get_length();
-       paddr_t original_paddr = pin.get_val();
-       SUBDEBUGT(seastore_tm, "{}~0x{:x} {} into {} remaps ... {}",
-         t, original_laddr, original_len, original_paddr, remaps.size(), pin);
-        ceph_assert(!pin.is_clone());
-       fut = pin.refresh().si_then([this, &t, &pin, original_paddr,
-                                   original_len](auto newpin) {
-         pin = std::move(newpin);
-         if (full_extent_integrity_check) {
-           return read_pin<T>(t, pin
-            ).si_then([](auto maybe_indirect_extent) {
-              assert(!maybe_indirect_extent.is_indirect());
-              assert(!maybe_indirect_extent.is_clone);
-              return maybe_indirect_extent.extent;
-            });
-         } else {
-           auto ret = get_extent_if_linked<T>(t, pin);
-           if (ret.index() == 1) {
-             return std::get<1>(ret
-             ).si_then([](auto extent) {
-               if (!extent->is_seen_by_users()) {
-                 // Note, no maybe_init available for data extents
-                 extent->set_seen_by_users();
-               }
-               return std::move(extent);
-             });
-           } else {
-             // absent
-             auto unlinked_child = std::move(std::get<0>(ret));
-             auto retired_placeholder = cache->retire_absent_extent_addr(
-               t, pin.get_key(), original_paddr, original_len
-             )->template cast<RetiredExtentPlaceholder>();
-             unlinked_child.child_pos.link_child(retired_placeholder.get());
-             return base_iertr::make_ready_future<TCachedExtentRef<T>>();
-           }
-         }
-       }).si_then([this, &t, &remaps, original_paddr,
-                   original_laddr, original_len, FNAME](auto ext) mutable {
-         if (full_extent_integrity_check) {
-           ceph_assert(ext && ext->is_fully_loaded());
-           // CRC_NULL shouldn't be possible when full extent
-           // integrity checks are enabled.
-           assert(ext->calc_crc32c() != CRC_NULL);
-         }
-         std::optional<ceph::bufferptr> original_bptr;
-         // TODO: preserve the bufferspace if partially loaded
-         if (ext && ext->is_fully_loaded()) {
-           ceph_assert(ext->is_data_stable());
-           ceph_assert(ext->get_length() >= original_len);
-           ceph_assert(ext->get_paddr() == original_paddr);
-           original_bptr = ext->get_bptr();
-         }
-         if (ext) {
-           assert(ext->is_seen_by_users());
-           cache->retire_extent(t, ext);
-         }
-         for (auto &remap : remaps) {
-           auto remap_offset = remap.offset;
-           auto remap_len = remap.len;
-           auto remap_laddr = (original_laddr + remap_offset).checked_to_laddr();
-           auto remap_paddr = original_paddr.add_offset(remap_offset);
-           SUBDEBUGT(seastore_tm, "remap direct pin into {}~0x{:x} {} ...",
-                     t, remap_laddr, remap_len, remap_paddr);
-           ceph_assert(remap_len < original_len);
-           ceph_assert(remap_offset + remap_len <= original_len);
-           ceph_assert(remap_len != 0);
-           ceph_assert(remap_offset % cache->get_block_size() == 0);
-           ceph_assert(remap_len % cache->get_block_size() == 0);
-           auto extent = cache->alloc_remapped_extent<T>(
-             t,
-             remap_laddr,
-             remap_paddr,
-             remap_len,
-             original_laddr,
-             original_bptr);
-           // user must initialize the logical extent themselves.
-           extent->set_seen_by_users();
-           remap.extent = extent.get();
-         }
-       });
+        auto ret = get_extent_if_linked<T>(t, pin);
+        if (std::holds_alternative<get_child_ifut<T>>(ret)) {
+          extent = co_await std::move(std::get<get_child_ifut<T>>(ret));
+          if (!extent->is_seen_by_users()) {
+            // Note, no maybe_init available for data extents
+            extent->set_seen_by_users();
+          }
+        } else if (std::holds_alternative<unlinked_child_t>(ret)) {
+          auto unlinked_child =  std::move(std::get<unlinked_child_t>(ret));
+          auto retired_placeholder = cache->retire_absent_extent_addr(
+            t, pin.get_key(), original_paddr, original_len
+          )->template cast<RetiredExtentPlaceholder>();
+          unlinked_child.child_pos.link_child(retired_placeholder.get());
+        } else {
+          ceph_abort("unexpected varaint in remap_pin");
+        }
       }
-      return fut.si_then([this, &t, &pin, &remaps, FNAME] {
-       return lba_manager->remap_mappings(
-         t,
-         std::move(pin),
-         std::vector<remap_entry_t>(remaps.begin(), remaps.end())
-       ).si_then([FNAME, &t](auto ret) {
-         SUBDEBUGT(seastore_tm, "remapped {} pins", t, ret.size());
-         return Cache::retire_extent_iertr::make_ready_future<
-           std::vector<LBAMapping>>(std::move(ret));
-       });
-      }).handle_error_interruptible(
-       remap_pin_iertr::pass_further{},
-       crimson::ct_error::assert_all{
-         "TransactionManager::remap_pin hit invalid error"
-       }
-      );
-    });
+
+      if (full_extent_integrity_check) {
+        ceph_assert(extent && extent->is_fully_loaded());
+        // CRC_NULL shouldn't be possible when full extent
+        // integrity checks are enabled.
+        assert(extent->calc_crc32c() != CRC_NULL);
+      }
+
+      std::optional<ceph::bufferptr> original_bptr;
+      // TODO: preserve the bufferspace if partially loaded
+      if (extent && extent->is_fully_loaded()) {
+        ceph_assert(extent->is_data_stable());
+        ceph_assert(extent->get_length() >= original_len);
+        ceph_assert(extent->get_paddr() == original_paddr);
+        original_bptr = extent->get_bptr();
+      }
+      if (extent) {
+        assert(extent->is_seen_by_users());
+        cache->retire_extent(t, extent);
+      }
+      for (auto &remap : remaps) {
+        auto remap_offset = remap.offset;
+        auto remap_len = remap.len;
+        auto remap_laddr = (original_laddr + remap_offset).checked_to_laddr();
+        auto remap_paddr = original_paddr.add_offset(remap_offset);
+        SUBDEBUGT(seastore_tm, "remap direct pin into {}~0x{:x} {} ...",
+                  t, remap_laddr, remap_len, remap_paddr);
+        ceph_assert(remap_len < original_len);
+        ceph_assert(remap_offset + remap_len <= original_len);
+        ceph_assert(remap_len != 0);
+        ceph_assert(remap_offset % cache->get_block_size() == 0);
+        ceph_assert(remap_len % cache->get_block_size() == 0);
+        auto remapped_extent = cache->alloc_remapped_extent<T>(
+          t,
+          remap_laddr,
+          remap_paddr,
+          remap_len,
+          original_laddr,
+          original_bptr);
+        // user must initialize the logical extent themselves.
+        remapped_extent->set_seen_by_users();
+        remap.extent = remapped_extent.get();
+      }
+     }
+
+     auto mapping_vec = co_await lba_manager->remap_mappings(
+       t,
+       pin,
+       std::vector<remap_entry_t>(remaps.begin(), remaps.end())
+       ).handle_error_interruptible(remap_pin_iertr::pass_further{},
+         crimson::ct_error::assert_all{
+         "TransactionManager::remap_pin hit invalid error"}
+       );
+       SUBDEBUGT(seastore_tm, "remapped {} pins", t, mapping_vec.size());
+       co_return std::move(mapping_vec);
   }
 
   using _remove_mapping_result_t = LBAManager::ref_update_result_t;