ceph/src/rgw/services/svc_notify.h

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab ft=cpp
   3
   4 #pragma once
   5
   6 #include "rgw/rgw_service.h"
   7
   8 #include "svc_rados.h"
   9
  10
  11 class Context;
  12
  13 class RGWSI_Zone;
  14 class RGWSI_Finisher;
  15
  16 class RGWWatcher;
  17 class RGWSI_Notify_ShutdownCB;
  18
  19 class RGWSI_Notify : public RGWServiceInstance
  20 {
  21   friend class RGWWatcher;
  22   friend class RGWSI_Notify_ShutdownCB;
  23   friend class RGWServices_Def;
  24
  25 public:
  26   class CB;
  27
  28 private:
  29   RGWSI_Zone *zone_svc{nullptr};
  30   RGWSI_RADOS *rados_svc{nullptr};
  31   RGWSI_Finisher *finisher_svc{nullptr};
  32
  33   ceph::shared_mutex watchers_lock = ceph::make_shared_mutex("watchers_lock");
  34   rgw_pool control_pool;
  35
  36   int num_watchers{0};
  37   RGWWatcher **watchers{nullptr};
  38   std::set<int> watchers_set;
  39   vector<RGWSI_RADOS::Obj> notify_objs;
  40
  41   bool enabled{false};
  42
  43   double inject_notify_timeout_probability{0};
  44   unsigned max_notify_retries{0};
  45
  46   string get_control_oid(int i);
  47   RGWSI_RADOS::Obj pick_control_obj(const string& key);
  48
  49   CB *cb{nullptr};
  50
  51   std::optional<int> finisher_handle;
  52   RGWSI_Notify_ShutdownCB *shutdown_cb{nullptr};
  53
  54   bool finalized{false};
  55
  56   int init_watch(optional_yield y);
  57   void finalize_watch();
  58
  59   void init(RGWSI_Zone *_zone_svc,
  60             RGWSI_RADOS *_rados_svc,
  61             RGWSI_Finisher *_finisher_svc) {
  62     zone_svc = _zone_svc;
  63     rados_svc = _rados_svc;
  64     finisher_svc = _finisher_svc;
  65   }
  66   int do_start(optional_yield) override;
  67   void shutdown() override;
  68
  69   int unwatch(RGWSI_RADOS::Obj& obj, uint64_t watch_handle);
  70   void add_watcher(int i);
  71   void remove_watcher(int i);
  72
  73   int watch_cb(uint64_t notify_id,
  74                uint64_t cookie,
  75                uint64_t notifier_id,
  76                bufferlist& bl);
  77   void _set_enabled(bool status);
  78   void set_enabled(bool status);
  79
  80   int robust_notify(RGWSI_RADOS::Obj& notify_obj, bufferlist& bl,
  81                     optional_yield y);
  82
  83   void schedule_context(Context *c);
  84 public:
  85   RGWSI_Notify(CephContext *cct): RGWServiceInstance(cct) {}
  86   ~RGWSI_Notify();
  87
  88   class CB {
  89     public:
  90       virtual ~CB() {}
  91       virtual int watch_cb(uint64_t notify_id,
  92                            uint64_t cookie,
  93                            uint64_t notifier_id,
  94                            bufferlist& bl) = 0;
  95       virtual void set_enabled(bool status) = 0;
  96   };
  97
  98   int distribute(const string& key, bufferlist& bl, optional_yield y);
  99
 100   void register_watch_cb(CB *cb);
 101 };