Merge pull request ceph#60714 from aclamk/wip-aclamk-bluefs-unittest-async-compaction

aclamk · web-flow · commit f1a896099835 · 2025-03-11T19:52:46.000+01:00
os/bluestore: New unit test for BlueFS async compaction
diff --git a/src/os/bluestore/BlueFS.cc b/src/os/bluestore/BlueFS.cc
@@ -2950,7 +2950,9 @@ void BlueFS::_compact_log_async_LD_LNF_D() //also locks FW for new_writer
   // we need to flush all bdev because we will be streaming all dirty files to log
   // TODO - think - if _flush_and_sync_log_jump will not add dirty files nor release pending allocations
   // then flush_bdev() will not be necessary
+  tracepoint_async_compact(1);
   _flush_bdev();
+  tracepoint_async_compact(2);
   _flush_and_sync_log_jump_D(old_log_jump_to);
 
   //
@@ -3073,14 +3075,18 @@ void BlueFS::_compact_log_async_LD_LNF_D() //also locks FW for new_writer
 
   // 3.2. flush and wait
   _flush_special(new_log_writer);
+  tracepoint_async_compact(3);
   _flush_bdev(new_log_writer, false); // do not check log.lock is locked
 
   // Part 4.
   // Write out new superblock to reflect all the changes.
   //
 
+  tracepoint_async_compact(4);
   _write_super(BDEV_DB);
+  tracepoint_async_compact(5);
   _flush_bdev();
+  tracepoint_async_compact(6);
 
   // Part 5.
   // Apply new log fnode
diff --git a/src/os/bluestore/BlueFS.h b/src/os/bluestore/BlueFS.h
@@ -216,6 +216,35 @@ struct bluefs_shared_alloc_context_t {
   }
 };
 
+/*
+  Class debug_point is a helper intended for inserting debug tracepoints
+  in a least intrusive way.
+  The intention is to minimize both visual and code footprints.
+
+  In code, it should look like:
+  debug_async_compact(1);
+
+  Which should translate to:
+  if (debug_async_compact) debug_async_compact(1);
+
+  And in release builds be eliminated completely.
+*/
+template <class T>
+class debug_point_t {
+public:
+  debug_point_t() : m_func(nullptr) {};
+  debug_point_t(T&& func)
+  : m_func(func) {}
+  template<typename... Arg>
+  void operator()(Arg... arg) { if (m_func) m_func(std::forward<Arg...>(arg...)); }
+  void operator=(T&& func) { m_func = std::move(func);}
+  void operator=(T& func) { m_func = func;}
+private:
+  T m_func;
+};
+
+
+
 class BlueFS {
 public:
   CephContext* cct;
@@ -776,6 +805,7 @@ class BlueFS {
   }
   uint64_t debug_get_dirty_seq(FileWriter *h);
   bool debug_get_is_dev_dirty(FileWriter *h, uint8_t dev);
+  debug_point_t<std::function<void(uint32_t)>> tracepoint_async_compact;
   void trim_free_space(const std::string& type, std::ostream& outss);
 
 private:
diff --git a/src/test/objectstore/CMakeLists.txt b/src/test/objectstore/CMakeLists.txt
@@ -110,6 +110,13 @@ if(WITH_BLUESTORE)
   add_ceph_unittest(unittest_bluefs)
   target_link_libraries(unittest_bluefs os global)
 
+  # unittest_bluefs_ex
+  add_executable(unittest_bluefs_ex
+    test_bluefs_ex.cc
+    )
+  add_ceph_unittest(unittest_bluefs_ex)
+  target_link_libraries(unittest_bluefs_ex os global)
+
   # unittest_bluestore_types
   add_executable(unittest_bluestore_types
     test_bluestore_types.cc
diff --git a/src/test/objectstore/test_bluefs_ex.cc b/src/test/objectstore/test_bluefs_ex.cc
@@ -0,0 +1,222 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include <cstdlib>
+#include <initializer_list>
+#include <stdio.h>
+#include <string.h>
+#include <iostream>
+#include <time.h>
+#include <fcntl.h>
+#include <unistd.h>
+#include <random>
+#include <thread>
+#include <stack>
+#include <gtest/gtest.h>
+#include "global/global_init.h"
+#include "common/ceph_argparse.h"
+#include "include/stringify.h"
+#include "include/scope_guard.h"
+#include "common/errno.h"
+
+#include "os/bluestore/Allocator.h"
+#include "os/bluestore/bluestore_common.h"
+#include "os/bluestore/BlueFS.h"
+
+using namespace std;
+
+int argc;
+char **argv;
+
+std::unique_ptr<char[]> gen_buffer(uint64_t size)
+{
+    std::unique_ptr<char[]> buffer = std::make_unique<char[]>(size);
+    std::independent_bits_engine<std::default_random_engine, CHAR_BIT, unsigned char> e;
+    std::generate(buffer.get(), buffer.get()+size, std::ref(e));
+    return buffer;
+}
+
+class TempBdev {
+public:
+  TempBdev() {}
+  ~TempBdev() {}
+  void choose_name(pid_t pid = getpid()) {
+    static int n = 0;
+    path = "ceph_test_bluefs.tmp.block." + stringify(pid)
+    + "." + stringify(++n);
+  }
+  void create_bdev(uint64_t size) {
+    ceph_assert(!path.empty());
+    int fd = ::open(path.c_str(), O_CREAT|O_RDWR|O_TRUNC, 0644);
+    ceph_assert(fd >= 0);
+    int r = ::ftruncate(fd, size);
+    ceph_assert(r >= 0);
+    ::close(fd);
+  }
+  void rm_bdev() {
+    ceph_assert(!path.empty());
+    ::unlink(path.c_str());
+  }
+  std::string path;
+};
+
+class ConfSaver {
+  std::stack<std::pair<std::string, std::string>> saved_settings;
+  ConfigProxy& conf;
+public:
+  ConfSaver(ConfigProxy& conf) : conf(conf) {
+    conf._clear_safe_to_start_threads();
+  };
+  ~ConfSaver() {
+    conf._clear_safe_to_start_threads();
+    while(saved_settings.size() > 0) {
+      auto& e = saved_settings.top();
+      conf.set_val_or_die(e.first, e.second);
+      saved_settings.pop();
+    }
+    conf.set_safe_to_start_threads();
+    conf.apply_changes(nullptr);
+  }
+  void SetVal(const char* key, const char* val) {
+    std::string skey(key);
+    std::string prev_val;
+    conf.get_val(skey, &prev_val);
+    conf.set_val_or_die(skey, val);
+    saved_settings.emplace(skey, prev_val);
+  }
+  void ApplyChanges() {
+    conf.set_safe_to_start_threads();
+    conf.apply_changes(nullptr);
+  }
+};
+
+
+class BlueFS_ex : virtual public ::testing::Test {
+
+public:
+  explicit BlueFS_ex()
+  {
+  }
+  boost::intrusive_ptr<CephContext> init_ceph()
+  {
+    boost::intrusive_ptr<CephContext> cct;
+    auto args = argv_to_vec(argc, argv);
+    map<string, string> defaults = {
+      {"debug_bluefs", "0/20"},
+      {"debug_bdev", "0/20"},
+      {"log_to_stderr", "false"}};
+    cct = global_init(
+      &defaults, args, CEPH_ENTITY_TYPE_CLIENT,
+      CODE_ENVIRONMENT_UTILITY, CINIT_FLAG_NO_DEFAULT_CONFIG_FILE);
+    common_init_finish(g_ceph_context);
+    g_ceph_context->_conf.set_val(
+        "enable_experimental_unrecoverable_data_corrupting_features", "*");
+    g_ceph_context->_conf.apply_changes(nullptr);
+    return cct;
+  }
+
+  void SetUp() override
+  {
+  }
+  void TearDown() override
+  {
+  }
+
+  void grow_log_interrupt_on_compact(pid_t parent_pid, uint32_t stop_point)
+  {
+    auto cct = init_ceph();
+    ConfSaver conf(g_ceph_context->_conf);
+    conf.SetVal("bluefs_alloc_size", "4096");
+    conf.SetVal("bluefs_shared_alloc_size", "4096");
+    conf.SetVal("bluefs_compact_log_sync", "false");
+    conf.SetVal("bluefs_log_compact_min_size", "1048576");
+    conf.ApplyChanges();
+
+    auto stop_at_fixed_point = [&](uint32_t i) -> void {
+      if (i == stop_point) exit(107);
+    };
+    BlueFS fs(g_ceph_context);
+    fs.tracepoint_async_compact = stop_at_fixed_point;
+    ASSERT_EQ(0, fs.add_block_device(BlueFS::BDEV_DB, bdev.path, false));
+    uuid_d fsid;
+    ASSERT_EQ(0, fs.mkfs(fsid, {BlueFS::BDEV_DB, false, false}));
+    ASSERT_EQ(0, fs.mount());
+    ASSERT_EQ(0, fs.maybe_verify_layout({BlueFS::BDEV_DB, false, false}));
+    ASSERT_EQ(0, fs.mkdir("dir"));
+
+    auto fill = [&](uint32_t filenum) {
+      char data[2000] = {'x'};
+      BlueFS::FileWriter *h;
+      ASSERT_EQ(0, fs.open_for_write("dir", "file"+to_string(filenum), &h, false));
+      for (size_t i = 0; i < 10000; i++) {
+        h->append(data, 2000);
+        fs.fsync(h);
+      }
+      fs.close_writer(h);
+    };
+    std::thread thr[10];
+    for (int i=0; i< 10;i++) {
+      thr[i] = std::thread(fill, i);
+    }
+    for (int i=0; i< 10;i++) {
+      thr[i].join();
+    }
+    EXPECT_TRUE(false && "reaching this point means test was not executed");
+    exit(111);
+  }
+
+  TempBdev bdev;
+};
+
+TEST_F(BlueFS_ex, test_interrupted_compaction)
+{
+  for (uint32_t stop_point = 1; stop_point <= 6; stop_point++)
+  {
+    pid_t fork_for_test = fork();
+    if (fork_for_test != 0) {
+      int stat;
+      waitpid(fork_for_test, &stat, 0);
+      ASSERT_TRUE(WIFEXITED(stat));
+      ASSERT_TRUE(WEXITSTATUS(stat) == 0);
+      continue;
+    }
+    pid_t parent_pid = getpid();
+    uint64_t size = 1048576LL * (2 * 1024 + 128);
+    bdev.choose_name();
+    bdev.create_bdev(size);
+    pid_t fork_pid = fork();
+    if (fork_pid == 0) {
+      std::cout << "growing BlueFS log for async compact, stop at #" << (int)stop_point << std::endl;
+      grow_log_interrupt_on_compact(parent_pid, stop_point);
+    } else {
+      int stat;
+      std::cout << "waiting for compaction to terminate" << std::endl;
+      waitpid(fork_pid, &stat, 0);
+      std::cout << "done code=" << WEXITSTATUS(stat) << std::endl;
+      if(!WIFEXITED(stat) || WEXITSTATUS(stat) != 107) exit(107);
+      auto cct = init_ceph();
+      ConfSaver conf(g_ceph_context->_conf);
+      conf.SetVal("bluefs_alloc_size", "4096");
+      conf.SetVal("bluefs_shared_alloc_size", "4096");
+      conf.SetVal("bluefs_compact_log_sync", "false");
+      conf.SetVal("bluefs_log_compact_min_size", "1048576");
+      conf.ApplyChanges();
+
+      BlueFS fs(g_ceph_context);
+      ASSERT_EQ(0, fs.add_block_device(BlueFS::BDEV_DB, bdev.path, false));
+      // fs.log_dump();
+      ASSERT_EQ(0, fs.mount());
+      fs.umount();
+    }
+    bdev.rm_bdev();
+    exit(0); //this terminates one loop of 'fork_for_test'
+  }
+
+}
+
+int main(int _argc, char **_argv) {
+  argc = _argc;
+  argv = _argv;
+  ::testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}
