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 class RGWRados;
  28
  29 class RGWRadosThread {
  30   class Worker : public Thread {
  31     CephContext *cct;
  32     RGWRadosThread *processor;
  33     ceph::mutex lock = ceph::make_mutex("RGWRadosThread::Worker");
  34     ceph::condition_variable cond;
  35
  36     void wait() {
  37       std::unique_lock l{lock};
  38       cond.wait(l);
  39     };
  40
  41     void wait_interval(const ceph::real_clock::duration& wait_time) {
  42       std::unique_lock l{lock};
  43       cond.wait_for(l, wait_time);
  44     }
  45
  46   public:
  47     Worker(CephContext *_cct, RGWRadosThread *_p) : cct(_cct), processor(_p) {}
  48     void *entry() override;
  49     void signal() {
  50       std::lock_guard l{lock};
  51       cond.notify_all();
  52     }
  53   };
  54
  55   Worker *worker;
  56
  57 protected:
  58   CephContext *cct;
  59   RGWRados *store;
  60
  61   std::atomic<bool> down_flag = { false };
  62
  63   string thread_name;
  64
  65   virtual uint64_t interval_msec() = 0;
  66   virtual void stop_process() {}
  67 public:
  68   RGWRadosThread(RGWRados *_store, const string& thread_name = "radosgw")
  69     : worker(NULL), cct(_store->ctx()), store(_store), thread_name(thread_name) {}
  70   virtual ~RGWRadosThread() {
  71     stop();
  72   }
  73
  74   virtual int init() { return 0; }
  75   virtual int process() = 0;
  76
  77   bool going_down() { return down_flag; }
  78
  79   void start();
  80   void stop();
  81
  82   void signal() {
  83     if (worker) {
  84       worker->signal();
  85     }
  86   }
  87 };
  88