--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2016 XSky <haomai@xsky.com>
+ *
+ * Author: Haomai Wang <haomaiwang@gmail.com>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation. See file COPYING.
+ *
+ */
+
+#include "include/compat.h"
+#include "common/Cond.h"
+#include "common/errno.h"
+#include "PosixStack.h"
+#ifdef HAVE_RDMA
+#include "rdma/RDMAStack.h"
+#endif
+#ifdef HAVE_DPDK
+#include "dpdk/DPDKStack.h"
+#endif
+
+#include "common/dout.h"
+#include "include/assert.h"
+
+#define dout_subsys ceph_subsys_ms
+#undef dout_prefix
+#define dout_prefix *_dout << "stack "
+
+std::function<void ()> NetworkStack::add_thread(unsigned i)
+{
+ Worker *w = workers[i];
+ return [this, w]() {
+ char tp_name[16];
+ sprintf(tp_name, "msgr-worker-%d", w->id);
+ ceph_pthread_setname(pthread_self(), tp_name);
+ const uint64_t EventMaxWaitUs = 30000000;
+ w->center.set_owner();
+ ldout(cct, 10) << __func__ << " starting" << dendl;
+ w->initialize();
+ w->init_done();
+ while (!w->done) {
+ ldout(cct, 30) << __func__ << " calling event process" << dendl;
+
+ ceph::timespan dur;
+ int r = w->center.process_events(EventMaxWaitUs, &dur);
+ if (r < 0) {
+ ldout(cct, 20) << __func__ << " process events failed: "
+ << cpp_strerror(errno) << dendl;
+ // TODO do something?
+ }
+ w->perf_logger->tinc(l_msgr_running_total_time, dur);
+ }
+ w->reset();
+ w->destroy();
+ };
+}
+
+std::shared_ptr<NetworkStack> NetworkStack::create(CephContext *c, const string &t)
+{
+ if (t == "posix")
+ return std::make_shared<PosixNetworkStack>(c, t);
+#ifdef HAVE_RDMA
+ else if (t == "rdma")
+ return std::make_shared<RDMAStack>(c, t);
+#endif
+#ifdef HAVE_DPDK
+ else if (t == "dpdk")
+ return std::make_shared<DPDKStack>(c, t);
+#endif
+
+ lderr(c) << __func__ << " ms_async_transport_type " << t <<
+ " is not supported! " << dendl;
+ ceph_abort();
+ return nullptr;
+}
+
+Worker* NetworkStack::create_worker(CephContext *c, const string &type, unsigned i)
+{
+ if (type == "posix")
+ return new PosixWorker(c, i);
+#ifdef HAVE_RDMA
+ else if (type == "rdma")
+ return new RDMAWorker(c, i);
+#endif
+#ifdef HAVE_DPDK
+ else if (type == "dpdk")
+ return new DPDKWorker(c, i);
+#endif
+
+ lderr(c) << __func__ << " ms_async_transport_type " << type <<
+ " is not supported! " << dendl;
+ ceph_abort();
+ return nullptr;
+}
+
+NetworkStack::NetworkStack(CephContext *c, const string &t): type(t), started(false), cct(c)
+{
+ assert(cct->_conf->ms_async_op_threads > 0);
+
+ const uint64_t InitEventNumber = 5000;
+ num_workers = cct->_conf->ms_async_op_threads;
+ if (num_workers >= EventCenter::MAX_EVENTCENTER) {
+ ldout(cct, 0) << __func__ << " max thread limit is "
+ << EventCenter::MAX_EVENTCENTER << ", switching to this now. "
+ << "Higher thread values are unnecessary and currently unsupported."
+ << dendl;
+ num_workers = EventCenter::MAX_EVENTCENTER;
+ }
+
+ for (unsigned i = 0; i < num_workers; ++i) {
+ Worker *w = create_worker(cct, type, i);
+ w->center.init(InitEventNumber, i, type);
+ workers.push_back(w);
+ }
+ cct->register_fork_watcher(this);
+}
+
+void NetworkStack::start()
+{
+ pool_spin.lock();
+ if (started) {
+ pool_spin.unlock();
+ return ;
+ }
+
+ for (unsigned i = 0; i < num_workers; ++i) {
+ if (workers[i]->is_init())
+ continue;
+ std::function<void ()> thread = add_thread(i);
+ spawn_worker(i, std::move(thread));
+ }
+ started = true;
+ pool_spin.unlock();
+
+ for (unsigned i = 0; i < num_workers; ++i)
+ workers[i]->wait_for_init();
+}
+
+Worker* NetworkStack::get_worker()
+{
+ ldout(cct, 30) << __func__ << dendl;
+
+ // start with some reasonably large number
+ unsigned min_load = std::numeric_limits<int>::max();
+ Worker* current_best = nullptr;
+
+ pool_spin.lock();
+ // find worker with least references
+ // tempting case is returning on references == 0, but in reality
+ // this will happen so rarely that there's no need for special case.
+ for (unsigned i = 0; i < num_workers; ++i) {
+ unsigned worker_load = workers[i]->references.load();
+ if (worker_load < min_load) {
+ current_best = workers[i];
+ min_load = worker_load;
+ }
+ }
+
+ pool_spin.unlock();
+ assert(current_best);
+ ++current_best->references;
+ return current_best;
+}
+
+void NetworkStack::stop()
+{
+ Spinlock::Locker l(pool_spin);
+ for (unsigned i = 0; i < num_workers; ++i) {
+ workers[i]->done = true;
+ workers[i]->center.wakeup();
+ join_worker(i);
+ }
+ started = false;
+}
+
+class C_drain : public EventCallback {
+ Mutex drain_lock;
+ Cond drain_cond;
+ unsigned drain_count;
+
+ public:
+ explicit C_drain(size_t c)
+ : drain_lock("C_drain::drain_lock"),
+ drain_count(c) {}
+ void do_request(int id) override {
+ Mutex::Locker l(drain_lock);
+ drain_count--;
+ if (drain_count == 0) drain_cond.Signal();
+ }
+ void wait() {
+ Mutex::Locker l(drain_lock);
+ while (drain_count)
+ drain_cond.Wait(drain_lock);
+ }
+};
+
+void NetworkStack::drain()
+{
+ ldout(cct, 30) << __func__ << " started." << dendl;
+ pthread_t cur = pthread_self();
+ pool_spin.lock();
+ C_drain drain(num_workers);
+ for (unsigned i = 0; i < num_workers; ++i) {
+ assert(cur != workers[i]->center.get_owner());
+ workers[i]->center.dispatch_event_external(EventCallbackRef(&drain));
+ }
+ pool_spin.unlock();
+ drain.wait();
+ ldout(cct, 30) << __func__ << " end." << dendl;
+}
Code Review / stor4nfv.git / blobdiff