feat:change thread sheduling method in ThreadPool class

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_ = 2; // 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