ceph/src/rgw/rgw_worker.h

   1
   2
   3 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   4 // vim: ts=8 sw=2 smarttab ft=cpp
   5
   6 /*
   7  * Ceph - scalable distributed file system
   8  *
   9  * Copyright (C) 2019 Red Hat, Inc.
  10  *
  11  * This is free software; you can redistribute it and/or
  12  * modify it under the terms of the GNU Lesser General Public
  13  * License version 2.1, as published by the Free Software
  14  * Foundation. See file COPYING.
  15  *
  16  */
  17
  18
  19 #pragma once
  20
  21 #include <atomic>
  22
  23 #include "common/Thread.h"
  24 #include "common/ceph_mutex.h"
  25 #include "include/common_fwd.h"
  26
  27 #define dout_subsys ceph_subsys_rgw
  28
  29 class RGWRados;
  30
  31 class RGWRadosThread {
  32   class Worker : public Thread, public DoutPrefixProvider {
  33     CephContext *cct;
  34     RGWRadosThread *processor;
  35     ceph::mutex lock = ceph::make_mutex("RGWRadosThread::Worker");
  36     ceph::condition_variable cond;
  37
  38     void wait() {
  39       std::unique_lock l{lock};
  40       cond.wait(l);
  41     };
  42
  43     void wait_interval(const ceph::real_clock::duration& wait_time) {
  44       std::unique_lock l{lock};
  45       cond.wait_for(l, wait_time);
  46     }
  47
  48   public:
  49     Worker(CephContext *_cct, RGWRadosThread *_p) : cct(_cct), processor(_p) {}
  50     void *entry() override;
  51     void signal() {
  52       std::lock_guard l{lock};
  53       cond.notify_all();
  54     }
  55
  56   CephContext *get_cct() const { return cct; }
  57   unsigned get_subsys() const { return dout_subsys; }
  58   std::ostream& gen_prefix(std::ostream& out) const { return out << "rgw rados thread: "; }
  59
  60   };
  61
  62   Worker *worker;
  63
  64 protected:
  65   CephContext *cct;
  66   RGWRados *store;
  67
  68   std::atomic<bool> down_flag = { false };
  69
  70   string thread_name;
  71
  72   virtual uint64_t interval_msec() = 0;
  73   virtual void stop_process() {}
  74 public:
  75   RGWRadosThread(RGWRados *_store, const string& thread_name = "radosgw")
  76     : worker(NULL), cct(_store->ctx()), store(_store), thread_name(thread_name) {}
  77   virtual ~RGWRadosThread() {
  78     stop();
  79   }
  80
  81   virtual int init(const DoutPrefixProvider *dpp) { return 0; }
  82   virtual int process(const DoutPrefixProvider *dpp) = 0;
  83
  84   bool going_down() { return down_flag; }
  85
  86   void start();
  87   void stop();
  88
  89   void signal() {
  90     if (worker) {
  91       worker->signal();
  92     }
  93   }
  94 };
  95