--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+#include "FileStoreTracker.h"
+#include <stdlib.h>
+#include <iostream>
+#include <boost/scoped_ptr.hpp>
+#include "include/Context.h"
+#include "common/Mutex.h"
+
+class OnApplied : public Context {
+ FileStoreTracker *tracker;
+ list<pair<pair<coll_t, string>, uint64_t> > in_flight;
+public:
+ OnApplied(FileStoreTracker *tracker,
+ list<pair<pair<coll_t, string>, uint64_t> > in_flight)
+ : tracker(tracker), in_flight(in_flight) {}
+
+ void finish(int r) override {
+ for (list<pair<pair<coll_t, string>, uint64_t> >::iterator i =
+ in_flight.begin();
+ i != in_flight.end();
+ ++i) {
+ tracker->applied(i->first, i->second);
+ }
+ }
+};
+
+class OnCommitted : public Context {
+ FileStoreTracker *tracker;
+ list<pair<pair<coll_t, string>, uint64_t> > in_flight;
+public:
+ OnCommitted(FileStoreTracker *tracker,
+ list<pair<pair<coll_t, string>, uint64_t> > in_flight)
+ : tracker(tracker), in_flight(in_flight) {}
+
+ void finish(int r) override {
+ for (list<pair<pair<coll_t, string>, uint64_t> >::iterator i =
+ in_flight.begin();
+ i != in_flight.end();
+ ++i) {
+ tracker->committed(i->first, i->second);
+ }
+ }
+};
+
+int FileStoreTracker::init()
+{
+ set<string> to_get;
+ to_get.insert("STATUS");
+ map<string, bufferlist> got;
+ db->get("STATUS", to_get, &got);
+ restart_seq = 0;
+ if (!got.empty()) {
+ bufferlist::iterator bp = got.begin()->second.begin();
+ ::decode(restart_seq, bp);
+ }
+ ++restart_seq;
+ KeyValueDB::Transaction t = db->get_transaction();
+ got.clear();
+ ::encode(restart_seq, got["STATUS"]);
+ t->set("STATUS", got);
+ db->submit_transaction(t);
+ return 0;
+}
+
+void FileStoreTracker::submit_transaction(Transaction &t)
+{
+ list<pair<pair<coll_t, string>, uint64_t> > in_flight;
+ OutTransaction out;
+ out.t = new ObjectStore::Transaction;
+ out.in_flight = &in_flight;
+ for (list<Transaction::Op*>::iterator i = t.ops.begin();
+ i != t.ops.end();
+ ++i) {
+ (**i)(this, &out);
+ }
+ store->queue_transaction(
+ 0, std::move(*out.t),
+ new OnApplied(this, in_flight),
+ new OnCommitted(this, in_flight));
+ delete out.t;
+}
+
+void FileStoreTracker::write(const pair<coll_t, string> &obj,
+ OutTransaction *out)
+{
+ Mutex::Locker l(lock);
+ std::cerr << "Writing " << obj << std::endl;
+ ObjectContents contents = get_current_content(obj);
+
+ uint64_t offset = rand() % (SIZE/2);
+ uint64_t len = rand() % (SIZE/2);
+ if (!len) len = 10;
+ contents.write(rand(), offset, len);
+
+ bufferlist to_write;
+ ObjectContents::Iterator iter = contents.get_iterator();
+ iter.seek_to(offset);
+ for (uint64_t i = offset;
+ i < offset + len;
+ ++i, ++iter) {
+ assert(iter.valid());
+ to_write.append(*iter);
+ }
+ out->t->write(coll_t(obj.first),
+ ghobject_t(hobject_t(sobject_t(obj.second, CEPH_NOSNAP))),
+ offset,
+ len,
+ to_write);
+ out->in_flight->push_back(make_pair(obj, set_content(obj, contents)));
+}
+
+void FileStoreTracker::remove(const pair<coll_t, string> &obj,
+ OutTransaction *out)
+{
+ std::cerr << "Deleting " << obj << std::endl;
+ Mutex::Locker l(lock);
+ ObjectContents old_contents = get_current_content(obj);
+ if (!old_contents.exists())
+ return;
+ out->t->remove(coll_t(obj.first),
+ ghobject_t(hobject_t(sobject_t(obj.second, CEPH_NOSNAP))));
+ ObjectContents contents;
+ out->in_flight->push_back(make_pair(obj, set_content(obj, contents)));
+}
+
+void FileStoreTracker::clone_range(const pair<coll_t, string> &from,
+ const pair<coll_t, string> &to,
+ OutTransaction *out) {
+ Mutex::Locker l(lock);
+ std::cerr << "CloningRange " << from << " to " << to << std::endl;
+ assert(from.first == to.first);
+ ObjectContents from_contents = get_current_content(from);
+ ObjectContents to_contents = get_current_content(to);
+ if (!from_contents.exists()) {
+ return;
+ }
+ if (from.second == to.second) {
+ return;
+ }
+
+ uint64_t new_size = from_contents.size();
+ interval_set<uint64_t> interval_to_clone;
+ uint64_t offset = rand() % (new_size/2);
+ uint64_t len = rand() % (new_size/2);
+ if (!len) len = 10;
+ interval_to_clone.insert(offset, len);
+ to_contents.clone_range(from_contents, interval_to_clone);
+ out->t->clone_range(coll_t(from.first),
+ ghobject_t(hobject_t(sobject_t(from.second, CEPH_NOSNAP))),
+ ghobject_t(hobject_t(sobject_t(to.second, CEPH_NOSNAP))),
+ offset,
+ len,
+ offset);
+ out->in_flight->push_back(make_pair(to, set_content(to, to_contents)));
+}
+
+void FileStoreTracker::clone(const pair<coll_t, string> &from,
+ const pair<coll_t, string> &to,
+ OutTransaction *out) {
+ Mutex::Locker l(lock);
+ std::cerr << "Cloning " << from << " to " << to << std::endl;
+ assert(from.first == to.first);
+ if (from.second == to.second) {
+ return;
+ }
+ ObjectContents from_contents = get_current_content(from);
+ ObjectContents to_contents = get_current_content(to);
+ if (!from_contents.exists()) {
+ return;
+ }
+
+ if (to_contents.exists())
+ out->t->remove(coll_t(to.first),
+ ghobject_t(hobject_t(sobject_t(to.second, CEPH_NOSNAP))));
+ out->t->clone(coll_t(from.first),
+ ghobject_t(hobject_t(sobject_t(from.second, CEPH_NOSNAP))),
+ ghobject_t(hobject_t(sobject_t(to.second, CEPH_NOSNAP))));
+ out->in_flight->push_back(make_pair(to, set_content(to, from_contents)));
+}
+
+
+string obj_to_prefix(const pair<coll_t, string> &obj) {
+ string sep;
+ sep.push_back('^');
+ return obj.first.to_str() + sep + obj.second + "_CONTENTS_";
+}
+
+string obj_to_meta_prefix(const pair<coll_t, string> &obj) {
+ string sep;
+ sep.push_back('^');
+ return obj.first.to_str() + sep + obj.second;
+}
+
+string seq_to_key(uint64_t seq) {
+ char buf[50];
+ snprintf(buf, sizeof(buf), "%*llu", 20, (unsigned long long int)seq);
+ return string(buf);
+}
+
+struct ObjStatus {
+ uint64_t last_applied;
+ uint64_t last_committed;
+ uint64_t restart_seq;
+ ObjStatus() : last_applied(0), last_committed(0), restart_seq(0) {}
+
+ uint64_t get_last_applied(uint64_t seq) const {
+ if (seq > restart_seq)
+ return last_committed;
+ else
+ return last_applied;
+ }
+ void set_last_applied(uint64_t _last_applied, uint64_t seq) {
+ last_applied = _last_applied;
+ restart_seq = seq;
+ }
+ uint64_t trim_to() const {
+ return last_applied < last_committed ?
+ last_applied : last_committed;
+ }
+};
+void encode(const ObjStatus &obj, bufferlist &bl) {
+ ::encode(obj.last_applied, bl);
+ ::encode(obj.last_committed, bl);
+ ::encode(obj.restart_seq, bl);
+}
+void decode(ObjStatus &obj, bufferlist::iterator &bl) {
+ ::decode(obj.last_applied, bl);
+ ::decode(obj.last_committed, bl);
+ ::decode(obj.restart_seq, bl);
+}
+
+
+ObjStatus get_obj_status(const pair<coll_t, string> &obj,
+ KeyValueDB *db)
+{
+ set<string> to_get;
+ to_get.insert("META");
+ map<string, bufferlist> got;
+ db->get(obj_to_meta_prefix(obj), to_get, &got);
+ ObjStatus retval;
+ if (!got.empty()) {
+ bufferlist::iterator bp = got.begin()->second.begin();
+ ::decode(retval, bp);
+ }
+ return retval;
+}
+
+void set_obj_status(const pair<coll_t, string> &obj,
+ const ObjStatus &status,
+ KeyValueDB::Transaction t)
+{
+ map<string, bufferlist> to_set;
+ ::encode(status, to_set["META"]);
+ t->set(obj_to_meta_prefix(obj), to_set);
+}
+
+void _clean_forward(const pair<coll_t, string> &obj,
+ uint64_t last_valid,
+ KeyValueDB *db)
+{
+ KeyValueDB::Transaction t = db->get_transaction();
+ KeyValueDB::Iterator i = db->get_iterator(obj_to_prefix(obj));
+ set<string> to_remove;
+ i->upper_bound(seq_to_key(last_valid));
+ for (; i->valid(); i->next()) {
+ to_remove.insert(i->key());
+ }
+ t->rmkeys(obj_to_prefix(obj), to_remove);
+ db->submit_transaction(t);
+}
+
+
+void FileStoreTracker::verify(const coll_t &coll, const string &obj,
+ bool on_start) {
+ Mutex::Locker l(lock);
+ std::cerr << "Verifying " << make_pair(coll, obj) << std::endl;
+
+ pair<uint64_t, uint64_t> valid_reads = get_valid_reads(make_pair(coll, obj));
+ std::cerr << "valid_reads is " << valid_reads << std::endl;
+ bufferlist contents;
+ int r = store->read(coll_t(coll),
+ ghobject_t(hobject_t(sobject_t(obj, CEPH_NOSNAP))),
+ 0,
+ 2*SIZE,
+ contents);
+ std::cerr << "exists: " << r << std::endl;
+
+
+ for (uint64_t i = valid_reads.first;
+ i < valid_reads.second;
+ ++i) {
+ ObjectContents old_contents = get_content(make_pair(coll, obj), i);
+
+ std::cerr << "old_contents exists " << old_contents.exists() << std::endl;
+ if (!old_contents.exists() && (r == -ENOENT))
+ return;
+
+ if (old_contents.exists() && (r == -ENOENT))
+ continue;
+
+ if (!old_contents.exists() && (r != -ENOENT))
+ continue;
+
+ if (contents.length() != old_contents.size()) {
+ std::cerr << "old_contents.size() is "
+ << old_contents.size() << std::endl;
+ continue;
+ }
+
+ bufferlist::iterator bp = contents.begin();
+ ObjectContents::Iterator iter = old_contents.get_iterator();
+ iter.seek_to_first();
+ bool matches = true;
+ uint64_t pos = 0;
+ for (; !bp.end() && iter.valid();
+ ++iter, ++bp, ++pos) {
+ if (*iter != *bp) {
+ std::cerr << "does not match at pos " << pos << std::endl;
+ matches = false;
+ break;
+ }
+ }
+ if (matches) {
+ if (on_start)
+ _clean_forward(make_pair(coll, obj), i, db);
+ return;
+ }
+ }
+ std::cerr << "Verifying " << make_pair(coll, obj) << " failed " << std::endl;
+ ceph_abort();
+}
+
+ObjectContents FileStoreTracker::get_current_content(
+ const pair<coll_t, string> &obj)
+{
+ KeyValueDB::Iterator iter = db->get_iterator(
+ obj_to_prefix(obj));
+ iter->seek_to_last();
+ if (iter->valid()) {
+ bufferlist bl = iter->value();
+ bufferlist::iterator bp = bl.begin();
+ pair<uint64_t, bufferlist> val;
+ ::decode(val, bp);
+ assert(seq_to_key(val.first) == iter->key());
+ bp = val.second.begin();
+ return ObjectContents(bp);
+ }
+ return ObjectContents();
+}
+
+ObjectContents FileStoreTracker::get_content(
+ const pair<coll_t, string> &obj, uint64_t version)
+{
+ set<string> to_get;
+ map<string, bufferlist> got;
+ to_get.insert(seq_to_key(version));
+ db->get(obj_to_prefix(obj), to_get, &got);
+ if (got.empty())
+ return ObjectContents();
+ pair<uint64_t, bufferlist> val;
+ bufferlist::iterator bp = got.begin()->second.begin();
+ ::decode(val, bp);
+ bp = val.second.begin();
+ assert(val.first == version);
+ return ObjectContents(bp);
+}
+
+pair<uint64_t, uint64_t> FileStoreTracker::get_valid_reads(
+ const pair<coll_t, string> &obj)
+{
+ pair<uint64_t, uint64_t> bounds = make_pair(0,1);
+ KeyValueDB::Iterator iter = db->get_iterator(
+ obj_to_prefix(obj));
+ iter->seek_to_last();
+ if (iter->valid()) {
+ pair<uint64_t, bufferlist> val;
+ bufferlist bl = iter->value();
+ bufferlist::iterator bp = bl.begin();
+ ::decode(val, bp);
+ bounds.second = val.first + 1;
+ }
+
+ ObjStatus obj_status = get_obj_status(obj, db);
+ bounds.first = obj_status.get_last_applied(restart_seq);
+ return bounds;
+}
+
+void clear_obsolete(const pair<coll_t, string> &obj,
+ const ObjStatus &status,
+ KeyValueDB *db,
+ KeyValueDB::Transaction t)
+{
+ KeyValueDB::Iterator iter = db->get_iterator(obj_to_prefix(obj));
+ set<string> to_remove;
+ iter->seek_to_first();
+ for (; iter->valid() && iter->key() < seq_to_key(status.trim_to());
+ iter->next())
+ to_remove.insert(iter->key());
+ t->rmkeys(obj_to_prefix(obj), to_remove);
+}
+
+void FileStoreTracker::committed(const pair<coll_t, string> &obj,
+ uint64_t seq) {
+ Mutex::Locker l(lock);
+ ObjStatus status = get_obj_status(obj, db);
+ assert(status.last_committed < seq);
+ status.last_committed = seq;
+ KeyValueDB::Transaction t = db->get_transaction();
+ clear_obsolete(obj, status, db, t);
+ set_obj_status(obj, status, t);
+ db->submit_transaction(t);
+}
+
+void FileStoreTracker::applied(const pair<coll_t, string> &obj,
+ uint64_t seq) {
+ Mutex::Locker l(lock);
+ std::cerr << "Applied " << obj << " version " << seq << std::endl;
+ ObjStatus status = get_obj_status(obj, db);
+ assert(status.last_applied < seq);
+ status.set_last_applied(seq, restart_seq);
+ KeyValueDB::Transaction t = db->get_transaction();
+ clear_obsolete(obj, status, db, t);
+ set_obj_status(obj, status, t);
+ db->submit_transaction(t);
+}
+
+
+uint64_t FileStoreTracker::set_content(const pair<coll_t, string> &obj,
+ ObjectContents &content) {
+ KeyValueDB::Transaction t = db->get_transaction();
+ KeyValueDB::Iterator iter = db->get_iterator(
+ obj_to_prefix(obj));
+ iter->seek_to_last();
+ uint64_t most_recent = 0;
+ if (iter->valid()) {
+ pair<uint64_t, bufferlist> val;
+ bufferlist bl = iter->value();
+ bufferlist::iterator bp = bl.begin();
+ ::decode(val, bp);
+ most_recent = val.first;
+ }
+ bufferlist buf_content;
+ content.encode(buf_content);
+ map<string, bufferlist> to_set;
+ ::encode(make_pair(most_recent + 1, buf_content),
+ to_set[seq_to_key(most_recent + 1)]);
+ t->set(obj_to_prefix(obj), to_set);
+ db->submit_transaction(t);
+ return most_recent + 1;
+}
Code Review / stor4nfv.git / blobdiff