feat:change thread sheduling method in ThreadPool class (ospp 2024)

src/net/include/likely.h

@@ -0,0 +1,17 @@
+// Copyright (c) 2024-present, Qihoo, Inc.  All rights reserved.
+// This source code is licensed under the BSD-style license found in the
+// LICENSE file in the root directory of this source tree. An additional grant
+// of patent rights can be found in the PATENTS file in the same directory.
+
+#ifndef LIKELY_H
+#define LIKELY_H
+
+#if defined(__GNUC__) && __GNUC__ >= 4
+#  define LIKELY(x) (__builtin_expect((x), 1))
+#  define UNLIKELY(x) (__builtin_expect((x), 0))
+#else
+#  define LIKELY(x) (x)
+#  define UNLIKELY(x) (x)
+#endif
+
+#endif  // LIKELY_H

src/net/include/net_define.h

@@ -6,9 +6,7 @@
 #ifndef NET_INCLUDE_NET_DEFINE_H_
 #define NET_INCLUDE_NET_DEFINE_H_
 
-#include <functional>
 #include <iostream>
-#include <map>
 
 namespace net {
 

src/net/include/random.h

@@ -0,0 +1,88 @@
+// Copyright (c) 2024-present, Qihoo, Inc.  All rights reserved.
+// This source code is licensed under the BSD-style license found in the
+// LICENSE file in the root directory of this source tree. An additional grant
+// of patent rights can be found in the PATENTS file in the same directory.
+
+#ifndef RANDOM_H
+#define RANDOM_H
+
+#include <random>
+#include <thread>
+
+#include "net/include/likely.h"
+
+namespace net {
+
+class Random {
+ private:
+  enum : uint32_t {
+    M = 2147483647L  // 2^31-1
+  };
+  enum : uint64_t {
+    A = 16807  // bits 14, 8, 7, 5, 2, 1, 0
+  };
+
+  uint32_t seed_;
+
+  static uint32_t GoodSeed(uint32_t s) { return (s & M) != 0 ? (s & M) : 1; }
+
+ public:
+  // This is the largest value that can be returned from Next()
+  enum : uint32_t { kMaxNext = M };
+
+  explicit Random(uint32_t s) : seed_(GoodSeed(s)) {}
+
+  void Reset(uint32_t s) { seed_ = GoodSeed(s); }
+
+  uint32_t Next() {
+    // We are computing
+    //       seed_ = (seed_ * A) % M,    where M = 2^31-1
+    //
+    // seed_ must not be zero or M, or else all subsequent computed values
+    // will be zero or M respectively.  For all other values, seed_ will end
+    // up cycling through every number in [1,M-1]
+    uint64_t product = seed_ * A;
+
+    // Compute (product % M) using the fact that ((x << 31) % M) == x.
+    seed_ = static_cast<uint32_t>((product >> 31) + (product & M));
+    // The first reduction may overflow by 1 bit, so we may need to
+    // repeat.  mod == M is not possible; using > allows the faster
+    // sign-bit-based test.
+    if (seed_ > M) {
+      seed_ -= M;
+    }
+    return seed_;
+  }
+
+  // Returns a uniformly distributed value in the range [0..n-1]
+  // REQUIRES: n > 0
+  uint32_t Uniform(int n) { return Next() % n; }
+
+  // Randomly returns true ~"1/n" of the time, and false otherwise.
+  // REQUIRES: n > 0
+  bool OneIn(int n) { return (Next() % n) == 0; }
+
+  // Skewed: pick "base" uniformly from range [0,max_log] and then
+  // return "base" random bits.  The effect is to pick a number in the
+  // range [0,2^max_log-1] with exponential bias towards smaller numbers.
+  uint32_t Skewed(int max_log) { return Uniform(1 << Uniform(max_log + 1)); }
+
+  // Returns a Random instance for use by the current thread without
+  // additional locking
+  static Random* GetTLSInstance() {
+    static __thread Random* tls_instance;
+    static __thread std::aligned_storage<sizeof(Random)>::type tls_instance_bytes;
+
+    auto rv = tls_instance;
+    if (UNLIKELY(rv == nullptr)) {
+      size_t seed = std::hash<std::thread::id>()(std::this_thread::get_id());
+      rv = new (&tls_instance_bytes) Random((uint32_t)seed);
+      tls_instance = rv;
+    }
+    return rv;
+  }
+};
+
+}  // namespace net
+
+#endif  // RANDOM_H

src/net/include/thread_pool.h

@@ -8,22 +8,16 @@
 
 #include <pthread.h>
 #include <atomic>
-#include <queue>
 #include <string>
+#include <vector>
 
 #include "net/include/net_define.h"
+#include "net/include/random.h"
 #include "pstd/include/pstd_mutex.h"
 
 namespace net {
 
-using TaskFunc = void (*)(void *);
-
-struct Task {
-  Task() = default;
-  TaskFunc func = nullptr;
-  void* arg = nullptr;
-  Task(TaskFunc _func, void* _arg) : func(_func), arg(_arg) {}
-};
+using TaskFunc = void (*)(void*);
 
 struct TimeTask {
   uint64_t exec_time;
@@ -37,7 +31,13 @@ class ThreadPool : public pstd::noncopyable {
  public:
   class Worker {
    public:
-    explicit Worker(ThreadPool* tp) : start_(false), thread_pool_(tp){};
+    struct Arg {
+      Arg(void* p, int i) : arg(p), idx(i) {}
+      void* arg;
+      int idx;
+    };
+
+    explicit Worker(ThreadPool* tp, int idx = 0) : start_(false), thread_pool_(tp), idx_(idx), arg_(tp, idx){};
     static void* WorkerMain(void* arg);
 
     int start();
@@ -48,9 +48,11 @@ class ThreadPool : public pstd::noncopyable {
     std::atomic<bool> start_;
     ThreadPool* const thread_pool_;
     std::string worker_name_;
+    int idx_;
+    Arg arg_;
   };
 
-  explicit ThreadPool(size_t worker_num, size_t max_queue_size, std::string  thread_pool_name = "ThreadPool");
+  explicit ThreadPool(size_t worker_num, size_t max_queue_size, std::string thread_pool_name = "ThreadPool");
   virtual ~ThreadPool();
 
   int start_thread_pool();
@@ -67,21 +69,73 @@ class ThreadPool : public pstd::noncopyable {
   std::string thread_pool_name();
 
  private:
-  void runInThread();
+  void runInThread(const int idx = 0);
+
+ public:
+  struct AdaptationContext {
+    std::atomic<int32_t> value;
+
+    explicit AdaptationContext() : value(0) {}
+  };
+
+ private:
+  struct Node {
+    Node* link_older = nullptr;
+    Node* link_newer = nullptr;
 
-  size_t worker_num_;
+    // true if task is TimeTask
+    bool is_time_task;
+    TimeTask task;
+
+    Node(TaskFunc func, void* arg) : is_time_task(false), task(0, func, arg) {}
+    Node(uint64_t exec_time, TaskFunc func, void* arg) : is_time_task(true), task(exec_time, func, arg) {}
+
+    inline void Exec() { task.func(task.arg); }
+    inline Node* Next() { return link_newer; }
+  };
+
+  // re-push some timer tasks which has been poped
+  void ReDelaySchedule(Node* nodes);
+
+  static inline void AsmVolatilePause() {
+#if defined(__i386__) || defined(__x86_64__)
+    asm volatile("pause");
+#elif defined(__aarch64__)
+    asm volatile("wfe");
+#elif defined(__powerpc64__)
+    asm volatile("or 27,27,27");
+#endif
+    // it's okay for other platforms to be no-ops
+  }
+
+  Node* CreateMissingNewerLinks(Node* head, int* cnt);
+  bool LinkOne(Node* node, std::atomic<Node*>* newest_node);
+
+  uint16_t task_idx_;
+
+  const uint8_t nworkers_per_link_ = 1;  // numer of workers per link
+  const uint8_t nlinks_;                 // number of links (upper around)
+  std::vector<std::atomic<Node*>> newest_node_;
+  std::atomic<int> node_cnt_;  // for task
+  std::vector<std::atomic<Node*>> time_newest_node_;
+  std::atomic<int> time_node_cnt_;  // for time task
+
+  const int queue_slow_size_;  // default value: min(worker_num_ * 10, max_queue_size_)
   size_t max_queue_size_;
+
+  const uint64_t max_yield_usec_;
+  const uint64_t slow_yield_usec_;
+
+  AdaptationContext adp_ctx;
+
+  const size_t worker_num_;
   std::string thread_pool_name_;
-  std::queue<Task> queue_;
-  std::priority_queue<TimeTask> time_queue_;
   std::vector<Worker*> workers_;
   std::atomic<bool> running_;
   std::atomic<bool> should_stop_;
 
-  pstd::Mutex mu_;
-  pstd::CondVar rsignal_;
-  pstd::CondVar wsignal_;
-
+  std::vector<pstd::Mutex> mu_;
+  std::vector<pstd::CondVar> rsignal_;
 };
 
 }  // namespace net