crimson/os/alienstore: parse crimson_alien_thread_cpu_cores as a cpuset(7)

the "List format" listed in cpuset(7) allows us to specify a range of
CPU cores in a comma-separated list. and the upper bound of the range
is optional. before this change, the upper bound is not optional.

before this change, the upper bound of the range is not optional. and
the upper bound is not inclusive. so we don't support the list format
of cpuset(7). Take cores "1,2,3,5,7,8" for example, we need to set the
option to "1-4,5-6,7-9" to represent this cpuset.
after this change, the upper bound is optional, so we can properly
support the list format defined by cpuset(7). and the upper bound is
inclusive. so we can use "1-3,5,7-8", which is compatible with
notation defined by cpuset(7).

in this change, the cpuset option is parsed using a seastar helper,
which is implemented using a regex. so we don't need to manually parse
it. as Seastar's parser returns an `std::optional<std::set<unsigned>>`.
if the string does not match with the regex of comma-separated list,
the returned cpuset does not have a value. this design is more explicit.
so in this change, instead of using `std::vector<uint64_t>`,
`std::optional<cpuset>` is used.

Signed-off-by: Jianxin Li <jianxin1.li@intel.com>
Signed-off-by: Kefu Chai <tchaikov@gmail.com>

diff --git a/src/common/options/crimson.yaml.in b/src/common/options/crimson.yaml.in
index 09a652788ecd5f3215b038373e4d88a48ebd09e0..9a83b1fd0eca8cc91f9b569c0c0f72b23c929109 100644 (file)
--- a/src/common/options/crimson.yaml.in
+++ b/src/common/options/crimson.yaml.in
@@ -22,7 +22,7 @@ options:
 - name: crimson_alien_thread_cpu_cores
   type: str
   level: advanced
-  desc: CPU cores on which alienstore threads will run
+  desc: CPU cores on which alienstore threads will run in cpuset(7) format
 - name: seastore_segment_size
   type: size
   desc: Segment size to use for SegmentManager

diff --git a/src/crimson/os/alienstore/alien_store.cc b/src/crimson/os/alienstore/alien_store.cc
index dd5b54084116a86b795c8b86d96b0f9dbdebca4d..68d268408cc33e9680f09f8eaf9b326785e1ff20 100644 (file)
--- a/src/crimson/os/alienstore/alien_store.cc
+++ b/src/crimson/os/alienstore/alien_store.cc
@@ -14,6 +14,7 @@
 #include <seastar/core/alien.hh>
 #include <seastar/core/future-util.hh>
 #include <seastar/core/reactor.hh>
+#include <seastar/core/resource.hh>
 
 #include "common/ceph_context.h"
 #include "global/global_context.h"
@@ -96,17 +97,20 @@ seastar::future<> AlienStore::start()
   if (!store) {
     ceph_abort_msgf("unsupported objectstore type: %s", type.c_str());
   }
-  std::vector<uint64_t> cpu_cores = _parse_cpu_cores();
+  auto cpu_cores = seastar::resource::parse_cpuset(
+    get_conf<std::string>("crimson_alien_thread_cpu_cores"));
   // cores except the first "N_CORES_FOR_SEASTAR" ones will
   // be used for alien threads scheduling:
   //   [0, N_CORES_FOR_SEASTAR) are reserved for seastar reactors
   //   [N_CORES_FOR_SEASTAR, ..] are assigned to alien threads.
-  if (cpu_cores.empty()) {
+  if (!cpu_cores.has_value()) {
     if (long nr_cpus = sysconf(_SC_NPROCESSORS_ONLN);
        nr_cpus > N_CORES_FOR_SEASTAR ) {
-      for (int i = N_CORES_FOR_SEASTAR; i < nr_cpus; i++) {
-        cpu_cores.push_back(i);
+      seastar::resource::cpuset cpuset;
+      for (unsigned i = N_CORES_FOR_SEASTAR; i < nr_cpus; i++) {
+        cpuset.insert(i);
       }
+      cpu_cores = cpuset;
     } else {
       logger().error("{}: unable to get nproc: {}", __func__, errno);
     }
@@ -606,26 +610,4 @@ AlienStore::read_errorator::future<std::map<uint64_t, uint64_t>> AlienStore::fie
   });
 }
 
-std::vector<uint64_t> AlienStore::_parse_cpu_cores()
-{
-  std::vector<uint64_t> cpu_cores;
-  auto cpu_string =
-    get_conf<std::string>("crimson_alien_thread_cpu_cores");
-
-  std::string token;
-  std::istringstream token_stream(cpu_string);
-  while (std::getline(token_stream, token, ',')) {
-    std::istringstream cpu_stream(token);
-    std::string cpu;
-    std::getline(cpu_stream, cpu, '-');
-    uint64_t start_cpu = std::stoull(cpu);
-    std::getline(cpu_stream, cpu, '-');
-    uint64_t end_cpu = std::stoull(cpu);
-    for (uint64_t i = start_cpu; i < end_cpu; i++) {
-      cpu_cores.push_back(i);
-    }
-  }
-  return cpu_cores;
-}
-
 }

diff --git a/src/crimson/os/alienstore/alien_store.h b/src/crimson/os/alienstore/alien_store.h
index 1b6a4276479f1a2a9d9bd4d10e3a4e9656a2515d..9209dff828b2c4f31377e527604214f74343ec27 100644 (file)
--- a/src/crimson/os/alienstore/alien_store.h
+++ b/src/crimson/os/alienstore/alien_store.h
@@ -123,6 +123,5 @@ private:
   std::unique_ptr<CephContext> cct;
   mutable seastar::gate op_gate;
   std::unordered_map<coll_t, CollectionRef> coll_map;
-  std::vector<uint64_t> _parse_cpu_cores();
 };
 }

diff --git a/src/crimson/os/alienstore/thread_pool.cc b/src/crimson/os/alienstore/thread_pool.cc
index 747d6714e3662e4772701f83a7c304bab2f00467..5cf9590e61e0daec0f8c9e341d3ac4abe73b52bc 100644 (file)
--- a/src/crimson/os/alienstore/thread_pool.cc
+++ b/src/crimson/os/alienstore/thread_pool.cc
@@ -15,7 +15,7 @@ namespace crimson::os {
 
 ThreadPool::ThreadPool(size_t n_threads,
                        size_t queue_sz,
-                       std::vector<uint64_t> cpus)
+                       const std::optional<seastar::resource::cpuset>& cpus)
   : n_threads(n_threads),
     queue_size{round_up_to(queue_sz, seastar::smp::count)},
     pending_queues(n_threads)
@@ -23,8 +23,8 @@ ThreadPool::ThreadPool(size_t n_threads,
   auto queue_max_wait = std::chrono::seconds(local_conf()->threadpool_empty_queue_max_wait);
   for (size_t i = 0; i < n_threads; i++) {
     threads.emplace_back([this, cpus, queue_max_wait, i] {
-      if (!cpus.empty()) {
-        pin(cpus);
+      if (cpus.has_value()) {
+        pin(*cpus);
       }
       block_sighup();
       (void) pthread_setname_np(pthread_self(), "alien-store-tp");
@@ -40,7 +40,7 @@ ThreadPool::~ThreadPool()
   }
 }
 
-void ThreadPool::pin(const std::vector<uint64_t>& cpus)
+void ThreadPool::pin(const seastar::resource::cpuset& cpus)
 {
   cpu_set_t cs;
   CPU_ZERO(&cs);

diff --git a/src/crimson/os/alienstore/thread_pool.h b/src/crimson/os/alienstore/thread_pool.h
index 8f3069af3a5a595291866beb76aaf3397da992b6..78e18219a88a7143a7a84bb777abd796ca32f7db 100644 (file)
--- a/src/crimson/os/alienstore/thread_pool.h
+++ b/src/crimson/os/alienstore/thread_pool.h
@@ -11,6 +11,7 @@
 #include <seastar/core/future.hh>
 #include <seastar/core/gate.hh>
 #include <seastar/core/reactor.hh>
+#include <seastar/core/resource.hh>
 #include <seastar/core/semaphore.hh>
 #include <seastar/core/sharded.hh>
 
@@ -125,7 +126,7 @@ public:
    * @note each @c Task has its own crimson::thread::Condition, which possesses
    * an fd, so we should keep the size of queue under a reasonable limit.
    */
-  ThreadPool(size_t n_threads, size_t queue_sz, std::vector<uint64_t> cpus);
+  ThreadPool(size_t n_threads, size_t queue_sz, const std::optional<seastar::resource::cpuset>& cpus);
   ~ThreadPool();
   seastar::future<> start();
   seastar::future<> stop();
@@ -163,7 +164,7 @@ private:
   bool is_stopping() const {
     return stopping.load(std::memory_order_relaxed);
   }
-  static void pin(const std::vector<uint64_t>& cpus);
+  static void pin(const seastar::resource::cpuset& cpus);
   static void block_sighup();
   seastar::semaphore& local_free_slots() {
     return submit_queue.local().free_slots;

diff --git a/src/test/crimson/test_alienstore_thread_pool.cc b/src/test/crimson/test_alienstore_thread_pool.cc
index 7cfffec757b7421eac01709a8b8dbd1882c06baf..dbeed26cd7dbf798a8e46587c13653ae91ef44b9 100644 (file)
--- a/src/test/crimson/test_alienstore_thread_pool.cc
+++ b/src/test/crimson/test_alienstore_thread_pool.cc
@@ -50,7 +50,7 @@ int main(int argc, char** argv)
     .then([conf_file_list] {
       return local_conf().parse_config_files(conf_file_list);
     }).then([] {
-      return seastar::do_with(std::make_unique<crimson::os::ThreadPool>(2, 128, (std::vector<uint64_t>){0}),
+      return seastar::do_with(std::make_unique<crimson::os::ThreadPool>(2, 128, seastar::resource::cpuset{0}),
                               [](auto& tp) {
         return tp->start().then([&tp] {
           return test_accumulate(*tp);