--- /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) 2004-2006 Sage Weil <sage@newdream.net>
+ *
+ * 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 "WorkQueue.h"
+#include "include/compat.h"
+#include "common/errno.h"
+
+#define dout_subsys ceph_subsys_tp
+#undef dout_prefix
+#define dout_prefix *_dout << name << " "
+
+
+ThreadPool::ThreadPool(CephContext *cct_, string nm, string tn, int n, const char *option)
+ : cct(cct_), name(std::move(nm)), thread_name(std::move(tn)),
+ lockname(name + "::lock"),
+ _lock(lockname.c_str()), // this should be safe due to declaration order
+ _stop(false),
+ _pause(0),
+ _draining(0),
+ ioprio_class(-1),
+ ioprio_priority(-1),
+ _num_threads(n),
+ processing(0)
+{
+ if (option) {
+ _thread_num_option = option;
+ // set up conf_keys
+ _conf_keys = new const char*[2];
+ _conf_keys[0] = _thread_num_option.c_str();
+ _conf_keys[1] = NULL;
+ } else {
+ _conf_keys = new const char*[1];
+ _conf_keys[0] = NULL;
+ }
+}
+
+void ThreadPool::TPHandle::suspend_tp_timeout()
+{
+ cct->get_heartbeat_map()->clear_timeout(hb);
+}
+
+void ThreadPool::TPHandle::reset_tp_timeout()
+{
+ cct->get_heartbeat_map()->reset_timeout(
+ hb, grace, suicide_grace);
+}
+
+ThreadPool::~ThreadPool()
+{
+ assert(_threads.empty());
+ delete[] _conf_keys;
+}
+
+void ThreadPool::handle_conf_change(const struct md_config_t *conf,
+ const std::set <std::string> &changed)
+{
+ if (changed.count(_thread_num_option)) {
+ char *buf;
+ int r = conf->get_val(_thread_num_option.c_str(), &buf, -1);
+ assert(r >= 0);
+ int v = atoi(buf);
+ free(buf);
+ if (v >= 0) {
+ _lock.Lock();
+ _num_threads = v;
+ start_threads();
+ _cond.SignalAll();
+ _lock.Unlock();
+ }
+ }
+}
+
+void ThreadPool::worker(WorkThread *wt)
+{
+ _lock.Lock();
+ ldout(cct,10) << "worker start" << dendl;
+
+ std::stringstream ss;
+ ss << name << " thread " << (void *)pthread_self();
+ heartbeat_handle_d *hb = cct->get_heartbeat_map()->add_worker(ss.str(), pthread_self());
+
+ while (!_stop) {
+
+ // manage dynamic thread pool
+ join_old_threads();
+ if (_threads.size() > _num_threads) {
+ ldout(cct,1) << " worker shutting down; too many threads (" << _threads.size() << " > " << _num_threads << ")" << dendl;
+ _threads.erase(wt);
+ _old_threads.push_back(wt);
+ break;
+ }
+
+ if (!_pause && !work_queues.empty()) {
+ WorkQueue_* wq;
+ int tries = work_queues.size();
+ bool did = false;
+ while (tries--) {
+ next_work_queue %= work_queues.size();
+ wq = work_queues[next_work_queue++];
+
+ void *item = wq->_void_dequeue();
+ if (item) {
+ processing++;
+ ldout(cct,12) << "worker wq " << wq->name << " start processing " << item
+ << " (" << processing << " active)" << dendl;
+ TPHandle tp_handle(cct, hb, wq->timeout_interval, wq->suicide_interval);
+ tp_handle.reset_tp_timeout();
+ _lock.Unlock();
+ wq->_void_process(item, tp_handle);
+ _lock.Lock();
+ wq->_void_process_finish(item);
+ processing--;
+ ldout(cct,15) << "worker wq " << wq->name << " done processing " << item
+ << " (" << processing << " active)" << dendl;
+ if (_pause || _draining)
+ _wait_cond.Signal();
+ did = true;
+ break;
+ }
+ }
+ if (did)
+ continue;
+ }
+
+ ldout(cct,20) << "worker waiting" << dendl;
+ cct->get_heartbeat_map()->reset_timeout(
+ hb,
+ cct->_conf->threadpool_default_timeout,
+ 0);
+ _cond.WaitInterval(_lock,
+ utime_t(
+ cct->_conf->threadpool_empty_queue_max_wait, 0));
+ }
+ ldout(cct,1) << "worker finish" << dendl;
+
+ cct->get_heartbeat_map()->remove_worker(hb);
+
+ _lock.Unlock();
+}
+
+void ThreadPool::start_threads()
+{
+ assert(_lock.is_locked());
+ while (_threads.size() < _num_threads) {
+ WorkThread *wt = new WorkThread(this);
+ ldout(cct, 10) << "start_threads creating and starting " << wt << dendl;
+ _threads.insert(wt);
+
+ int r = wt->set_ioprio(ioprio_class, ioprio_priority);
+ if (r < 0)
+ lderr(cct) << " set_ioprio got " << cpp_strerror(r) << dendl;
+
+ wt->create(thread_name.c_str());
+ }
+}
+
+void ThreadPool::join_old_threads()
+{
+ assert(_lock.is_locked());
+ while (!_old_threads.empty()) {
+ ldout(cct, 10) << "join_old_threads joining and deleting " << _old_threads.front() << dendl;
+ _old_threads.front()->join();
+ delete _old_threads.front();
+ _old_threads.pop_front();
+ }
+}
+
+void ThreadPool::start()
+{
+ ldout(cct,10) << "start" << dendl;
+
+ if (_thread_num_option.length()) {
+ ldout(cct, 10) << " registering config observer on " << _thread_num_option << dendl;
+ cct->_conf->add_observer(this);
+ }
+
+ _lock.Lock();
+ start_threads();
+ _lock.Unlock();
+ ldout(cct,15) << "started" << dendl;
+}
+
+void ThreadPool::stop(bool clear_after)
+{
+ ldout(cct,10) << "stop" << dendl;
+
+ if (_thread_num_option.length()) {
+ ldout(cct, 10) << " unregistering config observer on " << _thread_num_option << dendl;
+ cct->_conf->remove_observer(this);
+ }
+
+ _lock.Lock();
+ _stop = true;
+ _cond.Signal();
+ join_old_threads();
+ _lock.Unlock();
+ for (set<WorkThread*>::iterator p = _threads.begin();
+ p != _threads.end();
+ ++p) {
+ (*p)->join();
+ delete *p;
+ }
+ _threads.clear();
+ _lock.Lock();
+ for (unsigned i=0; i<work_queues.size(); i++)
+ work_queues[i]->_clear();
+ _stop = false;
+ _lock.Unlock();
+ ldout(cct,15) << "stopped" << dendl;
+}
+
+void ThreadPool::pause()
+{
+ ldout(cct,10) << "pause" << dendl;
+ _lock.Lock();
+ _pause++;
+ while (processing)
+ _wait_cond.Wait(_lock);
+ _lock.Unlock();
+ ldout(cct,15) << "paused" << dendl;
+}
+
+void ThreadPool::pause_new()
+{
+ ldout(cct,10) << "pause_new" << dendl;
+ _lock.Lock();
+ _pause++;
+ _lock.Unlock();
+}
+
+void ThreadPool::unpause()
+{
+ ldout(cct,10) << "unpause" << dendl;
+ _lock.Lock();
+ assert(_pause > 0);
+ _pause--;
+ _cond.Signal();
+ _lock.Unlock();
+}
+
+void ThreadPool::drain(WorkQueue_* wq)
+{
+ ldout(cct,10) << "drain" << dendl;
+ _lock.Lock();
+ _draining++;
+ while (processing || (wq != NULL && !wq->_empty()))
+ _wait_cond.Wait(_lock);
+ _draining--;
+ _lock.Unlock();
+}
+
+void ThreadPool::set_ioprio(int cls, int priority)
+{
+ Mutex::Locker l(_lock);
+ ioprio_class = cls;
+ ioprio_priority = priority;
+ for (set<WorkThread*>::iterator p = _threads.begin();
+ p != _threads.end();
+ ++p) {
+ ldout(cct,10) << __func__
+ << " class " << cls << " priority " << priority
+ << " pid " << (*p)->get_pid()
+ << dendl;
+ int r = (*p)->set_ioprio(cls, priority);
+ if (r < 0)
+ lderr(cct) << " set_ioprio got " << cpp_strerror(r) << dendl;
+ }
+}
+
+ShardedThreadPool::ShardedThreadPool(CephContext *pcct_, string nm, string tn,
+ uint32_t pnum_threads):
+ cct(pcct_),
+ name(std::move(nm)),
+ thread_name(std::move(tn)),
+ lockname(name + "::lock"),
+ shardedpool_lock(lockname.c_str()),
+ num_threads(pnum_threads),
+ num_paused(0),
+ num_drained(0),
+ wq(NULL) {}
+
+void ShardedThreadPool::shardedthreadpool_worker(uint32_t thread_index)
+{
+ assert(wq != NULL);
+ ldout(cct,10) << "worker start" << dendl;
+
+ std::stringstream ss;
+ ss << name << " thread " << (void *)pthread_self();
+ heartbeat_handle_d *hb = cct->get_heartbeat_map()->add_worker(ss.str(), pthread_self());
+
+ while (!stop_threads) {
+ if (pause_threads) {
+ shardedpool_lock.Lock();
+ ++num_paused;
+ wait_cond.Signal();
+ while (pause_threads) {
+ cct->get_heartbeat_map()->reset_timeout(
+ hb,
+ wq->timeout_interval, wq->suicide_interval);
+ shardedpool_cond.WaitInterval(shardedpool_lock,
+ utime_t(
+ cct->_conf->threadpool_empty_queue_max_wait, 0));
+ }
+ --num_paused;
+ shardedpool_lock.Unlock();
+ }
+ if (drain_threads) {
+ shardedpool_lock.Lock();
+ if (wq->is_shard_empty(thread_index)) {
+ ++num_drained;
+ wait_cond.Signal();
+ while (drain_threads) {
+ cct->get_heartbeat_map()->reset_timeout(
+ hb,
+ wq->timeout_interval, wq->suicide_interval);
+ shardedpool_cond.WaitInterval(shardedpool_lock,
+ utime_t(
+ cct->_conf->threadpool_empty_queue_max_wait, 0));
+ }
+ --num_drained;
+ }
+ shardedpool_lock.Unlock();
+ }
+
+ cct->get_heartbeat_map()->reset_timeout(
+ hb,
+ wq->timeout_interval, wq->suicide_interval);
+ wq->_process(thread_index, hb);
+
+ }
+
+ ldout(cct,10) << "sharded worker finish" << dendl;
+
+ cct->get_heartbeat_map()->remove_worker(hb);
+
+}
+
+void ShardedThreadPool::start_threads()
+{
+ assert(shardedpool_lock.is_locked());
+ int32_t thread_index = 0;
+ while (threads_shardedpool.size() < num_threads) {
+
+ WorkThreadSharded *wt = new WorkThreadSharded(this, thread_index);
+ ldout(cct, 10) << "start_threads creating and starting " << wt << dendl;
+ threads_shardedpool.push_back(wt);
+ wt->create(thread_name.c_str());
+ thread_index++;
+ }
+}
+
+void ShardedThreadPool::start()
+{
+ ldout(cct,10) << "start" << dendl;
+
+ shardedpool_lock.Lock();
+ start_threads();
+ shardedpool_lock.Unlock();
+ ldout(cct,15) << "started" << dendl;
+}
+
+void ShardedThreadPool::stop()
+{
+ ldout(cct,10) << "stop" << dendl;
+ stop_threads = true;
+ assert(wq != NULL);
+ wq->return_waiting_threads();
+ for (vector<WorkThreadSharded*>::iterator p = threads_shardedpool.begin();
+ p != threads_shardedpool.end();
+ ++p) {
+ (*p)->join();
+ delete *p;
+ }
+ threads_shardedpool.clear();
+ ldout(cct,15) << "stopped" << dendl;
+}
+
+void ShardedThreadPool::pause()
+{
+ ldout(cct,10) << "pause" << dendl;
+ shardedpool_lock.Lock();
+ pause_threads = true;
+ assert(wq != NULL);
+ wq->return_waiting_threads();
+ while (num_threads != num_paused){
+ wait_cond.Wait(shardedpool_lock);
+ }
+ shardedpool_lock.Unlock();
+ ldout(cct,10) << "paused" << dendl;
+}
+
+void ShardedThreadPool::pause_new()
+{
+ ldout(cct,10) << "pause_new" << dendl;
+ shardedpool_lock.Lock();
+ pause_threads = true;
+ assert(wq != NULL);
+ wq->return_waiting_threads();
+ shardedpool_lock.Unlock();
+ ldout(cct,10) << "paused_new" << dendl;
+}
+
+void ShardedThreadPool::unpause()
+{
+ ldout(cct,10) << "unpause" << dendl;
+ shardedpool_lock.Lock();
+ pause_threads = false;
+ shardedpool_cond.Signal();
+ shardedpool_lock.Unlock();
+ ldout(cct,10) << "unpaused" << dendl;
+}
+
+void ShardedThreadPool::drain()
+{
+ ldout(cct,10) << "drain" << dendl;
+ shardedpool_lock.Lock();
+ drain_threads = true;
+ assert(wq != NULL);
+ wq->return_waiting_threads();
+ while (num_threads != num_drained) {
+ wait_cond.Wait(shardedpool_lock);
+ }
+ drain_threads = false;
+ shardedpool_cond.Signal();
+ shardedpool_lock.Unlock();
+ ldout(cct,10) << "drained" << dendl;
+}
+
Code Review / stor4nfv.git / blobdiff