+++ /dev/null
-#ifndef CEPH_RGW_CR_RADOS_H
-#define CEPH_RGW_CR_RADOS_H
-
-#include <boost/intrusive_ptr.hpp>
-#include "include/assert.h"
-#include "rgw_coroutine.h"
-#include "rgw_rados.h"
-#include "common/WorkQueue.h"
-#include "common/Throttle.h"
-
-#include <atomic>
-
-class RGWAsyncRadosRequest : public RefCountedObject {
- RGWCoroutine *caller;
- RGWAioCompletionNotifier *notifier;
-
- int retcode;
-
- Mutex lock;
-
-protected:
- virtual int _send_request() = 0;
-public:
- RGWAsyncRadosRequest(RGWCoroutine *_caller, RGWAioCompletionNotifier *_cn) : caller(_caller), notifier(_cn), retcode(0),
- lock("RGWAsyncRadosRequest::lock") {
- }
- ~RGWAsyncRadosRequest() override {
- if (notifier) {
- notifier->put();
- }
- }
-
- void send_request() {
- get();
- retcode = _send_request();
- {
- Mutex::Locker l(lock);
- if (notifier) {
- notifier->cb(); // drops its own ref
- notifier = nullptr;
- }
- }
- put();
- }
-
- int get_ret_status() { return retcode; }
-
- void finish() {
- {
- Mutex::Locker l(lock);
- if (notifier) {
- // we won't call notifier->cb() to drop its ref, so drop it here
- notifier->put();
- notifier = nullptr;
- }
- }
- put();
- }
-};
-
-
-class RGWAsyncRadosProcessor {
- deque<RGWAsyncRadosRequest *> m_req_queue;
- std::atomic<bool> going_down = { false };
-protected:
- RGWRados *store;
- ThreadPool m_tp;
- Throttle req_throttle;
-
- struct RGWWQ : public ThreadPool::WorkQueue<RGWAsyncRadosRequest> {
- RGWAsyncRadosProcessor *processor;
- RGWWQ(RGWAsyncRadosProcessor *p, time_t timeout, time_t suicide_timeout, ThreadPool *tp)
- : ThreadPool::WorkQueue<RGWAsyncRadosRequest>("RGWWQ", timeout, suicide_timeout, tp), processor(p) {}
-
- bool _enqueue(RGWAsyncRadosRequest *req) override;
- void _dequeue(RGWAsyncRadosRequest *req) override {
- ceph_abort();
- }
- bool _empty() override;
- RGWAsyncRadosRequest *_dequeue() override;
- using ThreadPool::WorkQueue<RGWAsyncRadosRequest>::_process;
- void _process(RGWAsyncRadosRequest *req, ThreadPool::TPHandle& handle) override;
- void _dump_queue();
- void _clear() override {
- assert(processor->m_req_queue.empty());
- }
- } req_wq;
-
-public:
- RGWAsyncRadosProcessor(RGWRados *_store, int num_threads);
- ~RGWAsyncRadosProcessor() {}
- void start();
- void stop();
- void handle_request(RGWAsyncRadosRequest *req);
- void queue(RGWAsyncRadosRequest *req);
-
- bool is_going_down() {
- return going_down;
- }
-};
-
-
-class RGWAsyncGetSystemObj : public RGWAsyncRadosRequest {
- RGWRados *store;
- RGWObjectCtx *obj_ctx;
- RGWRados::SystemObject::Read::GetObjState read_state;
- RGWObjVersionTracker *objv_tracker;
- rgw_raw_obj obj;
- bufferlist *pbl;
- map<string, bufferlist> *pattrs;
- off_t ofs;
- off_t end;
-protected:
- int _send_request() override;
-public:
- RGWAsyncGetSystemObj(RGWCoroutine *caller, RGWAioCompletionNotifier *cn, RGWRados *_store, RGWObjectCtx *_obj_ctx,
- RGWObjVersionTracker *_objv_tracker, const rgw_raw_obj& _obj,
- bufferlist *_pbl, off_t _ofs, off_t _end);
- void set_read_attrs(map<string, bufferlist> *_pattrs) { pattrs = _pattrs; }
-};
-
-class RGWAsyncPutSystemObj : public RGWAsyncRadosRequest {
- RGWRados *store;
- RGWObjVersionTracker *objv_tracker;
- rgw_raw_obj obj;
- bool exclusive;
- bufferlist bl;
-
-protected:
- int _send_request() override;
-public:
- RGWAsyncPutSystemObj(RGWCoroutine *caller, RGWAioCompletionNotifier *cn, RGWRados *_store,
- RGWObjVersionTracker *_objv_tracker, rgw_raw_obj& _obj,
- bool _exclusive, bufferlist& _bl);
-};
-
-class RGWAsyncPutSystemObjAttrs : public RGWAsyncRadosRequest {
- RGWRados *store;
- RGWObjVersionTracker *objv_tracker;
- rgw_raw_obj obj;
- map<string, bufferlist> *attrs;
-
-protected:
- int _send_request() override;
-public:
- RGWAsyncPutSystemObjAttrs(RGWCoroutine *caller, RGWAioCompletionNotifier *cn, RGWRados *_store,
- RGWObjVersionTracker *_objv_tracker, const rgw_raw_obj& _obj,
- map<string, bufferlist> *_attrs);
-};
-
-class RGWAsyncLockSystemObj : public RGWAsyncRadosRequest {
- RGWRados *store;
- rgw_raw_obj obj;
- string lock_name;
- string cookie;
- uint32_t duration_secs;
-
-protected:
- int _send_request() override;
-public:
- RGWAsyncLockSystemObj(RGWCoroutine *caller, RGWAioCompletionNotifier *cn, RGWRados *_store,
- RGWObjVersionTracker *_objv_tracker, const rgw_raw_obj& _obj,
- const string& _name, const string& _cookie, uint32_t _duration_secs);
-};
-
-class RGWAsyncUnlockSystemObj : public RGWAsyncRadosRequest {
- RGWRados *store;
- rgw_raw_obj obj;
- string lock_name;
- string cookie;
-
-protected:
- int _send_request() override;
-public:
- RGWAsyncUnlockSystemObj(RGWCoroutine *caller, RGWAioCompletionNotifier *cn, RGWRados *_store,
- RGWObjVersionTracker *_objv_tracker, const rgw_raw_obj& _obj,
- const string& _name, const string& _cookie);
-};
-
-template <class T>
-class RGWSimpleRadosReadCR : public RGWSimpleCoroutine {
- RGWAsyncRadosProcessor *async_rados;
- RGWRados *store;
- RGWObjectCtx obj_ctx;
- bufferlist bl;
-
- rgw_raw_obj obj;
-
- map<string, bufferlist> *pattrs{nullptr};
-
- T *result;
- /// on ENOENT, call handle_data() with an empty object instead of failing
- const bool empty_on_enoent;
- RGWObjVersionTracker *objv_tracker;
-
- RGWAsyncGetSystemObj *req{nullptr};
-
-public:
- RGWSimpleRadosReadCR(RGWAsyncRadosProcessor *_async_rados, RGWRados *_store,
- const rgw_raw_obj& _obj,
- T *_result, bool empty_on_enoent = true,
- RGWObjVersionTracker *objv_tracker = nullptr)
- : RGWSimpleCoroutine(_store->ctx()), async_rados(_async_rados), store(_store),
- obj_ctx(store), obj(_obj), result(_result),
- empty_on_enoent(empty_on_enoent), objv_tracker(objv_tracker) {}
- ~RGWSimpleRadosReadCR() override {
- request_cleanup();
- }
-
- void request_cleanup() override {
- if (req) {
- req->finish();
- req = NULL;
- }
- }
-
- int send_request() override;
- int request_complete() override;
-
- virtual int handle_data(T& data) {
- return 0;
- }
-};
-
-template <class T>
-int RGWSimpleRadosReadCR<T>::send_request()
-{
- req = new RGWAsyncGetSystemObj(this, stack->create_completion_notifier(),
- store, &obj_ctx, objv_tracker,
- obj,
- &bl, 0, -1);
- if (pattrs) {
- req->set_read_attrs(pattrs);
- }
- async_rados->queue(req);
- return 0;
-}
-
-template <class T>
-int RGWSimpleRadosReadCR<T>::request_complete()
-{
- int ret = req->get_ret_status();
- retcode = ret;
- if (ret == -ENOENT && empty_on_enoent) {
- *result = T();
- } else {
- if (ret < 0) {
- return ret;
- }
- try {
- bufferlist::iterator iter = bl.begin();
- if (iter.end()) {
- // allow successful reads with empty buffers. ReadSyncStatus coroutines
- // depend on this to be able to read without locking, because the
- // cls lock from InitSyncStatus will create an empty object if it didnt
- // exist
- *result = T();
- } else {
- ::decode(*result, iter);
- }
- } catch (buffer::error& err) {
- return -EIO;
- }
- }
-
- return handle_data(*result);
-}
-
-class RGWSimpleRadosReadAttrsCR : public RGWSimpleCoroutine {
- RGWAsyncRadosProcessor *async_rados;
- RGWRados *store;
- RGWObjectCtx obj_ctx;
- bufferlist bl;
-
- rgw_raw_obj obj;
-
- map<string, bufferlist> *pattrs;
-
- RGWAsyncGetSystemObj *req;
-
-public:
- RGWSimpleRadosReadAttrsCR(RGWAsyncRadosProcessor *_async_rados, RGWRados *_store,
- const rgw_raw_obj& _obj,
- map<string, bufferlist> *_pattrs) : RGWSimpleCoroutine(_store->ctx()),
- async_rados(_async_rados), store(_store),
- obj_ctx(store),
- obj(_obj),
- pattrs(_pattrs),
- req(NULL) { }
- ~RGWSimpleRadosReadAttrsCR() override {
- request_cleanup();
- }
-
- void request_cleanup() override {
- if (req) {
- req->finish();
- req = NULL;
- }
- }
-
- int send_request() override;
- int request_complete() override;
-};
-
-template <class T>
-class RGWSimpleRadosWriteCR : public RGWSimpleCoroutine {
- RGWAsyncRadosProcessor *async_rados;
- RGWRados *store;
- bufferlist bl;
-
- rgw_raw_obj obj;
- RGWObjVersionTracker *objv_tracker;
-
- RGWAsyncPutSystemObj *req{nullptr};
-
-public:
- RGWSimpleRadosWriteCR(RGWAsyncRadosProcessor *_async_rados, RGWRados *_store,
- const rgw_raw_obj& _obj,
- const T& _data, RGWObjVersionTracker *objv_tracker = nullptr)
- : RGWSimpleCoroutine(_store->ctx()), async_rados(_async_rados),
- store(_store), obj(_obj), objv_tracker(objv_tracker) {
- ::encode(_data, bl);
- }
-
- ~RGWSimpleRadosWriteCR() override {
- request_cleanup();
- }
-
- void request_cleanup() override {
- if (req) {
- req->finish();
- req = NULL;
- }
- }
-
- int send_request() override {
- req = new RGWAsyncPutSystemObj(this, stack->create_completion_notifier(),
- store, objv_tracker, obj, false, bl);
- async_rados->queue(req);
- return 0;
- }
-
- int request_complete() override {
- return req->get_ret_status();
- }
-};
-
-class RGWSimpleRadosWriteAttrsCR : public RGWSimpleCoroutine {
- RGWAsyncRadosProcessor *async_rados;
- RGWRados *store;
-
- rgw_raw_obj obj;
-
- map<string, bufferlist> attrs;
-
- RGWAsyncPutSystemObjAttrs *req;
-
-public:
- RGWSimpleRadosWriteAttrsCR(RGWAsyncRadosProcessor *_async_rados, RGWRados *_store,
- const rgw_raw_obj& _obj,
- map<string, bufferlist>& _attrs) : RGWSimpleCoroutine(_store->ctx()),
- async_rados(_async_rados),
- store(_store),
- obj(_obj),
- attrs(_attrs), req(NULL) {
- }
- ~RGWSimpleRadosWriteAttrsCR() override {
- request_cleanup();
- }
-
- void request_cleanup() override {
- if (req) {
- req->finish();
- req = NULL;
- }
- }
-
- int send_request() override {
- req = new RGWAsyncPutSystemObjAttrs(this, stack->create_completion_notifier(),
- store, NULL, obj, &attrs);
- async_rados->queue(req);
- return 0;
- }
-
- int request_complete() override {
- return req->get_ret_status();
- }
-};
-
-class RGWRadosSetOmapKeysCR : public RGWSimpleCoroutine {
- RGWRados *store;
- map<string, bufferlist> entries;
-
- rgw_rados_ref ref;
-
- rgw_raw_obj obj;
-
- boost::intrusive_ptr<RGWAioCompletionNotifier> cn;
-
-public:
- RGWRadosSetOmapKeysCR(RGWRados *_store,
- const rgw_raw_obj& _obj,
- map<string, bufferlist>& _entries);
-
- int send_request() override;
- int request_complete() override;
-};
-
-class RGWRadosGetOmapKeysCR : public RGWSimpleCoroutine {
- RGWRados *store;
-
- string marker;
- map<string, bufferlist> *entries;
- int max_entries;
-
- int rval;
- rgw_rados_ref ref;
-
- rgw_raw_obj obj;
-
- boost::intrusive_ptr<RGWAioCompletionNotifier> cn;
-
-public:
- RGWRadosGetOmapKeysCR(RGWRados *_store,
- const rgw_raw_obj& _obj,
- const string& _marker,
- map<string, bufferlist> *_entries, int _max_entries);
-
- int send_request() override;
-
- int request_complete() override {
- return rval;
- }
-};
-
-class RGWRadosRemoveOmapKeysCR : public RGWSimpleCoroutine {
- RGWRados *store;
-
- rgw_rados_ref ref;
-
- set<string> keys;
-
- rgw_raw_obj obj;
-
- boost::intrusive_ptr<RGWAioCompletionNotifier> cn;
-
-public:
- RGWRadosRemoveOmapKeysCR(RGWRados *_store,
- const rgw_raw_obj& _obj,
- const set<string>& _keys);
-
- int send_request() override;
-
- int request_complete() override;
-};
-
-class RGWRadosRemoveCR : public RGWSimpleCoroutine {
- RGWRados *store;
- librados::IoCtx ioctx;
- const rgw_raw_obj obj;
- boost::intrusive_ptr<RGWAioCompletionNotifier> cn;
-
-public:
- RGWRadosRemoveCR(RGWRados *store, const rgw_raw_obj& obj);
-
- int send_request();
- int request_complete();
-};
-
-class RGWSimpleRadosLockCR : public RGWSimpleCoroutine {
- RGWAsyncRadosProcessor *async_rados;
- RGWRados *store;
- string lock_name;
- string cookie;
- uint32_t duration;
-
- rgw_raw_obj obj;
-
- RGWAsyncLockSystemObj *req;
-
-public:
- RGWSimpleRadosLockCR(RGWAsyncRadosProcessor *_async_rados, RGWRados *_store,
- const rgw_raw_obj& _obj,
- const string& _lock_name,
- const string& _cookie,
- uint32_t _duration);
- ~RGWSimpleRadosLockCR() override {
- request_cleanup();
- }
- void request_cleanup() override;
-
- int send_request() override;
- int request_complete() override;
-
- static std::string gen_random_cookie(CephContext* cct) {
-#define COOKIE_LEN 16
- char buf[COOKIE_LEN + 1];
- gen_rand_alphanumeric(cct, buf, sizeof(buf) - 1);
- return buf;
- }
-};
-
-class RGWSimpleRadosUnlockCR : public RGWSimpleCoroutine {
- RGWAsyncRadosProcessor *async_rados;
- RGWRados *store;
- string lock_name;
- string cookie;
-
- rgw_raw_obj obj;
-
- RGWAsyncUnlockSystemObj *req;
-
-public:
- RGWSimpleRadosUnlockCR(RGWAsyncRadosProcessor *_async_rados, RGWRados *_store,
- const rgw_raw_obj& _obj,
- const string& _lock_name,
- const string& _cookie);
- ~RGWSimpleRadosUnlockCR() override {
- request_cleanup();
- }
- void request_cleanup() override;
-
- int send_request() override;
- int request_complete() override;
-};
-
-#define OMAP_APPEND_MAX_ENTRIES_DEFAULT 100
-
-class RGWOmapAppend : public RGWConsumerCR<string> {
- RGWAsyncRadosProcessor *async_rados;
- RGWRados *store;
-
- rgw_raw_obj obj;
-
- bool going_down;
-
- int num_pending_entries;
- list<string> pending_entries;
-
- map<string, bufferlist> entries;
-
- uint64_t window_size;
- uint64_t total_entries;
-public:
- RGWOmapAppend(RGWAsyncRadosProcessor *_async_rados, RGWRados *_store,
- const rgw_raw_obj& _obj,
- uint64_t _window_size = OMAP_APPEND_MAX_ENTRIES_DEFAULT);
- int operate() override;
- void flush_pending();
- bool append(const string& s);
- bool finish();
-
- uint64_t get_total_entries() {
- return total_entries;
- }
-
- const rgw_raw_obj& get_obj() {
- return obj;
- }
-};
-
-class RGWAsyncWait : public RGWAsyncRadosRequest {
- CephContext *cct;
- Mutex *lock;
- Cond *cond;
- utime_t interval;
-protected:
- int _send_request() override {
- Mutex::Locker l(*lock);
- return cond->WaitInterval(*lock, interval);
- }
-public:
- RGWAsyncWait(RGWCoroutine *caller, RGWAioCompletionNotifier *cn, CephContext *_cct,
- Mutex *_lock, Cond *_cond, int _secs) : RGWAsyncRadosRequest(caller, cn),
- cct(_cct),
- lock(_lock), cond(_cond), interval(_secs, 0) {}
-
- void wakeup() {
- Mutex::Locker l(*lock);
- cond->Signal();
- }
-};
-
-class RGWWaitCR : public RGWSimpleCoroutine {
- CephContext *cct;
- RGWAsyncRadosProcessor *async_rados;
- Mutex *lock;
- Cond *cond;
- int secs;
-
- RGWAsyncWait *req;
-
-public:
- RGWWaitCR(RGWAsyncRadosProcessor *_async_rados, CephContext *_cct,
- Mutex *_lock, Cond *_cond,
- int _secs) : RGWSimpleCoroutine(_cct), cct(_cct),
- async_rados(_async_rados), lock(_lock), cond(_cond), secs(_secs), req(NULL) {
- }
- ~RGWWaitCR() override {
- request_cleanup();
- }
-
- void request_cleanup() override {
- if (req) {
- wakeup();
- req->finish();
- req = NULL;
- }
- }
-
- int send_request() override {
- req = new RGWAsyncWait(this, stack->create_completion_notifier(), cct, lock, cond, secs);
- async_rados->queue(req);
- return 0;
- }
-
- int request_complete() override {
- return req->get_ret_status();
- }
-
- void wakeup() {
- req->wakeup();
- }
-};
-
-class RGWShardedOmapCRManager {
- RGWAsyncRadosProcessor *async_rados;
- RGWRados *store;
- RGWCoroutine *op;
-
- int num_shards;
-
- vector<RGWOmapAppend *> shards;
-public:
- RGWShardedOmapCRManager(RGWAsyncRadosProcessor *_async_rados, RGWRados *_store, RGWCoroutine *_op, int _num_shards, const rgw_pool& pool, const string& oid_prefix)
- : async_rados(_async_rados),
- store(_store), op(_op), num_shards(_num_shards) {
- shards.reserve(num_shards);
- for (int i = 0; i < num_shards; ++i) {
- char buf[oid_prefix.size() + 16];
- snprintf(buf, sizeof(buf), "%s.%d", oid_prefix.c_str(), i);
- RGWOmapAppend *shard = new RGWOmapAppend(async_rados, store, rgw_raw_obj(pool, buf));
- shard->get();
- shards.push_back(shard);
- op->spawn(shard, false);
- }
- }
-
- ~RGWShardedOmapCRManager() {
- for (auto shard : shards) {
- shard->put();
- }
- }
-
- bool append(const string& entry, int shard_id) {
- return shards[shard_id]->append(entry);
- }
- bool finish() {
- bool success = true;
- for (vector<RGWOmapAppend *>::iterator iter = shards.begin(); iter != shards.end(); ++iter) {
- success &= ((*iter)->finish() && (!(*iter)->is_error()));
- }
- return success;
- }
-
- uint64_t get_total_entries(int shard_id) {
- return shards[shard_id]->get_total_entries();
- }
-};
-
-class RGWAsyncGetBucketInstanceInfo : public RGWAsyncRadosRequest {
- RGWRados *store;
- rgw_bucket bucket;
- RGWBucketInfo *bucket_info;
-
-protected:
- int _send_request() override;
-public:
- RGWAsyncGetBucketInstanceInfo(RGWCoroutine *caller, RGWAioCompletionNotifier *cn,
- RGWRados *_store, const rgw_bucket& bucket,
- RGWBucketInfo *_bucket_info)
- : RGWAsyncRadosRequest(caller, cn), store(_store),
- bucket(bucket), bucket_info(_bucket_info) {}
-};
-
-class RGWGetBucketInstanceInfoCR : public RGWSimpleCoroutine {
- RGWAsyncRadosProcessor *async_rados;
- RGWRados *store;
- rgw_bucket bucket;
- RGWBucketInfo *bucket_info;
-
- RGWAsyncGetBucketInstanceInfo *req;
-
-public:
- RGWGetBucketInstanceInfoCR(RGWAsyncRadosProcessor *_async_rados, RGWRados *_store,
- const rgw_bucket& bucket, RGWBucketInfo *_bucket_info)
- : RGWSimpleCoroutine(_store->ctx()), async_rados(_async_rados), store(_store),
- bucket(bucket), bucket_info(_bucket_info), req(NULL) {}
- ~RGWGetBucketInstanceInfoCR() override {
- request_cleanup();
- }
- void request_cleanup() override {
- if (req) {
- req->finish();
- req = NULL;
- }
- }
-
- int send_request() override {
- req = new RGWAsyncGetBucketInstanceInfo(this, stack->create_completion_notifier(), store, bucket, bucket_info);
- async_rados->queue(req);
- return 0;
- }
- int request_complete() override {
- return req->get_ret_status();
- }
-};
-
-class RGWAsyncFetchRemoteObj : public RGWAsyncRadosRequest {
- RGWRados *store;
- string source_zone;
-
- RGWBucketInfo bucket_info;
-
- rgw_obj_key key;
- uint64_t versioned_epoch;
-
- real_time src_mtime;
-
- bool copy_if_newer;
- rgw_zone_set *zones_trace;
-
-protected:
- int _send_request() override;
-public:
- RGWAsyncFetchRemoteObj(RGWCoroutine *caller, RGWAioCompletionNotifier *cn, RGWRados *_store,
- const string& _source_zone,
- RGWBucketInfo& _bucket_info,
- const rgw_obj_key& _key,
- uint64_t _versioned_epoch,
- bool _if_newer, rgw_zone_set *_zones_trace) : RGWAsyncRadosRequest(caller, cn), store(_store),
- source_zone(_source_zone),
- bucket_info(_bucket_info),
- key(_key),
- versioned_epoch(_versioned_epoch),
- copy_if_newer(_if_newer), zones_trace(_zones_trace) {}
-};
-
-class RGWFetchRemoteObjCR : public RGWSimpleCoroutine {
- CephContext *cct;
- RGWAsyncRadosProcessor *async_rados;
- RGWRados *store;
- string source_zone;
-
- RGWBucketInfo bucket_info;
-
- rgw_obj_key key;
- uint64_t versioned_epoch;
-
- real_time src_mtime;
-
- bool copy_if_newer;
-
- RGWAsyncFetchRemoteObj *req;
- rgw_zone_set *zones_trace;
-
-public:
- RGWFetchRemoteObjCR(RGWAsyncRadosProcessor *_async_rados, RGWRados *_store,
- const string& _source_zone,
- RGWBucketInfo& _bucket_info,
- const rgw_obj_key& _key,
- uint64_t _versioned_epoch,
- bool _if_newer, rgw_zone_set *_zones_trace) : RGWSimpleCoroutine(_store->ctx()), cct(_store->ctx()),
- async_rados(_async_rados), store(_store),
- source_zone(_source_zone),
- bucket_info(_bucket_info),
- key(_key),
- versioned_epoch(_versioned_epoch),
- copy_if_newer(_if_newer), req(NULL), zones_trace(_zones_trace) {}
-
-
- ~RGWFetchRemoteObjCR() override {
- request_cleanup();
- }
-
- void request_cleanup() override {
- if (req) {
- req->finish();
- req = NULL;
- }
- }
-
- int send_request() override {
- req = new RGWAsyncFetchRemoteObj(this, stack->create_completion_notifier(), store, source_zone, bucket_info,
- key, versioned_epoch, copy_if_newer, zones_trace);
- async_rados->queue(req);
- return 0;
- }
-
- int request_complete() override {
- return req->get_ret_status();
- }
-};
-
-class RGWAsyncStatRemoteObj : public RGWAsyncRadosRequest {
- RGWRados *store;
- string source_zone;
-
- RGWBucketInfo bucket_info;
-
- rgw_obj_key key;
-
- ceph::real_time *pmtime;
- uint64_t *psize;
- map<string, bufferlist> *pattrs;
-
-protected:
- int _send_request() override;
-public:
- RGWAsyncStatRemoteObj(RGWCoroutine *caller, RGWAioCompletionNotifier *cn, RGWRados *_store,
- const string& _source_zone,
- RGWBucketInfo& _bucket_info,
- const rgw_obj_key& _key,
- ceph::real_time *_pmtime,
- uint64_t *_psize,
- map<string, bufferlist> *_pattrs) : RGWAsyncRadosRequest(caller, cn), store(_store),
- source_zone(_source_zone),
- bucket_info(_bucket_info),
- key(_key),
- pmtime(_pmtime),
- psize(_psize),
- pattrs(_pattrs) {}
-};
-
-class RGWStatRemoteObjCR : public RGWSimpleCoroutine {
- CephContext *cct;
- RGWAsyncRadosProcessor *async_rados;
- RGWRados *store;
- string source_zone;
-
- RGWBucketInfo bucket_info;
-
- rgw_obj_key key;
-
- ceph::real_time *pmtime;
- uint64_t *psize;
- map<string, bufferlist> *pattrs;
-
- RGWAsyncStatRemoteObj *req;
-
-public:
- RGWStatRemoteObjCR(RGWAsyncRadosProcessor *_async_rados, RGWRados *_store,
- const string& _source_zone,
- RGWBucketInfo& _bucket_info,
- const rgw_obj_key& _key,
- ceph::real_time *_pmtime,
- uint64_t *_psize,
- map<string, bufferlist> *_pattrs) : RGWSimpleCoroutine(_store->ctx()), cct(_store->ctx()),
- async_rados(_async_rados), store(_store),
- source_zone(_source_zone),
- bucket_info(_bucket_info),
- key(_key),
- pmtime(_pmtime),
- psize(_psize),
- pattrs(_pattrs),
- req(NULL) {}
-
-
- ~RGWStatRemoteObjCR() override {
- request_cleanup();
- }
-
- void request_cleanup() override {
- if (req) {
- req->finish();
- req = NULL;
- }
- }
-
- int send_request() override {
- req = new RGWAsyncStatRemoteObj(this, stack->create_completion_notifier(), store, source_zone,
- bucket_info, key, pmtime, psize, pattrs);
- async_rados->queue(req);
- return 0;
- }
-
- int request_complete() override {
- return req->get_ret_status();
- }
-};
-
-class RGWAsyncRemoveObj : public RGWAsyncRadosRequest {
- RGWRados *store;
- string source_zone;
-
- RGWBucketInfo bucket_info;
-
- rgw_obj_key key;
- string owner;
- string owner_display_name;
- bool versioned;
- uint64_t versioned_epoch;
- string marker_version_id;
-
- bool del_if_older;
- ceph::real_time timestamp;
- rgw_zone_set *zones_trace;
-
-protected:
- int _send_request() override;
-public:
- RGWAsyncRemoveObj(RGWCoroutine *caller, RGWAioCompletionNotifier *cn, RGWRados *_store,
- const string& _source_zone,
- RGWBucketInfo& _bucket_info,
- const rgw_obj_key& _key,
- const string& _owner,
- const string& _owner_display_name,
- bool _versioned,
- uint64_t _versioned_epoch,
- bool _delete_marker,
- bool _if_older,
- real_time& _timestamp,
- rgw_zone_set* _zones_trace) : RGWAsyncRadosRequest(caller, cn), store(_store),
- source_zone(_source_zone),
- bucket_info(_bucket_info),
- key(_key),
- owner(_owner),
- owner_display_name(_owner_display_name),
- versioned(_versioned),
- versioned_epoch(_versioned_epoch),
- del_if_older(_if_older),
- timestamp(_timestamp), zones_trace(_zones_trace) {
- if (_delete_marker) {
- marker_version_id = key.instance;
- }
- }
-};
-
-class RGWRemoveObjCR : public RGWSimpleCoroutine {
- CephContext *cct;
- RGWAsyncRadosProcessor *async_rados;
- RGWRados *store;
- string source_zone;
-
- RGWBucketInfo bucket_info;
-
- rgw_obj_key key;
- bool versioned;
- uint64_t versioned_epoch;
- bool delete_marker;
- string owner;
- string owner_display_name;
-
- bool del_if_older;
- real_time timestamp;
-
- RGWAsyncRemoveObj *req;
-
- rgw_zone_set *zones_trace;
-
-public:
- RGWRemoveObjCR(RGWAsyncRadosProcessor *_async_rados, RGWRados *_store,
- const string& _source_zone,
- RGWBucketInfo& _bucket_info,
- const rgw_obj_key& _key,
- bool _versioned,
- uint64_t _versioned_epoch,
- string *_owner,
- string *_owner_display_name,
- bool _delete_marker,
- real_time *_timestamp,
- rgw_zone_set *_zones_trace) : RGWSimpleCoroutine(_store->ctx()), cct(_store->ctx()),
- async_rados(_async_rados), store(_store),
- source_zone(_source_zone),
- bucket_info(_bucket_info),
- key(_key),
- versioned(_versioned),
- versioned_epoch(_versioned_epoch),
- delete_marker(_delete_marker), req(NULL), zones_trace(_zones_trace) {
- del_if_older = (_timestamp != NULL);
- if (_timestamp) {
- timestamp = *_timestamp;
- }
-
- if (_owner) {
- owner = *_owner;
- }
-
- if (_owner_display_name) {
- owner_display_name = *_owner_display_name;
- }
- }
- ~RGWRemoveObjCR() override {
- request_cleanup();
- }
-
- void request_cleanup() override {
- if (req) {
- req->finish();
- req = NULL;
- }
- }
-
- int send_request() override {
- req = new RGWAsyncRemoveObj(this, stack->create_completion_notifier(), store, source_zone, bucket_info,
- key, owner, owner_display_name, versioned, versioned_epoch,
- delete_marker, del_if_older, timestamp, zones_trace);
- async_rados->queue(req);
- return 0;
- }
-
- int request_complete() override {
- return req->get_ret_status();
- }
-};
-
-class RGWContinuousLeaseCR : public RGWCoroutine {
- RGWAsyncRadosProcessor *async_rados;
- RGWRados *store;
-
- const rgw_raw_obj obj;
-
- const string lock_name;
- const string cookie;
-
- int interval;
-
- Mutex lock;
- std::atomic<bool> going_down = { false };
- bool locked{false};
-
- RGWCoroutine *caller;
-
- bool aborted{false};
-
-public:
- RGWContinuousLeaseCR(RGWAsyncRadosProcessor *_async_rados, RGWRados *_store,
- const rgw_raw_obj& _obj,
- const string& _lock_name, int _interval, RGWCoroutine *_caller)
- : RGWCoroutine(_store->ctx()), async_rados(_async_rados), store(_store),
- obj(_obj), lock_name(_lock_name),
- cookie(RGWSimpleRadosLockCR::gen_random_cookie(cct)),
- interval(_interval), lock("RGWContinuousLeaseCR"), caller(_caller)
- {}
-
- int operate() override;
-
- bool is_locked() {
- Mutex::Locker l(lock);
- return locked;
- }
-
- void set_locked(bool status) {
- Mutex::Locker l(lock);
- locked = status;
- }
-
- void go_down() {
- going_down = true;
- wakeup();
- }
-
- void abort() {
- aborted = true;
- }
-};
-
-class RGWRadosTimelogAddCR : public RGWSimpleCoroutine {
- RGWRados *store;
- list<cls_log_entry> entries;
-
- string oid;
-
- boost::intrusive_ptr<RGWAioCompletionNotifier> cn;
-
-public:
- RGWRadosTimelogAddCR(RGWRados *_store, const string& _oid,
- const cls_log_entry& entry);
-
- int send_request() override;
- int request_complete() override;
-};
-
-class RGWRadosTimelogTrimCR : public RGWSimpleCoroutine {
- RGWRados *store;
- boost::intrusive_ptr<RGWAioCompletionNotifier> cn;
- protected:
- std::string oid;
- real_time start_time;
- real_time end_time;
- std::string from_marker;
- std::string to_marker;
-
- public:
- RGWRadosTimelogTrimCR(RGWRados *store, const std::string& oid,
- const real_time& start_time, const real_time& end_time,
- const std::string& from_marker,
- const std::string& to_marker);
-
- int send_request() override;
- int request_complete() override;
-};
-
-// wrapper to update last_trim_marker on success
-class RGWSyncLogTrimCR : public RGWRadosTimelogTrimCR {
- CephContext *cct;
- std::string *last_trim_marker;
- public:
- RGWSyncLogTrimCR(RGWRados *store, const std::string& oid,
- const std::string& to_marker, std::string *last_trim_marker);
- int request_complete() override;
-};
-
-class RGWAsyncStatObj : public RGWAsyncRadosRequest {
- RGWRados *store;
- RGWBucketInfo bucket_info;
- rgw_obj obj;
- uint64_t *psize;
- real_time *pmtime;
- uint64_t *pepoch;
- RGWObjVersionTracker *objv_tracker;
-protected:
- int _send_request() override;
-public:
- RGWAsyncStatObj(RGWCoroutine *caller, RGWAioCompletionNotifier *cn, RGWRados *store,
- const RGWBucketInfo& _bucket_info, const rgw_obj& obj, uint64_t *psize = nullptr,
- real_time *pmtime = nullptr, uint64_t *pepoch = nullptr,
- RGWObjVersionTracker *objv_tracker = nullptr)
- : RGWAsyncRadosRequest(caller, cn), store(store), obj(obj), psize(psize),
- pmtime(pmtime), pepoch(pepoch), objv_tracker(objv_tracker) {}
-};
-
-class RGWStatObjCR : public RGWSimpleCoroutine {
- RGWRados *store;
- RGWAsyncRadosProcessor *async_rados;
- RGWBucketInfo bucket_info;
- rgw_obj obj;
- uint64_t *psize;
- real_time *pmtime;
- uint64_t *pepoch;
- RGWObjVersionTracker *objv_tracker;
- RGWAsyncStatObj *req = nullptr;
- public:
- RGWStatObjCR(RGWAsyncRadosProcessor *async_rados, RGWRados *store,
- const RGWBucketInfo& _bucket_info, const rgw_obj& obj, uint64_t *psize = nullptr,
- real_time* pmtime = nullptr, uint64_t *pepoch = nullptr,
- RGWObjVersionTracker *objv_tracker = nullptr);
- ~RGWStatObjCR() override {
- request_cleanup();
- }
- void request_cleanup() override;
-
- int send_request() override;
- int request_complete() override;
-};
-
-#endif
Code Review / stor4nfv.git / blobdiff