ceph/src/crimson/thread/ThreadPool.cc

   1 #include "ThreadPool.h"
   2
   3 #include <pthread.h>
   4 #include "crimson/net/Config.h"
   5 #include "include/intarith.h"
   6
   7 #include "include/ceph_assert.h"
   8
   9 namespace crimson::thread {
  10
  11 ThreadPool::ThreadPool(size_t n_threads,
  12                        size_t queue_sz,
  13                        unsigned cpu_id)
  14   : queue_size{round_up_to(queue_sz, seastar::smp::count)},
  15     pending{queue_size}
  16 {
  17   for (size_t i = 0; i < n_threads; i++) {
  18     threads.emplace_back([this, cpu_id] {
  19       pin(cpu_id);
  20       loop();
  21     });
  22   }
  23 }
  24
  25 ThreadPool::~ThreadPool()
  26 {
  27   for (auto& thread : threads) {
  28     thread.join();
  29   }
  30 }
  31
  32 void ThreadPool::pin(unsigned cpu_id)
  33 {
  34   cpu_set_t cs;
  35   CPU_ZERO(&cs);
  36   CPU_SET(cpu_id, &cs);
  37   [[maybe_unused]] auto r = pthread_setaffinity_np(pthread_self(),
  38                                                    sizeof(cs), &cs);
  39   ceph_assert(r == 0);
  40 }
  41
  42 void ThreadPool::loop()
  43 {
  44   for (;;) {
  45     WorkItem* work_item = nullptr;
  46     {
  47       std::unique_lock lock{mutex};
  48       cond.wait_for(lock,
  49                     crimson::net::conf.threadpool_empty_queue_max_wait,
  50                     [this, &work_item] {
  51         return pending.pop(work_item) || is_stopping();
  52       });
  53     }
  54     if (work_item) {
  55       work_item->process();
  56     } else if (is_stopping()) {
  57       break;
  58     }
  59   }
  60 }
  61
  62 seastar::future<> ThreadPool::start()
  63 {
  64   auto slots_per_shard = queue_size / seastar::smp::count;
  65   return submit_queue.start(slots_per_shard);
  66 }
  67
  68 seastar::future<> ThreadPool::stop()
  69 {
  70   return submit_queue.stop().then([this] {
  71     stopping = true;
  72     cond.notify_all();
  73   });
  74 }
  75
  76 } // namespace crimson::thread