+++ /dev/null
-// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-// vim: ts=8 sw=2 smarttab
-#include "include/int_types.h"
-
-#include "common/Mutex.h"
-#include "common/Cond.h"
-#include "include/rados/librados.hpp"
-
-#include <iostream>
-#include <sstream>
-#include <map>
-#include <set>
-#include <list>
-#include <string>
-#include <string.h>
-#include <stdlib.h>
-#include <errno.h>
-#include <time.h>
-#include "Object.h"
-#include "TestOpStat.h"
-#include "test/librados/test.h"
-#include "include/memory.h"
-#include "common/sharedptr_registry.hpp"
-#include "common/errno.h"
-#include "osd/HitSet.h"
-
-#ifndef RADOSMODEL_H
-#define RADOSMODEL_H
-
-using namespace std;
-
-class RadosTestContext;
-class TestOpStat;
-
-template <typename T>
-typename T::iterator rand_choose(T &cont) {
- if (cont.size() == 0) {
- return cont.end();
- }
- int index = rand() % cont.size();
- typename T::iterator retval = cont.begin();
-
- for (; index > 0; --index) ++retval;
- return retval;
-}
-
-enum TestOpType {
- TEST_OP_READ,
- TEST_OP_WRITE,
- TEST_OP_WRITE_EXCL,
- TEST_OP_WRITESAME,
- TEST_OP_DELETE,
- TEST_OP_SNAP_CREATE,
- TEST_OP_SNAP_REMOVE,
- TEST_OP_ROLLBACK,
- TEST_OP_SETATTR,
- TEST_OP_RMATTR,
- TEST_OP_WATCH,
- TEST_OP_COPY_FROM,
- TEST_OP_HIT_SET_LIST,
- TEST_OP_UNDIRTY,
- TEST_OP_IS_DIRTY,
- TEST_OP_CACHE_FLUSH,
- TEST_OP_CACHE_TRY_FLUSH,
- TEST_OP_CACHE_EVICT,
- TEST_OP_APPEND,
- TEST_OP_APPEND_EXCL,
- TEST_OP_SET_REDIRECT,
- TEST_OP_UNSET_REDIRECT
-};
-
-class TestWatchContext : public librados::WatchCtx2 {
- TestWatchContext(const TestWatchContext&);
-public:
- Cond cond;
- uint64_t handle;
- bool waiting;
- Mutex lock;
- TestWatchContext() : handle(0), waiting(false),
- lock("watch lock") {}
- void handle_notify(uint64_t notify_id, uint64_t cookie,
- uint64_t notifier_id,
- bufferlist &bl) override {
- Mutex::Locker l(lock);
- waiting = false;
- cond.SignalAll();
- }
- void handle_error(uint64_t cookie, int err) override {
- Mutex::Locker l(lock);
- cout << "watch handle_error " << err << std::endl;
- }
- void start() {
- Mutex::Locker l(lock);
- waiting = true;
- }
- void wait() {
- Mutex::Locker l(lock);
- while (waiting)
- cond.Wait(lock);
- }
- uint64_t &get_handle() {
- return handle;
- }
-};
-
-class TestOp {
-public:
- int num;
- RadosTestContext *context;
- TestOpStat *stat;
- bool done;
- TestOp(int n, RadosTestContext *context,
- TestOpStat *stat = 0)
- : num(n),
- context(context),
- stat(stat),
- done(false)
- {}
-
- virtual ~TestOp() {};
-
- /**
- * This struct holds data to be passed by a callback
- * to a TestOp::finish method.
- */
- struct CallbackInfo {
- uint64_t id;
- explicit CallbackInfo(uint64_t id) : id(id) {}
- virtual ~CallbackInfo() {};
- };
-
- virtual void _begin() = 0;
-
- /**
- * Called when the operation completes.
- * This should be overridden by asynchronous operations.
- *
- * @param info information stored by a callback, or NULL -
- * useful for multi-operation TestOps
- */
- virtual void _finish(CallbackInfo *info)
- {
- return;
- }
- virtual string getType() = 0;
- virtual bool finished()
- {
- return true;
- }
-
- void begin();
- void finish(CallbackInfo *info);
- virtual bool must_quiesce_other_ops() { return false; }
-};
-
-class TestOpGenerator {
-public:
- virtual ~TestOpGenerator() {};
- virtual TestOp *next(RadosTestContext &context) = 0;
-};
-
-class RadosTestContext {
-public:
- Mutex state_lock;
- Cond wait_cond;
- map<int, map<string,ObjectDesc> > pool_obj_cont;
- set<string> oid_in_use;
- set<string> oid_not_in_use;
- set<string> oid_flushing;
- set<string> oid_not_flushing;
- set<string> oid_redirect_not_in_use;
- set<string> oid_redirect_in_use;
- SharedPtrRegistry<int, int> snaps_in_use;
- int current_snap;
- string pool_name;
- librados::IoCtx io_ctx;
- librados::Rados rados;
- int next_oid;
- string prefix;
- int errors;
- int max_in_flight;
- int seq_num;
- map<int,uint64_t> snaps;
- uint64_t seq;
- const char *rados_id;
- bool initialized;
- map<string, TestWatchContext*> watches;
- const uint64_t max_size;
- const uint64_t min_stride_size;
- const uint64_t max_stride_size;
- AttrGenerator attr_gen;
- const bool no_omap;
- const bool no_sparse;
- bool pool_snaps;
- bool write_fadvise_dontneed;
- int snapname_num;
- map<string,string > redirect_objs;
-
- RadosTestContext(const string &pool_name,
- int max_in_flight,
- uint64_t max_size,
- uint64_t min_stride_size,
- uint64_t max_stride_size,
- bool no_omap,
- bool no_sparse,
- bool pool_snaps,
- bool write_fadvise_dontneed,
- const char *id = 0) :
- state_lock("Context Lock"),
- pool_obj_cont(),
- current_snap(0),
- pool_name(pool_name),
- next_oid(0),
- errors(0),
- max_in_flight(max_in_flight),
- seq_num(0), seq(0),
- rados_id(id), initialized(false),
- max_size(max_size),
- min_stride_size(min_stride_size), max_stride_size(max_stride_size),
- attr_gen(2000, 20000),
- no_omap(no_omap),
- no_sparse(no_sparse),
- pool_snaps(pool_snaps),
- write_fadvise_dontneed(write_fadvise_dontneed),
- snapname_num(0)
- {
- }
-
- int init()
- {
- int r = rados.init(rados_id);
- if (r < 0)
- return r;
- r = rados.conf_read_file(NULL);
- if (r < 0)
- return r;
- r = rados.conf_parse_env(NULL);
- if (r < 0)
- return r;
- r = rados.connect();
- if (r < 0)
- return r;
- r = rados.ioctx_create(pool_name.c_str(), io_ctx);
- if (r < 0) {
- rados.shutdown();
- return r;
- }
- bufferlist inbl;
- r = rados.mon_command(
- "{\"prefix\": \"osd pool set\", \"pool\": \"" + pool_name +
- "\", \"var\": \"write_fadvise_dontneed\", \"val\": \"" + (write_fadvise_dontneed ? "true" : "false") + "\"}",
- inbl, NULL, NULL);
- if (r < 0) {
- rados.shutdown();
- return r;
- }
- char hostname_cstr[100];
- gethostname(hostname_cstr, 100);
- stringstream hostpid;
- hostpid << hostname_cstr << getpid() << "-";
- prefix = hostpid.str();
- assert(!initialized);
- initialized = true;
- return 0;
- }
-
- void shutdown()
- {
- if (initialized) {
- rados.shutdown();
- }
- }
-
- void loop(TestOpGenerator *gen)
- {
- assert(initialized);
- list<TestOp*> inflight;
- state_lock.Lock();
-
- TestOp *next = gen->next(*this);
- TestOp *waiting = NULL;
-
- while (next || !inflight.empty()) {
- if (next && next->must_quiesce_other_ops() && !inflight.empty()) {
- waiting = next;
- next = NULL; // Force to wait for inflight to drain
- }
- if (next) {
- inflight.push_back(next);
- }
- state_lock.Unlock();
- if (next) {
- (*inflight.rbegin())->begin();
- }
- state_lock.Lock();
- while (1) {
- for (list<TestOp*>::iterator i = inflight.begin();
- i != inflight.end();) {
- if ((*i)->finished()) {
- cout << (*i)->num << ": done (" << (inflight.size()-1) << " left)" << std::endl;
- delete *i;
- inflight.erase(i++);
- } else {
- ++i;
- }
- }
-
- if (inflight.size() >= (unsigned) max_in_flight || (!next && !inflight.empty())) {
- cout << " waiting on " << inflight.size() << std::endl;
- wait();
- } else {
- break;
- }
- }
- if (waiting) {
- next = waiting;
- waiting = NULL;
- } else {
- next = gen->next(*this);
- }
- }
- state_lock.Unlock();
- }
-
- void wait()
- {
- wait_cond.Wait(state_lock);
- }
-
- void kick()
- {
- wait_cond.Signal();
- }
-
- TestWatchContext *get_watch_context(const string &oid) {
- return watches.count(oid) ? watches[oid] : 0;
- }
-
- TestWatchContext *watch(const string &oid) {
- assert(!watches.count(oid));
- return (watches[oid] = new TestWatchContext);
- }
-
- void unwatch(const string &oid) {
- assert(watches.count(oid));
- delete watches[oid];
- watches.erase(oid);
- }
-
- ObjectDesc get_most_recent(const string &oid) {
- ObjectDesc new_obj;
- for (map<int, map<string,ObjectDesc> >::reverse_iterator i =
- pool_obj_cont.rbegin();
- i != pool_obj_cont.rend();
- ++i) {
- map<string,ObjectDesc>::iterator j = i->second.find(oid);
- if (j != i->second.end()) {
- new_obj = j->second;
- break;
- }
- }
- return new_obj;
- }
-
- void rm_object_attrs(const string &oid, const set<string> &attrs)
- {
- ObjectDesc new_obj = get_most_recent(oid);
- for (set<string>::const_iterator i = attrs.begin();
- i != attrs.end();
- ++i) {
- new_obj.attrs.erase(*i);
- }
- new_obj.dirty = true;
- pool_obj_cont[current_snap].erase(oid);
- pool_obj_cont[current_snap].insert(pair<string,ObjectDesc>(oid, new_obj));
- }
-
- void remove_object_header(const string &oid)
- {
- ObjectDesc new_obj = get_most_recent(oid);
- new_obj.header = bufferlist();
- new_obj.dirty = true;
- pool_obj_cont[current_snap].erase(oid);
- pool_obj_cont[current_snap].insert(pair<string,ObjectDesc>(oid, new_obj));
- }
-
-
- void update_object_header(const string &oid, const bufferlist &bl)
- {
- ObjectDesc new_obj = get_most_recent(oid);
- new_obj.header = bl;
- new_obj.exists = true;
- new_obj.dirty = true;
- pool_obj_cont[current_snap].erase(oid);
- pool_obj_cont[current_snap].insert(pair<string,ObjectDesc>(oid, new_obj));
- }
-
- void update_object_attrs(const string &oid, const map<string, ContDesc> &attrs)
- {
- ObjectDesc new_obj = get_most_recent(oid);
- for (map<string, ContDesc>::const_iterator i = attrs.begin();
- i != attrs.end();
- ++i) {
- new_obj.attrs[i->first] = i->second;
- }
- new_obj.exists = true;
- new_obj.dirty = true;
- pool_obj_cont[current_snap].erase(oid);
- pool_obj_cont[current_snap].insert(pair<string,ObjectDesc>(oid, new_obj));
- }
-
- void update_object(ContentsGenerator *cont_gen,
- const string &oid, const ContDesc &contents)
- {
- ObjectDesc new_obj = get_most_recent(oid);
- new_obj.exists = true;
- new_obj.dirty = true;
- new_obj.update(cont_gen,
- contents);
- pool_obj_cont[current_snap].erase(oid);
- pool_obj_cont[current_snap].insert(pair<string,ObjectDesc>(oid, new_obj));
- }
-
- void update_object_full(const string &oid, const ObjectDesc &contents)
- {
- pool_obj_cont[current_snap].erase(oid);
- pool_obj_cont[current_snap].insert(pair<string,ObjectDesc>(oid, contents));
- pool_obj_cont[current_snap][oid].dirty = true;
- }
-
- void update_object_undirty(const string &oid)
- {
- ObjectDesc new_obj = get_most_recent(oid);
- new_obj.dirty = false;
- pool_obj_cont[current_snap].erase(oid);
- pool_obj_cont[current_snap].insert(pair<string,ObjectDesc>(oid, new_obj));
- }
-
- void update_object_version(const string &oid, uint64_t version,
- int snap = -1)
- {
- for (map<int, map<string,ObjectDesc> >::reverse_iterator i =
- pool_obj_cont.rbegin();
- i != pool_obj_cont.rend();
- ++i) {
- if (snap != -1 && snap < i->first)
- continue;
- map<string,ObjectDesc>::iterator j = i->second.find(oid);
- if (j != i->second.end()) {
- if (version)
- j->second.version = version;
- cout << __func__ << " oid " << oid
- << " v " << version << " " << j->second.most_recent()
- << " " << (j->second.dirty ? "dirty" : "clean")
- << " " << (j->second.exists ? "exists" : "dne")
- << std::endl;
- break;
- }
- }
- }
-
- void remove_object(const string &oid)
- {
- assert(!get_watch_context(oid));
- ObjectDesc new_obj;
- pool_obj_cont[current_snap].erase(oid);
- pool_obj_cont[current_snap].insert(pair<string,ObjectDesc>(oid, new_obj));
- }
-
- bool find_object(const string &oid, ObjectDesc *contents, int snap = -1) const
- {
- for (map<int, map<string,ObjectDesc> >::const_reverse_iterator i =
- pool_obj_cont.rbegin();
- i != pool_obj_cont.rend();
- ++i) {
- if (snap != -1 && snap < i->first) continue;
- if (i->second.count(oid) != 0) {
- *contents = i->second.find(oid)->second;
- return true;
- }
- }
- return false;
- }
-
- void update_object_redirect_target(const string &oid, const string &target)
- {
- redirect_objs[oid] = target;
- }
-
- bool object_existed_at(const string &oid, int snap = -1) const
- {
- ObjectDesc contents;
- bool found = find_object(oid, &contents, snap);
- return found && contents.exists;
- }
-
- void remove_snap(int snap)
- {
- map<int, map<string,ObjectDesc> >::iterator next_iter = pool_obj_cont.find(snap);
- assert(next_iter != pool_obj_cont.end());
- map<int, map<string,ObjectDesc> >::iterator current_iter = next_iter++;
- assert(current_iter != pool_obj_cont.end());
- map<string,ObjectDesc> ¤t = current_iter->second;
- map<string,ObjectDesc> &next = next_iter->second;
- for (map<string,ObjectDesc>::iterator i = current.begin();
- i != current.end();
- ++i) {
- if (next.count(i->first) == 0) {
- next.insert(pair<string,ObjectDesc>(i->first, i->second));
- }
- }
- pool_obj_cont.erase(current_iter);
- snaps.erase(snap);
- }
-
- void add_snap(uint64_t snap)
- {
- snaps[current_snap] = snap;
- current_snap++;
- pool_obj_cont[current_snap];
- seq = snap;
- }
-
- void roll_back(const string &oid, int snap)
- {
- assert(!get_watch_context(oid));
- ObjectDesc contents;
- find_object(oid, &contents, snap);
- contents.dirty = true;
- pool_obj_cont.rbegin()->second.erase(oid);
- pool_obj_cont.rbegin()->second.insert(pair<string,ObjectDesc>(oid, contents));
- }
-};
-
-void read_callback(librados::completion_t comp, void *arg);
-void write_callback(librados::completion_t comp, void *arg);
-
-class RemoveAttrsOp : public TestOp {
-public:
- string oid;
- librados::ObjectWriteOperation op;
- librados::AioCompletion *comp;
- RemoveAttrsOp(int n, RadosTestContext *context,
- const string &oid,
- TestOpStat *stat)
- : TestOp(n, context, stat), oid(oid), comp(NULL)
- {}
-
- void _begin() override
- {
- ContDesc cont;
- set<string> to_remove;
- {
- Mutex::Locker l(context->state_lock);
- ObjectDesc obj;
- if (!context->find_object(oid, &obj)) {
- context->kick();
- done = true;
- return;
- }
- cont = ContDesc(context->seq_num, context->current_snap,
- context->seq_num, "");
- context->oid_in_use.insert(oid);
- context->oid_not_in_use.erase(oid);
-
- if (rand() % 30) {
- ContentsGenerator::iterator iter = context->attr_gen.get_iterator(cont);
- for (map<string, ContDesc>::iterator i = obj.attrs.begin();
- i != obj.attrs.end();
- ++i, ++iter) {
- if (!(*iter % 3)) {
- to_remove.insert(i->first);
- op.rmxattr(i->first.c_str());
- }
- }
- if (to_remove.empty()) {
- context->kick();
- context->oid_in_use.erase(oid);
- context->oid_not_in_use.insert(oid);
- done = true;
- return;
- }
- if (!context->no_omap) {
- op.omap_rm_keys(to_remove);
- }
- } else {
- if (!context->no_omap) {
- op.omap_clear();
- }
- for (map<string, ContDesc>::iterator i = obj.attrs.begin();
- i != obj.attrs.end();
- ++i) {
- op.rmxattr(i->first.c_str());
- to_remove.insert(i->first);
- }
- context->remove_object_header(oid);
- }
- context->rm_object_attrs(oid, to_remove);
- }
-
- pair<TestOp*, TestOp::CallbackInfo*> *cb_arg =
- new pair<TestOp*, TestOp::CallbackInfo*>(this,
- new TestOp::CallbackInfo(0));
- comp = context->rados.aio_create_completion((void*) cb_arg, NULL,
- &write_callback);
- context->io_ctx.aio_operate(context->prefix+oid, comp, &op);
- }
-
- void _finish(CallbackInfo *info) override
- {
- Mutex::Locker l(context->state_lock);
- done = true;
- context->update_object_version(oid, comp->get_version64());
- context->oid_in_use.erase(oid);
- context->oid_not_in_use.insert(oid);
- context->kick();
- }
-
- bool finished() override
- {
- return done;
- }
-
- string getType() override
- {
- return "RemoveAttrsOp";
- }
-};
-
-class SetAttrsOp : public TestOp {
-public:
- string oid;
- librados::ObjectWriteOperation op;
- librados::AioCompletion *comp;
- SetAttrsOp(int n,
- RadosTestContext *context,
- const string &oid,
- TestOpStat *stat)
- : TestOp(n, context, stat),
- oid(oid), comp(NULL)
- {}
-
- void _begin() override
- {
- ContDesc cont;
- {
- Mutex::Locker l(context->state_lock);
- cont = ContDesc(context->seq_num, context->current_snap,
- context->seq_num, "");
- context->oid_in_use.insert(oid);
- context->oid_not_in_use.erase(oid);
- }
-
- map<string, bufferlist> omap_contents;
- map<string, ContDesc> omap;
- bufferlist header;
- ContentsGenerator::iterator keygen = context->attr_gen.get_iterator(cont);
- op.create(false);
- while (!*keygen) ++keygen;
- while (*keygen) {
- if (*keygen != '_')
- header.append(*keygen);
- ++keygen;
- }
- for (int i = 0; i < 20; ++i) {
- string key;
- while (!*keygen) ++keygen;
- while (*keygen && key.size() < 40) {
- key.push_back((*keygen % 20) + 'a');
- ++keygen;
- }
- ContDesc val(cont);
- val.seqnum += (unsigned)(*keygen);
- val.prefix = ("oid: " + oid);
- omap[key] = val;
- bufferlist val_buffer = context->attr_gen.gen_bl(val);
- omap_contents[key] = val_buffer;
- op.setxattr(key.c_str(), val_buffer);
- }
- if (!context->no_omap) {
- op.omap_set_header(header);
- op.omap_set(omap_contents);
- }
-
- {
- Mutex::Locker l(context->state_lock);
- context->update_object_header(oid, header);
- context->update_object_attrs(oid, omap);
- }
-
- pair<TestOp*, TestOp::CallbackInfo*> *cb_arg =
- new pair<TestOp*, TestOp::CallbackInfo*>(this,
- new TestOp::CallbackInfo(0));
- comp = context->rados.aio_create_completion((void*) cb_arg, NULL,
- &write_callback);
- context->io_ctx.aio_operate(context->prefix+oid, comp, &op);
- }
-
- void _finish(CallbackInfo *info) override
- {
- Mutex::Locker l(context->state_lock);
- int r;
- if ((r = comp->get_return_value())) {
- cerr << "err " << r << std::endl;
- ceph_abort();
- }
- done = true;
- context->update_object_version(oid, comp->get_version64());
- context->oid_in_use.erase(oid);
- context->oid_not_in_use.insert(oid);
- context->kick();
- }
-
- bool finished() override
- {
- return done;
- }
-
- string getType() override
- {
- return "SetAttrsOp";
- }
-};
-
-class WriteOp : public TestOp {
-public:
- string oid;
- ContDesc cont;
- set<librados::AioCompletion *> waiting;
- librados::AioCompletion *rcompletion;
- uint64_t waiting_on;
- uint64_t last_acked_tid;
-
- librados::ObjectReadOperation read_op;
- librados::ObjectWriteOperation write_op;
- bufferlist rbuffer;
-
- bool do_append;
- bool do_excl;
-
- WriteOp(int n,
- RadosTestContext *context,
- const string &oid,
- bool do_append,
- bool do_excl,
- TestOpStat *stat = 0)
- : TestOp(n, context, stat),
- oid(oid), rcompletion(NULL), waiting_on(0),
- last_acked_tid(0), do_append(do_append),
- do_excl(do_excl)
- {}
-
- void _begin() override
- {
- context->state_lock.Lock();
- done = 0;
- stringstream acc;
- acc << context->prefix << "OID: " << oid << " snap " << context->current_snap << std::endl;
- string prefix = acc.str();
-
- cont = ContDesc(context->seq_num, context->current_snap, context->seq_num, prefix);
-
- ContentsGenerator *cont_gen;
- if (do_append) {
- ObjectDesc old_value;
- bool found = context->find_object(oid, &old_value);
- uint64_t prev_length = found && old_value.has_contents() ?
- old_value.most_recent_gen()->get_length(old_value.most_recent()) :
- 0;
- bool requires;
- int r = context->io_ctx.pool_requires_alignment2(&requires);
- assert(r == 0);
- uint64_t alignment = 0;
- if (requires) {
- r = context->io_ctx.pool_required_alignment2(&alignment);
- assert(r == 0);
- assert(alignment != 0);
- }
- cont_gen = new AppendGenerator(
- prev_length,
- alignment,
- context->min_stride_size,
- context->max_stride_size,
- 3);
- } else {
- cont_gen = new VarLenGenerator(
- context->max_size, context->min_stride_size, context->max_stride_size);
- }
- context->update_object(cont_gen, oid, cont);
-
- context->oid_in_use.insert(oid);
- context->oid_not_in_use.erase(oid);
-
- map<uint64_t, uint64_t> ranges;
-
- cont_gen->get_ranges_map(cont, ranges);
- std::cout << num << ": seq_num " << context->seq_num << " ranges " << ranges << std::endl;
- context->seq_num++;
-
- waiting_on = ranges.size();
- ContentsGenerator::iterator gen_pos = cont_gen->get_iterator(cont);
- uint64_t tid = 1;
- for (map<uint64_t, uint64_t>::iterator i = ranges.begin();
- i != ranges.end();
- ++i, ++tid) {
- gen_pos.seek(i->first);
- bufferlist to_write = gen_pos.gen_bl_advance(i->second);
- assert(to_write.length() == i->second);
- assert(to_write.length() > 0);
- std::cout << num << ": writing " << context->prefix+oid
- << " from " << i->first
- << " to " << i->first + i->second << " tid " << tid << std::endl;
- pair<TestOp*, TestOp::CallbackInfo*> *cb_arg =
- new pair<TestOp*, TestOp::CallbackInfo*>(this,
- new TestOp::CallbackInfo(tid));
- librados::AioCompletion *completion =
- context->rados.aio_create_completion((void*) cb_arg, NULL,
- &write_callback);
- waiting.insert(completion);
- librados::ObjectWriteOperation op;
- if (do_append) {
- op.append(to_write);
- } else {
- op.write(i->first, to_write);
- }
- if (do_excl && tid == 1)
- op.assert_exists();
- context->io_ctx.aio_operate(
- context->prefix+oid, completion,
- &op);
- }
-
- bufferlist contbl;
- ::encode(cont, contbl);
- pair<TestOp*, TestOp::CallbackInfo*> *cb_arg =
- new pair<TestOp*, TestOp::CallbackInfo*>(
- this,
- new TestOp::CallbackInfo(++tid));
- librados::AioCompletion *completion = context->rados.aio_create_completion(
- (void*) cb_arg, NULL, &write_callback);
- waiting.insert(completion);
- waiting_on++;
- write_op.setxattr("_header", contbl);
- if (!do_append) {
- write_op.truncate(cont_gen->get_length(cont));
- }
- context->io_ctx.aio_operate(
- context->prefix+oid, completion, &write_op);
-
- cb_arg =
- new pair<TestOp*, TestOp::CallbackInfo*>(
- this,
- new TestOp::CallbackInfo(++tid));
- rcompletion = context->rados.aio_create_completion(
- (void*) cb_arg, NULL, &write_callback);
- waiting_on++;
- read_op.read(0, 1, &rbuffer, 0);
- context->io_ctx.aio_operate(
- context->prefix+oid, rcompletion,
- &read_op,
- librados::OPERATION_ORDER_READS_WRITES, // order wrt previous write/update
- 0);
- context->state_lock.Unlock();
- }
-
- void _finish(CallbackInfo *info) override
- {
- assert(info);
- context->state_lock.Lock();
- uint64_t tid = info->id;
-
- cout << num << ": finishing write tid " << tid << " to " << context->prefix + oid << std::endl;
-
- if (tid <= last_acked_tid) {
- cerr << "Error: finished tid " << tid
- << " when last_acked_tid was " << last_acked_tid << std::endl;
- ceph_abort();
- }
- last_acked_tid = tid;
-
- assert(!done);
- waiting_on--;
- if (waiting_on == 0) {
- uint64_t version = 0;
- for (set<librados::AioCompletion *>::iterator i = waiting.begin();
- i != waiting.end();
- ) {
- assert((*i)->is_complete());
- if (int err = (*i)->get_return_value()) {
- cerr << "Error: oid " << oid << " write returned error code "
- << err << std::endl;
- }
- if ((*i)->get_version64() > version)
- version = (*i)->get_version64();
- (*i)->release();
- waiting.erase(i++);
- }
-
- context->update_object_version(oid, version);
- if (rcompletion->get_version64() != version) {
- cerr << "Error: racing read on " << oid << " returned version "
- << rcompletion->get_version64() << " rather than version "
- << version << std::endl;
- assert(0 == "racing read got wrong version");
- }
-
- {
- ObjectDesc old_value;
- assert(context->find_object(oid, &old_value, -1));
- if (old_value.deleted())
- std::cout << num << ": left oid " << oid << " deleted" << std::endl;
- else
- std::cout << num << ": left oid " << oid << " "
- << old_value.most_recent() << std::endl;
- }
-
- rcompletion->release();
- context->oid_in_use.erase(oid);
- context->oid_not_in_use.insert(oid);
- context->kick();
- done = true;
- }
- context->state_lock.Unlock();
- }
-
- bool finished() override
- {
- return done;
- }
-
- string getType() override
- {
- return "WriteOp";
- }
-};
-
-class WriteSameOp : public TestOp {
-public:
- string oid;
- ContDesc cont;
- set<librados::AioCompletion *> waiting;
- librados::AioCompletion *rcompletion;
- uint64_t waiting_on;
- uint64_t last_acked_tid;
-
- librados::ObjectReadOperation read_op;
- librados::ObjectWriteOperation write_op;
- bufferlist rbuffer;
-
- WriteSameOp(int n,
- RadosTestContext *context,
- const string &oid,
- TestOpStat *stat = 0)
- : TestOp(n, context, stat),
- oid(oid), rcompletion(NULL), waiting_on(0),
- last_acked_tid(0)
- {}
-
- void _begin() override
- {
- context->state_lock.Lock();
- done = 0;
- stringstream acc;
- acc << context->prefix << "OID: " << oid << " snap " << context->current_snap << std::endl;
- string prefix = acc.str();
-
- cont = ContDesc(context->seq_num, context->current_snap, context->seq_num, prefix);
-
- ContentsGenerator *cont_gen;
- cont_gen = new VarLenGenerator(
- context->max_size, context->min_stride_size, context->max_stride_size);
- context->update_object(cont_gen, oid, cont);
-
- context->oid_in_use.insert(oid);
- context->oid_not_in_use.erase(oid);
-
- map<uint64_t, uint64_t> ranges;
-
- cont_gen->get_ranges_map(cont, ranges);
- std::cout << num << ": seq_num " << context->seq_num << " ranges " << ranges << std::endl;
- context->seq_num++;
-
- waiting_on = ranges.size();
- ContentsGenerator::iterator gen_pos = cont_gen->get_iterator(cont);
- uint64_t tid = 1;
- for (map<uint64_t, uint64_t>::iterator i = ranges.begin();
- i != ranges.end();
- ++i, ++tid) {
- gen_pos.seek(i->first);
- bufferlist to_write = gen_pos.gen_bl_advance(i->second);
- assert(to_write.length() == i->second);
- assert(to_write.length() > 0);
- std::cout << num << ": writing " << context->prefix+oid
- << " from " << i->first
- << " to " << i->first + i->second << " tid " << tid << std::endl;
- pair<TestOp*, TestOp::CallbackInfo*> *cb_arg =
- new pair<TestOp*, TestOp::CallbackInfo*>(this,
- new TestOp::CallbackInfo(tid));
- librados::AioCompletion *completion =
- context->rados.aio_create_completion((void*) cb_arg, NULL,
- &write_callback);
- waiting.insert(completion);
- librados::ObjectWriteOperation op;
- /* no writesame multiplication factor for now */
- op.writesame(i->first, to_write.length(), to_write);
-
- context->io_ctx.aio_operate(
- context->prefix+oid, completion,
- &op);
- }
-
- bufferlist contbl;
- ::encode(cont, contbl);
- pair<TestOp*, TestOp::CallbackInfo*> *cb_arg =
- new pair<TestOp*, TestOp::CallbackInfo*>(
- this,
- new TestOp::CallbackInfo(++tid));
- librados::AioCompletion *completion = context->rados.aio_create_completion(
- (void*) cb_arg, NULL, &write_callback);
- waiting.insert(completion);
- waiting_on++;
- write_op.setxattr("_header", contbl);
- write_op.truncate(cont_gen->get_length(cont));
- context->io_ctx.aio_operate(
- context->prefix+oid, completion, &write_op);
-
- cb_arg =
- new pair<TestOp*, TestOp::CallbackInfo*>(
- this,
- new TestOp::CallbackInfo(++tid));
- rcompletion = context->rados.aio_create_completion(
- (void*) cb_arg, NULL, &write_callback);
- waiting_on++;
- read_op.read(0, 1, &rbuffer, 0);
- context->io_ctx.aio_operate(
- context->prefix+oid, rcompletion,
- &read_op,
- librados::OPERATION_ORDER_READS_WRITES, // order wrt previous write/update
- 0);
- context->state_lock.Unlock();
- }
-
- void _finish(CallbackInfo *info) override
- {
- assert(info);
- context->state_lock.Lock();
- uint64_t tid = info->id;
-
- cout << num << ": finishing writesame tid " << tid << " to " << context->prefix + oid << std::endl;
-
- if (tid <= last_acked_tid) {
- cerr << "Error: finished tid " << tid
- << " when last_acked_tid was " << last_acked_tid << std::endl;
- ceph_abort();
- }
- last_acked_tid = tid;
-
- assert(!done);
- waiting_on--;
- if (waiting_on == 0) {
- uint64_t version = 0;
- for (set<librados::AioCompletion *>::iterator i = waiting.begin();
- i != waiting.end();
- ) {
- assert((*i)->is_complete());
- if (int err = (*i)->get_return_value()) {
- cerr << "Error: oid " << oid << " writesame returned error code "
- << err << std::endl;
- }
- if ((*i)->get_version64() > version)
- version = (*i)->get_version64();
- (*i)->release();
- waiting.erase(i++);
- }
-
- context->update_object_version(oid, version);
- if (rcompletion->get_version64() != version) {
- cerr << "Error: racing read on " << oid << " returned version "
- << rcompletion->get_version64() << " rather than version "
- << version << std::endl;
- assert(0 == "racing read got wrong version");
- }
-
- {
- ObjectDesc old_value;
- assert(context->find_object(oid, &old_value, -1));
- if (old_value.deleted())
- std::cout << num << ": left oid " << oid << " deleted" << std::endl;
- else
- std::cout << num << ": left oid " << oid << " "
- << old_value.most_recent() << std::endl;
- }
-
- rcompletion->release();
- context->oid_in_use.erase(oid);
- context->oid_not_in_use.insert(oid);
- context->kick();
- done = true;
- }
- context->state_lock.Unlock();
- }
-
- bool finished() override
- {
- return done;
- }
-
- string getType() override
- {
- return "WriteSameOp";
- }
-};
-
-class DeleteOp : public TestOp {
-public:
- string oid;
-
- DeleteOp(int n,
- RadosTestContext *context,
- const string &oid,
- TestOpStat *stat = 0)
- : TestOp(n, context, stat), oid(oid)
- {}
-
- void _begin() override
- {
- context->state_lock.Lock();
- if (context->get_watch_context(oid)) {
- context->kick();
- context->state_lock.Unlock();
- return;
- }
-
- ObjectDesc contents;
- context->find_object(oid, &contents);
- bool present = !contents.deleted();
-
- context->oid_in_use.insert(oid);
- context->oid_not_in_use.erase(oid);
- context->seq_num++;
-
- context->remove_object(oid);
-
- interval_set<uint64_t> ranges;
- context->state_lock.Unlock();
-
- int r = 0;
- if (rand() % 2) {
- librados::ObjectWriteOperation op;
- op.assert_exists();
- op.remove();
- r = context->io_ctx.operate(context->prefix+oid, &op);
- } else {
- r = context->io_ctx.remove(context->prefix+oid);
- }
- if (r && !(r == -ENOENT && !present)) {
- cerr << "r is " << r << " while deleting " << oid << " and present is " << present << std::endl;
- ceph_abort();
- }
-
- context->state_lock.Lock();
- context->oid_in_use.erase(oid);
- context->oid_not_in_use.insert(oid);
- context->kick();
- context->state_lock.Unlock();
- }
-
- string getType() override
- {
- return "DeleteOp";
- }
-};
-
-class ReadOp : public TestOp {
-public:
- vector<librados::AioCompletion *> completions;
- librados::ObjectReadOperation op;
- string oid;
- ObjectDesc old_value;
- int snap;
- bool balance_reads;
-
- ceph::shared_ptr<int> in_use;
-
- vector<bufferlist> results;
- vector<int> retvals;
- vector<std::map<uint64_t, uint64_t>> extent_results;
- vector<bool> is_sparse_read;
- uint64_t waiting_on;
-
- vector<bufferlist> checksums;
- vector<int> checksum_retvals;
-
- map<string, bufferlist> attrs;
- int attrretval;
-
- set<string> omap_requested_keys;
- map<string, bufferlist> omap_returned_values;
- set<string> omap_keys;
- map<string, bufferlist> omap;
- bufferlist header;
-
- map<string, bufferlist> xattrs;
- ReadOp(int n,
- RadosTestContext *context,
- const string &oid,
- bool balance_reads,
- TestOpStat *stat = 0)
- : TestOp(n, context, stat),
- completions(3),
- oid(oid),
- snap(0),
- balance_reads(balance_reads),
- results(3),
- retvals(3),
- extent_results(3),
- is_sparse_read(3, false),
- waiting_on(0),
- checksums(3),
- checksum_retvals(3),
- attrretval(0)
- {}
-
- void _do_read(librados::ObjectReadOperation& read_op, int index) {
- uint64_t len = 0;
- if (old_value.has_contents())
- len = old_value.most_recent_gen()->get_length(old_value.most_recent());
- if (context->no_sparse || rand() % 2) {
- is_sparse_read[index] = false;
- read_op.read(0,
- len,
- &results[index],
- &retvals[index]);
- bufferlist init_value_bl;
- ::encode(static_cast<uint32_t>(-1), init_value_bl);
- read_op.checksum(LIBRADOS_CHECKSUM_TYPE_CRC32C, init_value_bl, 0, len,
- 0, &checksums[index], &checksum_retvals[index]);
- } else {
- is_sparse_read[index] = true;
- read_op.sparse_read(0,
- len,
- &extent_results[index],
- &results[index],
- &retvals[index]);
- }
- }
-
- void _begin() override
- {
- context->state_lock.Lock();
- if (!(rand() % 4) && !context->snaps.empty()) {
- snap = rand_choose(context->snaps)->first;
- in_use = context->snaps_in_use.lookup_or_create(snap, snap);
- } else {
- snap = -1;
- }
- std::cout << num << ": read oid " << oid << " snap " << snap << std::endl;
- done = 0;
- for (uint32_t i = 0; i < 3; i++) {
- completions[i] = context->rados.aio_create_completion((void *) this, &read_callback, 0);
- }
-
- context->oid_in_use.insert(oid);
- context->oid_not_in_use.erase(oid);
- assert(context->find_object(oid, &old_value, snap));
- if (old_value.deleted())
- std::cout << num << ": expect deleted" << std::endl;
- else
- std::cout << num << ": expect " << old_value.most_recent() << std::endl;
-
- TestWatchContext *ctx = context->get_watch_context(oid);
- context->state_lock.Unlock();
- if (ctx) {
- assert(old_value.exists);
- TestAlarm alarm;
- std::cerr << num << ": about to start" << std::endl;
- ctx->start();
- std::cerr << num << ": started" << std::endl;
- bufferlist bl;
- context->io_ctx.set_notify_timeout(600);
- int r = context->io_ctx.notify2(context->prefix+oid, bl, 0, NULL);
- if (r < 0) {
- std::cerr << "r is " << r << std::endl;
- ceph_abort();
- }
- std::cerr << num << ": notified, waiting" << std::endl;
- ctx->wait();
- }
- context->state_lock.Lock();
- if (snap >= 0) {
- context->io_ctx.snap_set_read(context->snaps[snap]);
- }
- _do_read(op, 0);
- for (map<string, ContDesc>::iterator i = old_value.attrs.begin();
- i != old_value.attrs.end();
- ++i) {
- if (rand() % 2) {
- string key = i->first;
- if (rand() % 2)
- key.push_back((rand() % 26) + 'a');
- omap_requested_keys.insert(key);
- }
- }
- if (!context->no_omap) {
- op.omap_get_vals_by_keys(omap_requested_keys, &omap_returned_values, 0);
- // NOTE: we're ignore pmore here, which assumes the OSD limit is high
- // enough for us.
- op.omap_get_keys2("", -1, &omap_keys, nullptr, nullptr);
- op.omap_get_vals2("", -1, &omap, nullptr, nullptr);
- op.omap_get_header(&header, 0);
- }
- op.getxattrs(&xattrs, 0);
-
- unsigned flags = 0;
- if (balance_reads)
- flags |= librados::OPERATION_BALANCE_READS;
-
- assert(!context->io_ctx.aio_operate(context->prefix+oid, completions[0], &op,
- flags, NULL));
- waiting_on++;
-
- // send 2 pipelined reads on the same object/snap. This can help testing
- // OSD's read behavior in some scenarios
- for (uint32_t i = 1; i < 3; ++i) {
- librados::ObjectReadOperation pipeline_op;
- _do_read(pipeline_op, i);
- assert(!context->io_ctx.aio_operate(context->prefix+oid, completions[i], &pipeline_op, 0));
- waiting_on++;
- }
-
- if (snap >= 0) {
- context->io_ctx.snap_set_read(0);
- }
- context->state_lock.Unlock();
- }
-
- void _finish(CallbackInfo *info) override
- {
- Mutex::Locker l(context->state_lock);
- assert(!done);
- assert(waiting_on > 0);
- if (--waiting_on) {
- return;
- }
-
- context->oid_in_use.erase(oid);
- context->oid_not_in_use.insert(oid);
- int retval = completions[0]->get_return_value();
- for (vector<librados::AioCompletion *>::iterator it = completions.begin();
- it != completions.end(); ++it) {
- assert((*it)->is_complete());
- uint64_t version = (*it)->get_version64();
- int err = (*it)->get_return_value();
- if (err != retval) {
- cerr << num << ": Error: oid " << oid << " read returned different error codes: "
- << retval << " and " << err << std::endl;
- ceph_abort();
- }
- if (err) {
- if (!(err == -ENOENT && old_value.deleted())) {
- cerr << num << ": Error: oid " << oid << " read returned error code "
- << err << std::endl;
- ceph_abort();
- }
- } else if (version != old_value.version) {
- cerr << num << ": oid " << oid << " version is " << version
- << " and expected " << old_value.version << std::endl;
- assert(version == old_value.version);
- }
- }
- if (!retval) {
- map<string, bufferlist>::iterator iter = xattrs.find("_header");
- bufferlist headerbl;
- if (iter == xattrs.end()) {
- if (old_value.has_contents()) {
- cerr << num << ": Error: did not find header attr, has_contents: "
- << old_value.has_contents()
- << std::endl;
- assert(!old_value.has_contents());
- }
- } else {
- headerbl = iter->second;
- xattrs.erase(iter);
- }
- if (old_value.deleted()) {
- std::cout << num << ": expect deleted" << std::endl;
- assert(0 == "expected deleted");
- } else {
- std::cout << num << ": expect " << old_value.most_recent() << std::endl;
- }
- if (old_value.has_contents()) {
- ContDesc to_check;
- bufferlist::iterator p = headerbl.begin();
- ::decode(to_check, p);
- if (to_check != old_value.most_recent()) {
- cerr << num << ": oid " << oid << " found incorrect object contents " << to_check
- << ", expected " << old_value.most_recent() << std::endl;
- context->errors++;
- }
- for (unsigned i = 0; i < results.size(); i++) {
- if (is_sparse_read[i]) {
- if (!old_value.check_sparse(extent_results[i], results[i])) {
- cerr << num << ": oid " << oid << " contents " << to_check << " corrupt" << std::endl;
- context->errors++;
- }
- } else {
- if (!old_value.check(results[i])) {
- cerr << num << ": oid " << oid << " contents " << to_check << " corrupt" << std::endl;
- context->errors++;
- }
-
- uint32_t checksum = 0;
- if (checksum_retvals[i] == 0) {
- try {
- auto bl_it = checksums[i].begin();
- uint32_t csum_count;
- ::decode(csum_count, bl_it);
- ::decode(checksum, bl_it);
- } catch (const buffer::error &err) {
- checksum_retvals[i] = -EBADMSG;
- }
- }
- if (checksum_retvals[i] != 0 || checksum != results[i].crc32c(-1)) {
- cerr << num << ": oid " << oid << " checksum " << checksums[i]
- << " incorrect, expecting " << results[i].crc32c(-1)
- << std::endl;
- context->errors++;
- }
- }
- }
- if (context->errors) ceph_abort();
- }
-
- // Attributes
- if (!context->no_omap) {
- if (!(old_value.header == header)) {
- cerr << num << ": oid " << oid << " header does not match, old size: "
- << old_value.header.length() << " new size " << header.length()
- << std::endl;
- assert(old_value.header == header);
- }
- if (omap.size() != old_value.attrs.size()) {
- cerr << num << ": oid " << oid << " omap.size() is " << omap.size()
- << " and old is " << old_value.attrs.size() << std::endl;
- assert(omap.size() == old_value.attrs.size());
- }
- if (omap_keys.size() != old_value.attrs.size()) {
- cerr << num << ": oid " << oid << " omap.size() is " << omap_keys.size()
- << " and old is " << old_value.attrs.size() << std::endl;
- assert(omap_keys.size() == old_value.attrs.size());
- }
- }
- if (xattrs.size() != old_value.attrs.size()) {
- cerr << num << ": oid " << oid << " xattrs.size() is " << xattrs.size()
- << " and old is " << old_value.attrs.size() << std::endl;
- assert(xattrs.size() == old_value.attrs.size());
- }
- for (map<string, ContDesc>::iterator iter = old_value.attrs.begin();
- iter != old_value.attrs.end();
- ++iter) {
- bufferlist bl = context->attr_gen.gen_bl(
- iter->second);
- if (!context->no_omap) {
- map<string, bufferlist>::iterator omap_iter = omap.find(iter->first);
- assert(omap_iter != omap.end());
- assert(bl.length() == omap_iter->second.length());
- bufferlist::iterator k = bl.begin();
- for(bufferlist::iterator l = omap_iter->second.begin();
- !k.end() && !l.end();
- ++k, ++l) {
- assert(*l == *k);
- }
- }
- map<string, bufferlist>::iterator xattr_iter = xattrs.find(iter->first);
- assert(xattr_iter != xattrs.end());
- assert(bl.length() == xattr_iter->second.length());
- bufferlist::iterator k = bl.begin();
- for (bufferlist::iterator j = xattr_iter->second.begin();
- !k.end() && !j.end();
- ++j, ++k) {
- assert(*j == *k);
- }
- }
- if (!context->no_omap) {
- for (set<string>::iterator i = omap_requested_keys.begin();
- i != omap_requested_keys.end();
- ++i) {
- if (!omap_returned_values.count(*i))
- assert(!old_value.attrs.count(*i));
- if (!old_value.attrs.count(*i))
- assert(!omap_returned_values.count(*i));
- }
- for (map<string, bufferlist>::iterator i = omap_returned_values.begin();
- i != omap_returned_values.end();
- ++i) {
- assert(omap_requested_keys.count(i->first));
- assert(omap.count(i->first));
- assert(old_value.attrs.count(i->first));
- assert(i->second == omap[i->first]);
- }
- }
- }
- for (vector<librados::AioCompletion *>::iterator it = completions.begin();
- it != completions.end(); ++it) {
- (*it)->release();
- }
- context->kick();
- done = true;
- }
-
- bool finished() override
- {
- return done;
- }
-
- string getType() override
- {
- return "ReadOp";
- }
-};
-
-class SnapCreateOp : public TestOp {
-public:
- SnapCreateOp(int n,
- RadosTestContext *context,
- TestOpStat *stat = 0)
- : TestOp(n, context, stat)
- {}
-
- void _begin() override
- {
- uint64_t snap;
- string snapname;
-
- if (context->pool_snaps) {
- stringstream ss;
-
- ss << context->prefix << "snap" << ++context->snapname_num;
- snapname = ss.str();
-
- int ret = context->io_ctx.snap_create(snapname.c_str());
- if (ret) {
- cerr << "snap_create returned " << ret << std::endl;
- ceph_abort();
- }
- assert(!context->io_ctx.snap_lookup(snapname.c_str(), &snap));
-
- } else {
- assert(!context->io_ctx.selfmanaged_snap_create(&snap));
- }
-
- context->state_lock.Lock();
- context->add_snap(snap);
-
- if (context->pool_snaps) {
- context->state_lock.Unlock();
- } else {
- vector<uint64_t> snapset(context->snaps.size());
-
- int j = 0;
- for (map<int,uint64_t>::reverse_iterator i = context->snaps.rbegin();
- i != context->snaps.rend();
- ++i, ++j) {
- snapset[j] = i->second;
- }
-
- context->state_lock.Unlock();
-
- int r = context->io_ctx.selfmanaged_snap_set_write_ctx(context->seq, snapset);
- if (r) {
- cerr << "r is " << r << " snapset is " << snapset << " seq is " << context->seq << std::endl;
- ceph_abort();
- }
- }
- }
-
- string getType() override
- {
- return "SnapCreateOp";
- }
- bool must_quiesce_other_ops() override { return context->pool_snaps; }
-};
-
-class SnapRemoveOp : public TestOp {
-public:
- int to_remove;
- SnapRemoveOp(int n, RadosTestContext *context,
- int snap,
- TestOpStat *stat = 0)
- : TestOp(n, context, stat),
- to_remove(snap)
- {}
-
- void _begin() override
- {
- context->state_lock.Lock();
- uint64_t snap = context->snaps[to_remove];
- context->remove_snap(to_remove);
-
- if (context->pool_snaps) {
- string snapname;
-
- assert(!context->io_ctx.snap_get_name(snap, &snapname));
- assert(!context->io_ctx.snap_remove(snapname.c_str()));
- } else {
- assert(!context->io_ctx.selfmanaged_snap_remove(snap));
-
- vector<uint64_t> snapset(context->snaps.size());
- int j = 0;
- for (map<int,uint64_t>::reverse_iterator i = context->snaps.rbegin();
- i != context->snaps.rend();
- ++i, ++j) {
- snapset[j] = i->second;
- }
-
- int r = context->io_ctx.selfmanaged_snap_set_write_ctx(context->seq, snapset);
- if (r) {
- cerr << "r is " << r << " snapset is " << snapset << " seq is " << context->seq << std::endl;
- ceph_abort();
- }
- }
- context->state_lock.Unlock();
- }
-
- string getType() override
- {
- return "SnapRemoveOp";
- }
-};
-
-class WatchOp : public TestOp {
- string oid;
-public:
- WatchOp(int n,
- RadosTestContext *context,
- const string &_oid,
- TestOpStat *stat = 0)
- : TestOp(n, context, stat),
- oid(_oid)
- {}
-
- void _begin() override
- {
- context->state_lock.Lock();
- ObjectDesc contents;
- context->find_object(oid, &contents);
- if (contents.deleted()) {
- context->kick();
- context->state_lock.Unlock();
- return;
- }
- context->oid_in_use.insert(oid);
- context->oid_not_in_use.erase(oid);
-
- TestWatchContext *ctx = context->get_watch_context(oid);
- context->state_lock.Unlock();
- int r;
- if (!ctx) {
- {
- Mutex::Locker l(context->state_lock);
- ctx = context->watch(oid);
- }
-
- r = context->io_ctx.watch2(context->prefix+oid,
- &ctx->get_handle(),
- ctx);
- } else {
- r = context->io_ctx.unwatch2(ctx->get_handle());
- {
- Mutex::Locker l(context->state_lock);
- context->unwatch(oid);
- }
- }
-
- if (r) {
- cerr << "r is " << r << std::endl;
- ceph_abort();
- }
-
- {
- Mutex::Locker l(context->state_lock);
- context->oid_in_use.erase(oid);
- context->oid_not_in_use.insert(oid);
- }
- }
-
- string getType() override
- {
- return "WatchOp";
- }
-};
-
-class RollbackOp : public TestOp {
-public:
- string oid;
- int roll_back_to;
- librados::ObjectWriteOperation zero_write_op1;
- librados::ObjectWriteOperation zero_write_op2;
- librados::ObjectWriteOperation op;
- vector<librados::AioCompletion *> comps;
- ceph::shared_ptr<int> in_use;
- int last_finished;
- int outstanding;
-
- RollbackOp(int n,
- RadosTestContext *context,
- const string &_oid,
- TestOpStat *stat = 0)
- : TestOp(n, context, stat),
- oid(_oid), roll_back_to(-1),
- comps(3, NULL),
- last_finished(-1), outstanding(3)
- {}
-
- void _begin() override
- {
- context->state_lock.Lock();
- if (context->get_watch_context(oid)) {
- context->kick();
- context->state_lock.Unlock();
- return;
- }
-
- if (context->snaps.empty()) {
- context->kick();
- context->state_lock.Unlock();
- done = true;
- return;
- }
-
- context->oid_in_use.insert(oid);
- context->oid_not_in_use.erase(oid);
-
- roll_back_to = rand_choose(context->snaps)->first;
- in_use = context->snaps_in_use.lookup_or_create(
- roll_back_to,
- roll_back_to);
-
-
- cout << "rollback oid " << oid << " to " << roll_back_to << std::endl;
-
- bool existed_before = context->object_existed_at(oid);
- bool existed_after = context->object_existed_at(oid, roll_back_to);
-
- context->roll_back(oid, roll_back_to);
- uint64_t snap = context->snaps[roll_back_to];
-
- outstanding -= (!existed_before) + (!existed_after);
-
- context->state_lock.Unlock();
-
- bufferlist bl, bl2;
- zero_write_op1.append(bl);
- zero_write_op2.append(bl2);
-
- if (context->pool_snaps) {
- op.snap_rollback(snap);
- } else {
- op.selfmanaged_snap_rollback(snap);
- }
-
- if (existed_before) {
- pair<TestOp*, TestOp::CallbackInfo*> *cb_arg =
- new pair<TestOp*, TestOp::CallbackInfo*>(this,
- new TestOp::CallbackInfo(0));
- comps[0] =
- context->rados.aio_create_completion((void*) cb_arg, NULL,
- &write_callback);
- context->io_ctx.aio_operate(
- context->prefix+oid, comps[0], &zero_write_op1);
- }
- {
- pair<TestOp*, TestOp::CallbackInfo*> *cb_arg =
- new pair<TestOp*, TestOp::CallbackInfo*>(this,
- new TestOp::CallbackInfo(1));
- comps[1] =
- context->rados.aio_create_completion((void*) cb_arg, NULL,
- &write_callback);
- context->io_ctx.aio_operate(
- context->prefix+oid, comps[1], &op);
- }
- if (existed_after) {
- pair<TestOp*, TestOp::CallbackInfo*> *cb_arg =
- new pair<TestOp*, TestOp::CallbackInfo*>(this,
- new TestOp::CallbackInfo(2));
- comps[2] =
- context->rados.aio_create_completion((void*) cb_arg, NULL,
- &write_callback);
- context->io_ctx.aio_operate(
- context->prefix+oid, comps[2], &zero_write_op2);
- }
- }
-
- void _finish(CallbackInfo *info) override
- {
- Mutex::Locker l(context->state_lock);
- uint64_t tid = info->id;
- cout << num << ": finishing rollback tid " << tid
- << " to " << context->prefix + oid << std::endl;
- assert((int)(info->id) > last_finished);
- last_finished = info->id;
-
- int r;
- if ((r = comps[last_finished]->get_return_value()) != 0) {
- cerr << "err " << r << std::endl;
- ceph_abort();
- }
- if (--outstanding == 0) {
- done = true;
- context->update_object_version(oid, comps[tid]->get_version64());
- context->oid_in_use.erase(oid);
- context->oid_not_in_use.insert(oid);
- in_use = ceph::shared_ptr<int>();
- context->kick();
- }
- }
-
- bool finished() override
- {
- return done;
- }
-
- string getType() override
- {
- return "RollBackOp";
- }
-};
-
-class CopyFromOp : public TestOp {
-public:
- string oid, oid_src;
- ObjectDesc src_value;
- librados::ObjectWriteOperation op;
- librados::ObjectReadOperation rd_op;
- librados::AioCompletion *comp;
- librados::AioCompletion *comp_racing_read;
- ceph::shared_ptr<int> in_use;
- int snap;
- int done;
- uint64_t version;
- int r;
- CopyFromOp(int n,
- RadosTestContext *context,
- const string &oid,
- const string &oid_src,
- TestOpStat *stat)
- : TestOp(n, context, stat),
- oid(oid), oid_src(oid_src),
- comp(NULL), snap(-1), done(0),
- version(0), r(0)
- {}
-
- void _begin() override
- {
- ContDesc cont;
- {
- Mutex::Locker l(context->state_lock);
- cont = ContDesc(context->seq_num, context->current_snap,
- context->seq_num, "");
- context->oid_in_use.insert(oid);
- context->oid_not_in_use.erase(oid);
- context->oid_in_use.insert(oid_src);
- context->oid_not_in_use.erase(oid_src);
-
- // choose source snap
- if (0 && !(rand() % 4) && !context->snaps.empty()) {
- snap = rand_choose(context->snaps)->first;
- in_use = context->snaps_in_use.lookup_or_create(snap, snap);
- } else {
- snap = -1;
- }
- context->find_object(oid_src, &src_value, snap);
- if (!src_value.deleted())
- context->update_object_full(oid, src_value);
- }
-
- string src = context->prefix+oid_src;
- op.copy_from(src.c_str(), context->io_ctx, src_value.version);
-
- pair<TestOp*, TestOp::CallbackInfo*> *cb_arg =
- new pair<TestOp*, TestOp::CallbackInfo*>(this,
- new TestOp::CallbackInfo(0));
- comp = context->rados.aio_create_completion((void*) cb_arg, NULL,
- &write_callback);
- context->io_ctx.aio_operate(context->prefix+oid, comp, &op);
-
- // queue up a racing read, too.
- pair<TestOp*, TestOp::CallbackInfo*> *read_cb_arg =
- new pair<TestOp*, TestOp::CallbackInfo*>(this,
- new TestOp::CallbackInfo(1));
- comp_racing_read = context->rados.aio_create_completion((void*) read_cb_arg, NULL, &write_callback);
- rd_op.stat(NULL, NULL, NULL);
- context->io_ctx.aio_operate(context->prefix+oid, comp_racing_read, &rd_op,
- librados::OPERATION_ORDER_READS_WRITES, // order wrt previous write/update
- NULL);
-
- }
-
- void _finish(CallbackInfo *info) override
- {
- Mutex::Locker l(context->state_lock);
-
- // note that the read can (and atm will) come back before the
- // write reply, but will reflect the update and the versions will
- // match.
-
- if (info->id == 0) {
- // copy_from
- assert(comp->is_complete());
- cout << num << ": finishing copy_from to " << context->prefix + oid << std::endl;
- if ((r = comp->get_return_value())) {
- if (r == -ENOENT && src_value.deleted()) {
- cout << num << ": got expected ENOENT (src dne)" << std::endl;
- } else {
- cerr << "Error: oid " << oid << " copy_from " << oid_src << " returned error code "
- << r << std::endl;
- ceph_abort();
- }
- } else {
- assert(!version || comp->get_version64() == version);
- version = comp->get_version64();
- context->update_object_version(oid, comp->get_version64());
- }
- } else if (info->id == 1) {
- // racing read
- assert(comp_racing_read->is_complete());
- cout << num << ": finishing copy_from racing read to " << context->prefix + oid << std::endl;
- if ((r = comp_racing_read->get_return_value())) {
- if (!(r == -ENOENT && src_value.deleted())) {
- cerr << "Error: oid " << oid << " copy_from " << oid_src << " returned error code "
- << r << std::endl;
- }
- } else {
- assert(comp_racing_read->get_return_value() == 0);
- assert(!version || comp_racing_read->get_version64() == version);
- version = comp_racing_read->get_version64();
- }
- }
- if (++done == 2) {
- context->oid_in_use.erase(oid);
- context->oid_not_in_use.insert(oid);
- context->oid_in_use.erase(oid_src);
- context->oid_not_in_use.insert(oid_src);
- context->kick();
- }
- }
-
- bool finished() override
- {
- return done == 2;
- }
-
- string getType() override
- {
- return "CopyFromOp";
- }
-};
-
-class SetRedirectOp : public TestOp {
-public:
- string oid, oid_tgt, tgt_pool_name;
- ObjectDesc src_value, tgt_value;
- librados::ObjectWriteOperation op;
- librados::ObjectReadOperation rd_op;
- librados::AioCompletion *comp;
- ceph::shared_ptr<int> in_use;
- int done;
- int r;
- SetRedirectOp(int n,
- RadosTestContext *context,
- const string &oid,
- const string &oid_tgt,
- const string &tgt_pool_name,
- TestOpStat *stat = 0)
- : TestOp(n, context, stat),
- oid(oid), oid_tgt(oid_tgt), tgt_pool_name(tgt_pool_name),
- comp(NULL), done(0),
- r(0)
- {}
-
- void _begin() override
- {
- Mutex::Locker l(context->state_lock);
- context->oid_in_use.insert(oid);
- context->oid_not_in_use.erase(oid);
- context->oid_redirect_in_use.insert(oid_tgt);
- context->oid_redirect_not_in_use.erase(oid_tgt);
-
- context->find_object(oid, &src_value);
- if(!context->redirect_objs[oid].empty()) {
- /* update target's user_version */
- rd_op.stat(NULL, NULL, NULL);
- comp = context->rados.aio_create_completion();
- context->io_ctx.aio_operate(context->prefix+oid_tgt, comp, &rd_op,
- librados::OPERATION_ORDER_READS_WRITES,
- NULL);
- comp->wait_for_safe();
- context->update_object_version(oid_tgt, comp->get_version64());
- comp->release();
-
- /* unset redirect target */
- comp = context->rados.aio_create_completion();
- bool present = !src_value.deleted();
- context->remove_object(oid);
- op.remove();
- context->io_ctx.aio_operate(context->prefix+oid, comp, &op,
- librados::OPERATION_ORDER_READS_WRITES |
- librados::OPERATION_IGNORE_REDIRECT);
- comp->wait_for_safe();
- if ((r = comp->get_return_value())) {
- if (!(r == -ENOENT && !present)) {
- cerr << "r is " << r << " while deleting " << oid << " and present is " << present << std::endl;
- ceph_abort();
- }
- }
- comp->release();
-
- context->oid_redirect_not_in_use.insert(context->redirect_objs[oid]);
- context->oid_redirect_in_use.erase(context->redirect_objs[oid]);
-
- /* copy_from oid_tgt --> oid */
- comp = context->rados.aio_create_completion();
- context->find_object(oid_tgt, &tgt_value);
- string src = context->prefix+oid_tgt;
- op.copy_from(src.c_str(), context->io_ctx, tgt_value.version);
- context->io_ctx.aio_operate(context->prefix+oid, comp, &op,
- librados::OPERATION_ORDER_READS_WRITES);
- comp->wait_for_safe();
- if ((r = comp->get_return_value())) {
- cerr << "Error: oid " << oid << " copy_from " << oid_tgt << " returned error code "
- << r << std::endl;
- ceph_abort();
- }
- context->update_object_full(oid, tgt_value);
- context->update_object_version(oid, comp->get_version64());
- comp->release();
- }
-
- comp = context->rados.aio_create_completion();
- rd_op.stat(NULL, NULL, NULL);
- context->io_ctx.aio_operate(context->prefix+oid, comp, &rd_op,
- librados::OPERATION_ORDER_READS_WRITES |
- librados::OPERATION_IGNORE_REDIRECT,
- NULL);
- comp->wait_for_safe();
- if ((r = comp->get_return_value()) && !src_value.deleted()) {
- cerr << "Error: oid " << oid << " stat returned error code "
- << r << std::endl;
- ceph_abort();
- }
- context->update_object_version(oid, comp->get_version64());
- comp->release();
-
- context->find_object(oid, &src_value);
- context->find_object(oid_tgt, &tgt_value);
-
- if (!src_value.deleted() && !tgt_value.deleted())
- context->update_object_full(oid, tgt_value);
-
- if (src_value.version != 0 && !src_value.deleted())
- op.assert_version(src_value.version);
- op.set_redirect(context->prefix+oid_tgt, context->io_ctx, tgt_value.version);
-
- pair<TestOp*, TestOp::CallbackInfo*> *cb_arg =
- new pair<TestOp*, TestOp::CallbackInfo*>(this,
- new TestOp::CallbackInfo(0));
- comp = context->rados.aio_create_completion((void*) cb_arg, NULL,
- &write_callback);
- context->io_ctx.aio_operate(context->prefix+oid, comp, &op,
- librados::OPERATION_ORDER_READS_WRITES);
- }
-
- void _finish(CallbackInfo *info) override
- {
- Mutex::Locker l(context->state_lock);
-
- if (info->id == 0) {
- assert(comp->is_complete());
- cout << num << ": finishing set_redirect to oid " << oid << std::endl;
- if ((r = comp->get_return_value())) {
- if (r == -ENOENT && src_value.deleted()) {
- cout << num << ": got expected ENOENT (src dne)" << std::endl;
- } else {
- cerr << "Error: oid " << oid << " set_redirect " << oid_tgt << " returned error code "
- << r << std::endl;
- ceph_abort();
- }
- } else {
- context->update_object_redirect_target(oid, oid_tgt);
- context->update_object_version(oid, comp->get_version64());
- }
- }
-
- if (++done == 1) {
- context->oid_in_use.erase(oid);
- context->oid_not_in_use.insert(oid);
- context->kick();
- }
- }
-
- bool finished() override
- {
- return done == 1;
- }
-
- string getType() override
- {
- return "SetRedirectOp";
- }
-};
-
-class UnsetRedirectOp : public TestOp {
-public:
- string oid;
- librados::ObjectWriteOperation op;
- librados::AioCompletion *completion;
- librados::AioCompletion *comp;
-
- UnsetRedirectOp(int n,
- RadosTestContext *context,
- const string &oid,
- TestOpStat *stat = 0)
- : TestOp(n, context, stat), oid(oid)
- {}
-
- void _begin() override
- {
- context->state_lock.Lock();
- if (context->get_watch_context(oid)) {
- context->kick();
- context->state_lock.Unlock();
- return;
- }
-
- ObjectDesc contents;
- context->find_object(oid, &contents);
- bool present = !contents.deleted();
-
- context->oid_in_use.insert(oid);
- context->oid_not_in_use.erase(oid);
- context->seq_num++;
-
- context->remove_object(oid);
-
- context->state_lock.Unlock();
-
- comp = context->rados.aio_create_completion();
- op.remove();
- context->io_ctx.aio_operate(context->prefix+oid, comp, &op,
- librados::OPERATION_ORDER_READS_WRITES |
- librados::OPERATION_IGNORE_REDIRECT);
- comp->wait_for_safe();
- int r = comp->get_return_value();
- if (r && !(r == -ENOENT && !present)) {
- cerr << "r is " << r << " while deleting " << oid << " and present is " << present << std::endl;
- ceph_abort();
- }
-
- context->state_lock.Lock();
- context->oid_in_use.erase(oid);
- context->oid_not_in_use.insert(oid);
- if(!context->redirect_objs[oid].empty()) {
- context->oid_redirect_not_in_use.insert(context->redirect_objs[oid]);
- context->oid_redirect_in_use.erase(context->redirect_objs[oid]);
- context->update_object_redirect_target(oid, string());
- }
- context->kick();
- context->state_lock.Unlock();
- }
-
- string getType() override
- {
- return "UnsetRedirectOp";
- }
-};
-
-class HitSetListOp : public TestOp {
- librados::AioCompletion *comp1, *comp2;
- uint32_t hash;
- std::list< std::pair<time_t, time_t> > ls;
- bufferlist bl;
-
-public:
- HitSetListOp(int n,
- RadosTestContext *context,
- uint32_t hash,
- TestOpStat *stat = 0)
- : TestOp(n, context, stat),
- comp1(NULL), comp2(NULL),
- hash(hash)
- {}
-
- void _begin() override
- {
- pair<TestOp*, TestOp::CallbackInfo*> *cb_arg =
- new pair<TestOp*, TestOp::CallbackInfo*>(this,
- new TestOp::CallbackInfo(0));
- comp1 = context->rados.aio_create_completion((void*) cb_arg, NULL,
- &write_callback);
- int r = context->io_ctx.hit_set_list(hash, comp1, &ls);
- assert(r == 0);
- }
-
- void _finish(CallbackInfo *info) override {
- Mutex::Locker l(context->state_lock);
- if (!comp2) {
- if (ls.empty()) {
- cerr << num << ": no hitsets" << std::endl;
- done = true;
- } else {
- cerr << num << ": hitsets are " << ls << std::endl;
- int r = rand() % ls.size();
- std::list<pair<time_t,time_t> >::iterator p = ls.begin();
- while (r--)
- ++p;
- pair<TestOp*, TestOp::CallbackInfo*> *cb_arg =
- new pair<TestOp*, TestOp::CallbackInfo*>(this,
- new TestOp::CallbackInfo(0));
- comp2 = context->rados.aio_create_completion((void*) cb_arg, NULL,
- &write_callback);
- r = context->io_ctx.hit_set_get(hash, comp2, p->second, &bl);
- assert(r == 0);
- }
- } else {
- int r = comp2->get_return_value();
- if (r == 0) {
- HitSet hitset;
- bufferlist::iterator p = bl.begin();
- ::decode(hitset, p);
- cout << num << ": got hitset of type " << hitset.get_type_name()
- << " size " << bl.length()
- << std::endl;
- } else {
- // FIXME: we could verify that we did in fact race with a trim...
- assert(r == -ENOENT);
- }
- done = true;
- }
-
- context->kick();
- }
-
- bool finished() override {
- return done;
- }
-
- string getType() override {
- return "HitSetListOp";
- }
-};
-
-class UndirtyOp : public TestOp {
-public:
- librados::AioCompletion *completion;
- librados::ObjectWriteOperation op;
- string oid;
-
- UndirtyOp(int n,
- RadosTestContext *context,
- const string &oid,
- TestOpStat *stat = 0)
- : TestOp(n, context, stat),
- completion(NULL),
- oid(oid)
- {}
-
- void _begin() override
- {
- context->state_lock.Lock();
- pair<TestOp*, TestOp::CallbackInfo*> *cb_arg =
- new pair<TestOp*, TestOp::CallbackInfo*>(this,
- new TestOp::CallbackInfo(0));
- completion = context->rados.aio_create_completion((void *) cb_arg, NULL,
- &write_callback);
-
- context->oid_in_use.insert(oid);
- context->oid_not_in_use.erase(oid);
- context->update_object_undirty(oid);
- context->state_lock.Unlock();
-
- op.undirty();
- int r = context->io_ctx.aio_operate(context->prefix+oid, completion,
- &op, 0);
- assert(!r);
- }
-
- void _finish(CallbackInfo *info) override
- {
- context->state_lock.Lock();
- assert(!done);
- assert(completion->is_complete());
- context->oid_in_use.erase(oid);
- context->oid_not_in_use.insert(oid);
- context->update_object_version(oid, completion->get_version64());
- context->kick();
- done = true;
- context->state_lock.Unlock();
- }
-
- bool finished() override
- {
- return done;
- }
-
- string getType() override
- {
- return "UndirtyOp";
- }
-};
-
-class IsDirtyOp : public TestOp {
-public:
- librados::AioCompletion *completion;
- librados::ObjectReadOperation op;
- string oid;
- bool dirty;
- ObjectDesc old_value;
- int snap;
- ceph::shared_ptr<int> in_use;
-
- IsDirtyOp(int n,
- RadosTestContext *context,
- const string &oid,
- TestOpStat *stat = 0)
- : TestOp(n, context, stat),
- completion(NULL),
- oid(oid),
- dirty(false)
- {}
-
- void _begin() override
- {
- context->state_lock.Lock();
-
- if (!(rand() % 4) && !context->snaps.empty()) {
- snap = rand_choose(context->snaps)->first;
- in_use = context->snaps_in_use.lookup_or_create(snap, snap);
- } else {
- snap = -1;
- }
- std::cout << num << ": is_dirty oid " << oid << " snap " << snap
- << std::endl;
-
- pair<TestOp*, TestOp::CallbackInfo*> *cb_arg =
- new pair<TestOp*, TestOp::CallbackInfo*>(this,
- new TestOp::CallbackInfo(0));
- completion = context->rados.aio_create_completion((void *) cb_arg, NULL,
- &write_callback);
-
- context->oid_in_use.insert(oid);
- context->oid_not_in_use.erase(oid);
- context->state_lock.Unlock();
-
- if (snap >= 0) {
- context->io_ctx.snap_set_read(context->snaps[snap]);
- }
-
- op.is_dirty(&dirty, NULL);
- int r = context->io_ctx.aio_operate(context->prefix+oid, completion,
- &op, 0);
- assert(!r);
-
- if (snap >= 0) {
- context->io_ctx.snap_set_read(0);
- }
- }
-
- void _finish(CallbackInfo *info) override
- {
- context->state_lock.Lock();
- assert(!done);
- assert(completion->is_complete());
- context->oid_in_use.erase(oid);
- context->oid_not_in_use.insert(oid);
-
- assert(context->find_object(oid, &old_value, snap));
-
- int r = completion->get_return_value();
- if (r == 0) {
- cout << num << ": " << (dirty ? "dirty" : "clean") << std::endl;
- assert(!old_value.deleted());
- assert(dirty == old_value.dirty);
- } else {
- cout << num << ": got " << r << std::endl;
- assert(r == -ENOENT);
- assert(old_value.deleted());
- }
- context->kick();
- done = true;
- context->state_lock.Unlock();
- }
-
- bool finished() override
- {
- return done;
- }
-
- string getType() override
- {
- return "IsDirtyOp";
- }
-};
-
-
-
-class CacheFlushOp : public TestOp {
-public:
- librados::AioCompletion *completion;
- librados::ObjectReadOperation op;
- string oid;
- bool blocking;
- int snap;
- bool can_fail;
- ceph::shared_ptr<int> in_use;
-
- CacheFlushOp(int n,
- RadosTestContext *context,
- const string &oid,
- TestOpStat *stat,
- bool b)
- : TestOp(n, context, stat),
- completion(NULL),
- oid(oid),
- blocking(b),
- snap(0),
- can_fail(false)
- {}
-
- void _begin() override
- {
- context->state_lock.Lock();
-
- if (!(rand() % 4) && !context->snaps.empty()) {
- snap = rand_choose(context->snaps)->first;
- in_use = context->snaps_in_use.lookup_or_create(snap, snap);
- } else {
- snap = -1;
- }
- // not being particularly specific here about knowing which
- // flushes are on the oldest clean snap and which ones are not.
- can_fail = !blocking || !context->snaps.empty();
- // FIXME: we could fail if we've ever removed a snap due to
- // the async snap trimming.
- can_fail = true;
- cout << num << ": " << (blocking ? "cache_flush" : "cache_try_flush")
- << " oid " << oid << " snap " << snap << std::endl;
-
- if (snap >= 0) {
- context->io_ctx.snap_set_read(context->snaps[snap]);
- }
-
- pair<TestOp*, TestOp::CallbackInfo*> *cb_arg =
- new pair<TestOp*, TestOp::CallbackInfo*>(this,
- new TestOp::CallbackInfo(0));
- completion = context->rados.aio_create_completion((void *) cb_arg, NULL,
- &write_callback);
- context->oid_flushing.insert(oid);
- context->oid_not_flushing.erase(oid);
- context->state_lock.Unlock();
-
- unsigned flags = librados::OPERATION_IGNORE_CACHE;
- if (blocking) {
- op.cache_flush();
- } else {
- op.cache_try_flush();
- flags = librados::OPERATION_SKIPRWLOCKS;
- }
- int r = context->io_ctx.aio_operate(context->prefix+oid, completion,
- &op, flags, NULL);
- assert(!r);
-
- if (snap >= 0) {
- context->io_ctx.snap_set_read(0);
- }
- }
-
- void _finish(CallbackInfo *info) override
- {
- context->state_lock.Lock();
- assert(!done);
- assert(completion->is_complete());
- context->oid_flushing.erase(oid);
- context->oid_not_flushing.insert(oid);
- int r = completion->get_return_value();
- cout << num << ": got " << cpp_strerror(r) << std::endl;
- if (r == 0) {
- context->update_object_version(oid, 0, snap);
- } else if (r == -EBUSY) {
- assert(can_fail);
- } else if (r == -EINVAL) {
- // caching not enabled?
- } else if (r == -ENOENT) {
- // may have raced with a remove?
- } else {
- assert(0 == "shouldn't happen");
- }
- context->kick();
- done = true;
- context->state_lock.Unlock();
- }
-
- bool finished() override
- {
- return done;
- }
-
- string getType() override
- {
- return "CacheFlushOp";
- }
-};
-
-class CacheEvictOp : public TestOp {
-public:
- librados::AioCompletion *completion;
- librados::ObjectReadOperation op;
- string oid;
- ceph::shared_ptr<int> in_use;
-
- CacheEvictOp(int n,
- RadosTestContext *context,
- const string &oid,
- TestOpStat *stat)
- : TestOp(n, context, stat),
- completion(NULL),
- oid(oid)
- {}
-
- void _begin() override
- {
- context->state_lock.Lock();
-
- int snap;
- if (!(rand() % 4) && !context->snaps.empty()) {
- snap = rand_choose(context->snaps)->first;
- in_use = context->snaps_in_use.lookup_or_create(snap, snap);
- } else {
- snap = -1;
- }
- cout << num << ": cache_evict oid " << oid << " snap " << snap << std::endl;
-
- if (snap >= 0) {
- context->io_ctx.snap_set_read(context->snaps[snap]);
- }
-
- pair<TestOp*, TestOp::CallbackInfo*> *cb_arg =
- new pair<TestOp*, TestOp::CallbackInfo*>(this,
- new TestOp::CallbackInfo(0));
- completion = context->rados.aio_create_completion((void *) cb_arg, NULL,
- &write_callback);
- context->state_lock.Unlock();
-
- op.cache_evict();
- int r = context->io_ctx.aio_operate(context->prefix+oid, completion,
- &op, librados::OPERATION_IGNORE_CACHE,
- NULL);
- assert(!r);
-
- if (snap >= 0) {
- context->io_ctx.snap_set_read(0);
- }
- }
-
- void _finish(CallbackInfo *info) override
- {
- context->state_lock.Lock();
- assert(!done);
- assert(completion->is_complete());
-
- int r = completion->get_return_value();
- cout << num << ": got " << cpp_strerror(r) << std::endl;
- if (r == 0) {
- // yay!
- } else if (r == -EBUSY) {
- // raced with something that dirtied the object
- } else if (r == -EINVAL) {
- // caching not enabled?
- } else if (r == -ENOENT) {
- // may have raced with a remove?
- } else {
- assert(0 == "shouldn't happen");
- }
- context->kick();
- done = true;
- context->state_lock.Unlock();
- }
-
- bool finished() override
- {
- return done;
- }
-
- string getType() override
- {
- return "CacheEvictOp";
- }
-};
-
-
-#endif
Code Review / stor4nfv.git / blobdiff