summaryrefslogtreecommitdiffstats
path: root/src/rgw/services/svc_notify.h
blob: f7329136ece1718d6951700bd3b1cd65711e0066 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab ft=cpp

#pragma once

#include "rgw_service.h"

#include "svc_rados.h"


class Context;

class RGWSI_Zone;
class RGWSI_Finisher;

class RGWWatcher;
class RGWSI_Notify_ShutdownCB;
struct RGWCacheNotifyInfo;

class RGWSI_Notify : public RGWServiceInstance
{
  friend class RGWWatcher;
  friend class RGWSI_Notify_ShutdownCB;
  friend class RGWServices_Def;

public:
  class CB;

private:
  RGWSI_Zone *zone_svc{nullptr};
  RGWSI_RADOS *rados_svc{nullptr};
  RGWSI_Finisher *finisher_svc{nullptr};

  ceph::shared_mutex watchers_lock = ceph::make_shared_mutex("watchers_lock");
  rgw_pool control_pool;

  int num_watchers{0};
  RGWWatcher **watchers{nullptr};
  std::set<int> watchers_set;
  std::vector<RGWSI_RADOS::Obj> notify_objs;

  bool enabled{false};

  double inject_notify_timeout_probability{0};
  uint64_t max_notify_retries = 10;

  std::string get_control_oid(int i);
  RGWSI_RADOS::Obj pick_control_obj(const std::string& key);

  CB *cb{nullptr};

  std::optional<int> finisher_handle;
  RGWSI_Notify_ShutdownCB *shutdown_cb{nullptr};

  bool finalized{false};

  int init_watch(const DoutPrefixProvider *dpp, optional_yield y);
  void finalize_watch();

  void init(RGWSI_Zone *_zone_svc,
            RGWSI_RADOS *_rados_svc,
            RGWSI_Finisher *_finisher_svc) {
    zone_svc = _zone_svc;
    rados_svc = _rados_svc;
    finisher_svc = _finisher_svc;
  }
  int do_start(optional_yield, const DoutPrefixProvider *dpp) override;
  void shutdown() override;

  int unwatch(RGWSI_RADOS::Obj& obj, uint64_t watch_handle);
  void add_watcher(int i);
  void remove_watcher(int i);

  int watch_cb(const DoutPrefixProvider *dpp,
               uint64_t notify_id,
               uint64_t cookie,
               uint64_t notifier_id,
               bufferlist& bl);
  void _set_enabled(bool status);
  void set_enabled(bool status);

  int robust_notify(const DoutPrefixProvider *dpp, RGWSI_RADOS::Obj& notify_obj,
		    const RGWCacheNotifyInfo& bl, optional_yield y);

  void schedule_context(Context *c);
public:
  RGWSI_Notify(CephContext *cct): RGWServiceInstance(cct) {}

  virtual ~RGWSI_Notify() override;

  class CB {
    public:
      virtual ~CB() {}
      virtual int watch_cb(const DoutPrefixProvider *dpp,
                           uint64_t notify_id,
                           uint64_t cookie,
                           uint64_t notifier_id,
                           bufferlist& bl) = 0;
      virtual void set_enabled(bool status) = 0;
  };

  int distribute(const DoutPrefixProvider *dpp, const std::string& key, const RGWCacheNotifyInfo& bl,
		 optional_yield y);

  void register_watch_cb(CB *cb);
};