From 25dbf0875f9d33f09617c961c06dee9b00bd7ed2 Mon Sep 17 00:00:00 2001
From: Chunsong Feng <fengchunsong@huawei.com>
Date: Tue, 28 Jun 2022 01:54:41 +0000
Subject: [PATCH] msg/async: Supports the load balancing feature

Periodically collects load statistics. When the load imbalance exceeds
the threashold, selects a connection of the worker thread with the
highest load and migrates the connection to the worker thread with
lowest load.

Signed-off-by: Chunsong Feng <fengchunsong@huawei.com>
---
 src/common/options.cc            | 12 +++++
 src/msg/async/AsyncConnection.cc | 75 ++++++++++++++++++++++++--
 src/msg/async/AsyncConnection.h  | 13 ++++-
 src/msg/async/ProtocolV1.cc      | 14 +++--
 src/msg/async/ProtocolV2.cc      | 13 +++--
 src/msg/async/Stack.cc           | 92 ++++++++++++++++++++++++++++++++
 src/msg/async/Stack.h            | 44 +++++++++++++++
 7 files changed, 251 insertions(+), 12 deletions(-)

diff --git a/src/common/options.cc b/src/common/options.cc
index 2f28123ba5..f615beeec4 100644
--- a/src/common/options.cc
+++ b/src/common/options.cc
@@ -1153,6 +1153,18 @@ std::vector<Option> get_global_options() {
     .set_description("Maximum threadpool size of AsyncMessenger")
     .add_see_also("ms_async_op_threads"),
 
+    Option("ms_async_auto_balance", Option::TYPE_BOOL, Option::LEVEL_DEV)
+    .set_default(false)
+    .set_description("Automatically balance work load between msgr-worker threads")
+
+    Option("ms_async_balance_ms", Option::TYPE_UINT, Option::LEVEL_DEV)
+    .set_default(80)
+    .set_description("Automatically balance Interval (millseconds)"),	
+	
+    Option("ms_async_balance_ratio", Option::TYPE_FLOAT, Option::LEVEL_DEV)
+    .set_default(0.8)
+    .set_description("Automatically balance ratio"),	
+		
     Option("ms_async_reap_threshold", Option::TYPE_UINT, Option::LEVEL_DEV)
     .set_default(5)
     .set_min(1)
diff --git a/src/msg/async/AsyncConnection.cc b/src/msg/async/AsyncConnection.cc
index 5769c580e0..a65443f808 100644
--- a/src/msg/async/AsyncConnection.cc
+++ b/src/msg/async/AsyncConnection.cc
@@ -69,6 +69,7 @@ class C_handle_read : public EventCallback {
   explicit C_handle_read(AsyncConnectionRef c): conn(c) {}
   void do_request(uint64_t fd_or_id) override {
     conn->process();
+    conn->migrate();
   }
 };
 
@@ -78,7 +79,7 @@ class C_handle_write : public EventCallback {
  public:
   explicit C_handle_write(AsyncConnectionRef c): conn(c) {}
   void do_request(uint64_t fd) override {
-    conn->handle_write();
+    conn->handle_write(fd);
   }
 };
 
@@ -124,7 +125,7 @@ AsyncConnection::AsyncConnection(CephContext *cct, AsyncMessenger *m, DispatchQu
     connect_timeout_us(cct->_conf->ms_connection_ready_timeout*1000*1000),
     inactive_timeout_us(cct->_conf->ms_connection_idle_timeout*1000*1000),
     msgr2(m2), state_offset(0),
-    worker(w), center(&w->center),read_buffer(nullptr)
+    worker(w), center(&w->center), read_buffer(nullptr), load(0)
 {
 #ifdef UNIT_TESTS_BUILT
   this->interceptor = m->interceptor;
@@ -144,6 +145,7 @@ AsyncConnection::AsyncConnection(CephContext *cct, AsyncMessenger *m, DispatchQu
     protocol = std::unique_ptr<Protocol>(new ProtocolV1(this));
   }
   logger->inc(l_msgr_created_connections);
+  worker->register_conn(this);
 }
 
 AsyncConnection::~AsyncConnection()
@@ -600,12 +602,17 @@ void AsyncConnection::_stop() {
   writeCallback.reset();
   dispatch_queue->discard_queue(conn_id);
   async_msgr->unregister_conn(this);
+  worker->unregister_conn(this);
   worker->release_worker();
 
   state = STATE_CLOSED;
   open_write = false;
 
   state_offset = 0;
+  if (dest_worker) {
+    dest_worker = nullptr;
+    async_msgr->get_stack()->finish_migration();
+  }
   // Make sure in-queue events will been processed
   center->dispatch_event_external(EventCallbackRef(new C_clean_handler(this)));
 }
@@ -701,12 +708,16 @@ void AsyncConnection::mark_down()
   protocol->stop();
 }
 
-void AsyncConnection::handle_write()
+void AsyncConnection::handle_write(uint64_t fd)
 {
   ldout(async_msgr->cct, 10) << __func__ << dendl;
+  if (unlikely(!fd && (center->in_thread() == 0))) {
+    center->dispatch_event_external(write_handler);
+    return;
+  }
   protocol->write_event();
 }
-
+		
 void AsyncConnection::handle_write_callback() {
   std::lock_guard<std::mutex> l(lock);
   last_active = ceph::coarse_mono_clock::now();
@@ -743,6 +754,62 @@ void AsyncConnection::cleanup() {
   }
 }
 
+void AsyncConnection::migrate(void) {
+  if(!dest_worker)
+    return;
+  std::lock_guard<std::mutex> locker(lock);
+  if (!is_connected()) {
+    dest_worker = nullptr;
+    async_msgr->get_stack()->finish_migration();
+    return;
+  }
+  EventCenter *new_center = &dest_worker->center;
+  for (auto &&t : register_time_events)
+    center->delete_time_event(t);
+  register_time_events.clear();
+  if (last_tick_id) {
+    center->delete_time_event(last_tick_id);
+    last_tick_id = 0;
+  }
+  center->delete_file_event(cs.fd(), EVENT_READABLE | EVENT_WRITABLE);
+  load = 0;
+  worker->references--;
+  logger->dec(l_msgr_active_connections);
+  dest_worker->references++;
+  worker->unregister_conn(this);
+  ldout(async_msgr->cct, 5) << __func__ << " migrate " << worker->id << " to " << dest_worker->id << " thread " << pthread_self() << dendl;
+  dest_worker->register_conn(this);
+  logger = dest_worker->get_perf_counter();
+  
+  write_lock.lock();
+  worker = dest_worker;
+  center = new_center;
+  if (delay_state)
+    delay_state->set_center(new_center);
+  write_lock.unlock();
+  
+  logger->inc(l_msgr_active_connections);
+  auto transfer_handler = [this]() mutable {
+    std::lock_guard<std::mutex> locker(lock);
+    if (!is_connected()) {
+      async_msgr->get_stack()->finish_migration();
+      return;
+    }
+    write_lock.lock();
+    if (open_write) {
+      center->create_file_event(cs.fd(), EVENT_WRITABLE, write_handler);
+    }
+    center->create_file_event(cs.fd(), EVENT_READABLE, read_handler);
+    last_tick_id = center->create_time_event(inactive_timeout_us, tick_handler);
+    center->dispatch_event_external(read_handler);
+    center->dispatch_event_external(write_handler);
+    write_lock.unlock();
+    async_msgr->get_stack()->finish_migration();
+  };
+  center->submit_to(center->get_id(), std::move(transfer_handler), true);
+  dest_worker = nullptr;
+}
+
 void AsyncConnection::wakeup_from(uint64_t id)
 {
   lock.lock();
diff --git a/src/msg/async/AsyncConnection.h b/src/msg/async/AsyncConnection.h
index c7f0f9fe86..bbca82515c 100644
--- a/src/msg/async/AsyncConnection.h
+++ b/src/msg/async/AsyncConnection.h
@@ -225,10 +225,13 @@ private:
   std::function<void(char *, ssize_t)> readCallback;
   std::optional<unsigned> pendingReadLen;
   char *read_buffer;
+  std::atomic_ulong load;
+  Worker *dest_worker = nullptr;
 
  public:
   // used by eventcallback
-  void handle_write();
+  void handle_write(uint64_t fd);
+  void migrate();
   void handle_write_callback();
   void process();
   void wakeup_from(uint64_t id);
@@ -239,6 +242,14 @@ private:
     return logger;
   }
 
+  uint64_t get_and_clear_load() {
+    return load.exchange(0);
+  }
+
+  void set_dest_worker(Worker *w) {
+    dest_worker = w;
+  }
+
   bool is_msgr2() const override;
 
   friend class Protocol;
diff --git a/src/msg/async/ProtocolV1.cc b/src/msg/async/ProtocolV1.cc
index 43363371bc..81cf067aab 100644
--- a/src/msg/async/ProtocolV1.cc
+++ b/src/msg/async/ProtocolV1.cc
@@ -1007,16 +1007,20 @@ CtPtr ProtocolV1::handle_message_footer(char *buffer, int r) {
 
   ceph::mono_time fast_dispatch_time;
 
+  uint64_t recv_bytes = cur_msg_size + sizeof(ceph_msg_header) + sizeof(ceph_msg_footer);
   if (connection->is_blackhole()) {
     ldout(cct, 10) << __func__ << " blackhole " << *message << dendl;
     message->put();
     goto out;
   }
-
+  if (cct->_conf.get_val<bool>("ms_async_auto_balance") &&
+      (message->get_type() == CEPH_MSG_OSD_OP ||
+      message->get_type() == MSG_OSD_REPOP)) {
+    connection->load.fetch_add(recv_bytes);
+    connection->worker->request_balance();
+  }
   connection->logger->inc(l_msgr_recv_messages);
-  connection->logger->inc(
-      l_msgr_recv_bytes,
-      cur_msg_size + sizeof(ceph_msg_header) + sizeof(ceph_msg_footer));
+  connection->logger->inc(l_msgr_recv_bytes, recv_bytes);
 
   messenger->ms_fast_preprocess(message);
   fast_dispatch_time = ceph::mono_clock::now();
@@ -2404,8 +2408,10 @@ CtPtr ProtocolV1::replace(const AsyncConnectionRef& existing,
             if (exproto->state == NONE) {
               existing->shutdown_socket();
               existing->cs = std::move(cs);
+			  existing->worker->unregister_conn(get_pointer(existing));
               existing->worker->references--;
               new_worker->references++;
+			  new_worker->register_conn(get_pointer(existing));
               existing->logger = new_worker->get_perf_counter();
               existing->worker = new_worker;
               existing->center = new_center;
diff --git a/src/msg/async/ProtocolV2.cc b/src/msg/async/ProtocolV2.cc
index d2f879954c..d2561ea8ad 100644
--- a/src/msg/async/ProtocolV2.cc
+++ b/src/msg/async/ProtocolV2.cc
@@ -1448,16 +1448,21 @@ CtPtr ProtocolV2::handle_message() {
   state = READY;
 
   ceph::mono_time fast_dispatch_time;
-
+  uint64_t recv_bytes = rx_frame_asm.get_frame_onwire_len();
   if (connection->is_blackhole()) {
     ldout(cct, 10) << __func__ << " blackhole " << *message << dendl;
     message->put();
     goto out;
   }
 
+  if (cct->_conf.get_val<bool>("ms_async_auto_balance") &&
+      (message->get_type() == CEPH_MSG_OSD_OP ||
+      message->get_type() == MSG_OSD_REPOP)) {
+    connection->load.fetch_add(recv_bytes);
+    connection->worker->request_balance();
+  }
   connection->logger->inc(l_msgr_recv_messages);
-  connection->logger->inc(l_msgr_recv_bytes,
-                          rx_frame_asm.get_frame_onwire_len());
+  connection->logger->inc(l_msgr_recv_bytes, recv_bytes);
 
   messenger->ms_fast_preprocess(message);
   fast_dispatch_time = ceph::mono_clock::now();
@@ -2708,8 +2713,10 @@ CtPtr ProtocolV2::reuse_connection(const AsyncConnectionRef& existing,
           if (exproto->state == NONE) {
             existing->shutdown_socket();
             existing->cs = std::move(cs);
+			existing->worker->unregister_conn(get_pointer(existing));
             existing->worker->references--;
             new_worker->references++;
+			new_worker->register_conn(get_pointer(existing));
             existing->logger = new_worker->get_perf_counter();
             existing->worker = new_worker;
             existing->center = new_center;
diff --git a/src/msg/async/Stack.cc b/src/msg/async/Stack.cc
index 37e15634d7..4ed6e48005 100644
--- a/src/msg/async/Stack.cc
+++ b/src/msg/async/Stack.cc
@@ -20,6 +20,8 @@
 #include "common/Cond.h"
 #include "common/errno.h"
 #include "PosixStack.h"
+#include "AsyncConnection.h"
+
 #ifdef HAVE_RDMA
 #include "rdma/RDMAStack.h"
 #endif
@@ -85,6 +87,7 @@ std::shared_ptr<NetworkStack> NetworkStack::create(CephContext *c,
     ceph_abort();
     return nullptr;
   }
+  stack->balance_interval = clock_type::now();
   
   const int InitEventNumber = 5000;
   for (unsigned worker_id = 0; worker_id < stack->num_workers; ++worker_id) {
@@ -92,6 +95,7 @@ std::shared_ptr<NetworkStack> NetworkStack::create(CephContext *c,
     int ret = w->center.init(InitEventNumber, worker_id, t);
     if (ret)
       throw std::system_error(-ret, std::generic_category());
+    w->balance_handler = new C_balance_msg(stack, w);
     stack->workers.push_back(w);
   }
 
@@ -204,3 +208,91 @@ void NetworkStack::drain()
   drain.wait();
   ldout(cct, 30) << __func__ << " end." << dendl;
 }
+
+uint64_t Worker::calculate_loads(std::map<AsyncConnection *, uint64_t> &loads) {
+  std::unique_lock<std::mutex> locker(conns_lock);
+  uint64_t sum = 0;
+  for (auto conn: conns) {
+    uint64_t load = conn->get_and_clear_load();
+    if (load == 0)
+      continue;
+    loads[conn] = load;
+    sum += load;
+  }
+  return sum;
+}
+
+void NetworkStack::balance(void)
+{
+  static pthread_t tid = 0;
+  
+  if(__sync_bool_compare_and_swap(&tid, 0, pthread_self()) == 0)
+    return;
+  if (migrating || clock_type::now() < balance_interval) {
+    tid = 0;
+    return;
+  }
+  
+  std::map<AsyncConnection *, uint64_t> conn_loads_map;
+  Worker * w_max = nullptr;
+  Worker * w_min = nullptr;
+  uint64_t load_max = 0;
+  uint64_t load_min = std::numeric_limits<int>::max();
+  uint64_t avg_load = 0;
+  unsigned num_workers = get_num_worker();
+  pool_spin.lock();
+  for (unsigned i = 0; i < num_workers; i++) {
+    uint64_t sum;
+    std::map<AsyncConnection *, uint64_t> loads;
+    sum = workers[i]->calculate_loads(loads);
+    if (sum > load_max) {
+      load_max = sum;
+      w_max = workers[i];
+      conn_loads_map.swap(loads);
+    }
+    loads.clear();
+    if (sum < load_min) {
+      load_min = sum;
+      w_min = workers[i];
+    }
+    avg_load += sum / num_workers;
+  }
+  pool_spin.unlock();
+  
+  balance_interval = clock_type::now() + std::chrono::milliseconds(cct->_conf->ms_async_balance_ms);
+  uint64_t delta = load_max - load_min;
+  if (!load_max || (delta < cct->_conf->ms_async_balance_ratio * load_max)) {
+    conn_loads_map.clear();
+    tid = 0;
+    return;
+  }
+  #define pow2(x) ((x) * (x))
+  uint64_t org_stdval = pow2(load_max - avg_load) + pow2(load_min - avg_load);
+  uint64_t min_stdval = org_stdval;
+  AsyncConnection *conn_opt = nullptr;
+  uint64_t conn_load = 0;
+  for (auto it = conn_loads_map.begin(); it != conn_loads_map.end(); it++) {
+    uint64_t load = it->second;
+    if (load >= delta)
+      continue;
+    uint64_t stdval = pow2(load_max - avg_load - load) + pow2(load_min - avg_load + load);
+    if (stdval < min_stdval) {
+      conn_opt = it->first;
+      min_stdval = stdval;
+      conn_load = load;
+    }
+  }
+  conn_loads_map.clear();
+  if (org_stdval == min_stdval) {
+    tid = 0;
+    return;
+  }
+  if (w_max->conns.find(conn_opt) != w_max->conns.end()) {
+    conn_opt->set_dest_worker(w_min);
+    migrating = 1;
+  } else {
+    ldout(cct, 0) << __func__ << " conns " << conn_opt << " is deregisted, do nothing." << dendl;
+  }
+  ldout(cct, 1) << __func__ << " stdval " << org_stdval << " new stdval " << min_stdval << " conn_opt " << conn_opt << " conn_laod " << conn_load << " num conns " << w_max->conns.size() << dendl;
+  tid = 0;
+}
diff --git a/src/msg/async/Stack.h b/src/msg/async/Stack.h
index 6a2188b5a9..32a9fd8755 100644
--- a/src/msg/async/Stack.h
+++ b/src/msg/async/Stack.h
@@ -19,6 +19,7 @@
 
 #include "include/spinlock.h"
 #include "common/perf_counters.h"
+#include "common/ceph_time.h"
 #include "msg/msg_types.h"
 #include "msg/async/Event.h"
 
@@ -206,6 +207,7 @@ enum {
   l_msgr_last,
 };
 
+class AsyncConnection;
 class Worker {
   std::mutex init_lock;
   std::condition_variable init_cond;
@@ -220,6 +222,11 @@ class Worker {
 
   std::atomic_uint references;
   EventCenter center;
+  EventCallbackRef balance_handler;
+  bool requested = false;
+  
+  std::mutex conns_lock;
+  std::set<AsyncConnection *> conns;
 
   Worker(const Worker&) = delete;
   Worker& operator=(const Worker&) = delete;
@@ -254,6 +261,8 @@ class Worker {
       cct->get_perfcounters_collection()->remove(perf_logger);
       delete perf_logger;
     }
+    if (balance_handler)
+      delete balance_handler;
   }
 
   virtual int listen(entity_addr_t &addr, unsigned addr_slot,
@@ -290,12 +299,32 @@ class Worker {
     init_lock.unlock();
     done = false;
   }
+  void register_conn(AsyncConnection *conn) {
+    std::unique_lock<std::mutex> locker(conns_lock);
+    conns.insert(conn);
+  }
+  
+  void unregister_conn(AsyncConnection *conn) {
+    std::unique_lock<std::mutex> locker(conns_lock);
+    conns.erase(conn);
+  }
+  
+  uint64_t calculate_loads(std::map<AsyncConnection *, uint64_t> &loads);
+  void request_balance() {
+    if (requested)
+    	return;
+    center.dispatch_event_external(balance_handler);
+    requested = true;
+  }
 };
 
 class NetworkStack {
+  using clock_type = ceph::coarse_real_clock;
   unsigned num_workers = 0;
   ceph::spinlock pool_spin;
   bool started = false;
+  clock_type::time_point balance_interval;
+  bool migrating = 0;
 
   std::function<void ()> add_thread(unsigned i);
 
@@ -337,6 +366,11 @@ class NetworkStack {
     return num_workers;
   }
 
+  void balance(void);
+  void finish_migration() {
+    balance_interval = clock_type::now() + std::chrono::milliseconds(cct->_conf->ms_async_balance_ms);
+    migrating = 0;
+  }
   // direct is used in tests only
   virtual void spawn_worker(unsigned i, std::function<void ()> &&) = 0;
   virtual void join_worker(unsigned i) = 0;
@@ -345,4 +379,14 @@ class NetworkStack {
   virtual void ready() { };
 };
 
+class C_balance_msg : public EventCallback {
+  std::shared_ptr<NetworkStack> stack;
+  Worker *worker;
+  public:
+  explicit C_balance_msg(std::shared_ptr<NetworkStack> s, Worker *w): stack(s), worker(w) {}
+  void do_request(uint64_t fd) override {
+    worker->requested = false;
+    stack->balance();
+  }
+};
 #endif //CEPH_MSG_ASYNC_STACK_H
-- 
Gitee