+++ /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.
- *
- */
-
-#ifndef CEPH_FINISHER_H
-#define CEPH_FINISHER_H
-
-#include "common/Mutex.h"
-#include "common/Cond.h"
-#include "common/perf_counters.h"
-
-class CephContext;
-
-/// Finisher queue length performance counter ID.
-enum {
- l_finisher_first = 997082,
- l_finisher_queue_len,
- l_finisher_complete_lat,
- l_finisher_last
-};
-
-/** @brief Asynchronous cleanup class.
- * Finisher asynchronously completes Contexts, which are simple classes
- * representing callbacks, in a dedicated worker thread. Enqueuing
- * contexts to complete is thread-safe.
- */
-class Finisher {
- CephContext *cct;
- Mutex finisher_lock; ///< Protects access to queues and finisher_running.
- Cond finisher_cond; ///< Signaled when there is something to process.
- Cond finisher_empty_cond; ///< Signaled when the finisher has nothing more to process.
- bool finisher_stop; ///< Set when the finisher should stop.
- bool finisher_running; ///< True when the finisher is currently executing contexts.
- bool finisher_empty_wait; ///< True mean someone wait finisher empty.
- /// Queue for contexts for which complete(0) will be called.
- /// NULLs in this queue indicate that an item from finisher_queue_rval
- /// should be completed in that place instead.
- vector<Context*> finisher_queue;
-
- string thread_name;
-
- /// Queue for contexts for which the complete function will be called
- /// with a parameter other than 0.
- list<pair<Context*,int> > finisher_queue_rval;
-
- /// Performance counter for the finisher's queue length.
- /// Only active for named finishers.
- PerfCounters *logger;
-
- void *finisher_thread_entry();
-
- struct FinisherThread : public Thread {
- Finisher *fin;
- explicit FinisherThread(Finisher *f) : fin(f) {}
- void* entry() override { return (void*)fin->finisher_thread_entry(); }
- } finisher_thread;
-
- public:
- /// Add a context to complete, optionally specifying a parameter for the complete function.
- void queue(Context *c, int r = 0) {
- finisher_lock.Lock();
- if (finisher_queue.empty()) {
- finisher_cond.Signal();
- }
- if (r) {
- finisher_queue_rval.push_back(pair<Context*, int>(c, r));
- finisher_queue.push_back(NULL);
- } else
- finisher_queue.push_back(c);
- if (logger)
- logger->inc(l_finisher_queue_len);
- finisher_lock.Unlock();
- }
- void queue(vector<Context*>& ls) {
- finisher_lock.Lock();
- if (finisher_queue.empty()) {
- finisher_cond.Signal();
- }
- finisher_queue.insert(finisher_queue.end(), ls.begin(), ls.end());
- if (logger)
- logger->inc(l_finisher_queue_len, ls.size());
- finisher_lock.Unlock();
- ls.clear();
- }
- void queue(deque<Context*>& ls) {
- finisher_lock.Lock();
- if (finisher_queue.empty()) {
- finisher_cond.Signal();
- }
- finisher_queue.insert(finisher_queue.end(), ls.begin(), ls.end());
- if (logger)
- logger->inc(l_finisher_queue_len, ls.size());
- finisher_lock.Unlock();
- ls.clear();
- }
- void queue(list<Context*>& ls) {
- finisher_lock.Lock();
- if (finisher_queue.empty()) {
- finisher_cond.Signal();
- }
- finisher_queue.insert(finisher_queue.end(), ls.begin(), ls.end());
- if (logger)
- logger->inc(l_finisher_queue_len, ls.size());
- finisher_lock.Unlock();
- ls.clear();
- }
-
- /// Start the worker thread.
- void start();
-
- /** @brief Stop the worker thread.
- *
- * Does not wait until all outstanding contexts are completed.
- * To ensure that everything finishes, you should first shut down
- * all sources that can add contexts to this finisher and call
- * wait_for_empty() before calling stop(). */
- void stop();
-
- /** @brief Blocks until the finisher has nothing left to process.
- * This function will also return when a concurrent call to stop()
- * finishes, but this class should never be used in this way. */
- void wait_for_empty();
-
- /// Construct an anonymous Finisher.
- /// Anonymous finishers do not log their queue length.
- explicit Finisher(CephContext *cct_) :
- cct(cct_), finisher_lock("Finisher::finisher_lock"),
- finisher_stop(false), finisher_running(false), finisher_empty_wait(false),
- thread_name("fn_anonymous"), logger(0),
- finisher_thread(this) {}
-
- /// Construct a named Finisher that logs its queue length.
- Finisher(CephContext *cct_, string name, string tn) :
- cct(cct_), finisher_lock("Finisher::" + name),
- finisher_stop(false), finisher_running(false), finisher_empty_wait(false),
- thread_name(tn), logger(0),
- finisher_thread(this) {
- PerfCountersBuilder b(cct, string("finisher-") + name,
- l_finisher_first, l_finisher_last);
- b.add_u64(l_finisher_queue_len, "queue_len");
- b.add_time_avg(l_finisher_complete_lat, "complete_latency");
- logger = b.create_perf_counters();
- cct->get_perfcounters_collection()->add(logger);
- logger->set(l_finisher_queue_len, 0);
- logger->set(l_finisher_complete_lat, 0);
- }
-
- ~Finisher() {
- if (logger && cct) {
- cct->get_perfcounters_collection()->remove(logger);
- delete logger;
- }
- }
-};
-
-/// Context that is completed asynchronously on the supplied finisher.
-class C_OnFinisher : public Context {
- Context *con;
- Finisher *fin;
-public:
- C_OnFinisher(Context *c, Finisher *f) : con(c), fin(f) {
- assert(fin != NULL);
- assert(con != NULL);
- }
-
- ~C_OnFinisher() override {
- if (con != nullptr) {
- delete con;
- con = nullptr;
- }
- }
-
- void finish(int r) override {
- fin->queue(con, r);
- con = nullptr;
- }
-};
-
-#endif
Code Review / stor4nfv.git / blobdiff