2 * OSD classes for the key value store
4 * Created on: Aug 10, 2012
5 * Author: Eleanor Cawthon
8 #include "objclass/objclass.h"
10 #include "key_value_store/kvs_arg_types.h"
11 #include "include/types.h"
17 * finds the index_data where a key belongs.
19 * @param key: the key to search for
20 * @param idata: the index_data for the first index value such that idata.key
21 * is greater than key.
22 * @param next_idata: the index_data for the next index entry after idata
23 * @pre: key is not encoded
24 * @post: idata contains complete information
27 static int get_idata_from_key(cls_method_context_t hctx, const string &key,
28 index_data &idata, index_data &next_idata) {
31 std::map<std::string, bufferlist> kvmap;
35 r = cls_cxx_map_get_vals(hctx, key_data(key).encoded(), "", 2, &kvmap, &more);
37 CLS_LOG(20, "error reading index for range %s: %d", key.c_str(), r);
41 r = cls_cxx_map_get_val(hctx, key_data(key).encoded(), &raw_val);
43 CLS_LOG(20, "%s is already in the index: %d", key.c_str(), r);
44 bufferlist::iterator b = raw_val.begin();
47 bufferlist::iterator b = kvmap.begin()->second.begin();
51 } else if (r == -ENOENT || r == -ENODATA) {
52 bufferlist::iterator b = kvmap.begin()->second.begin();
54 if (idata.kdata.prefix != "1") {
55 bufferlist::iterator nb = (++kvmap.begin())->second.begin();
56 next_idata.decode(nb);
60 CLS_LOG(20, "error reading index for duplicates %s: %d", key.c_str(), r);
64 CLS_LOG(20, "idata is %s", idata.str().c_str());
69 static int get_idata_from_key_op(cls_method_context_t hctx,
70 bufferlist *in, bufferlist *out) {
71 CLS_LOG(20, "get_idata_from_key_op");
72 idata_from_key_args op;
73 bufferlist::iterator it = in->begin();
76 } catch (buffer::error& err) {
77 CLS_LOG(20, "error decoding idata_from_key_args.");
80 int r = get_idata_from_key(hctx, op.key, op.idata, op.next_idata);
90 * finds the object in the index with the lowest key value that is greater
91 * than idata.key. If idata.key is the max key, returns -EOVERFLOW. If
92 * idata has a prefix and has timed out, cleans up.
94 * @param idata: idata for the object to search for.
95 * @param out_data: the idata for the next object.
97 * @pre: idata must contain a key.
98 * @post: out_data contains complete information
100 static int get_next_idata(cls_method_context_t hctx, const index_data &idata,
101 index_data &out_data) {
103 std::map<std::string, bufferlist> kvs;
105 r = cls_cxx_map_get_vals(hctx, idata.kdata.encoded(), "", 1, &kvs, &more);
107 CLS_LOG(20, "getting kvs failed with error %d", r);
112 out_data.kdata.parse(kvs.begin()->first);
113 bufferlist::iterator b = kvs.begin()->second.begin();
122 static int get_next_idata_op(cls_method_context_t hctx,
123 bufferlist *in, bufferlist *out) {
124 CLS_LOG(20, "get_next_idata_op");
125 idata_from_idata_args op;
126 bufferlist::iterator it = in->begin();
129 } catch (buffer::error& err) {
132 int r = get_next_idata(hctx, op.idata, op.next_idata);
142 * finds the object in the index with the highest key value that is less
143 * than idata.key. If idata.key is the lowest key, returns -ERANGE If
144 * idata has a prefix and has timed out, cleans up.
146 * @param idata: idata for the object to search for.
147 * @param out_data: the idata for the next object.
149 * @pre: idata must contain a key.
150 * @ost: out_data contains complete information
152 static int get_prev_idata(cls_method_context_t hctx, const index_data &idata,
153 index_data &out_data) {
155 std::map<std::string, bufferlist> kvs;
157 r = cls_cxx_map_get_vals(hctx, "", "", LONG_MAX, &kvs, &more);
159 CLS_LOG(20, "getting kvs failed with error %d", r);
163 std::map<std::string, bufferlist>::iterator it =
164 kvs.lower_bound(idata.kdata.encoded());
165 if (it->first != idata.kdata.encoded()) {
166 CLS_LOG(20, "object %s not found in the index (expected %s, found %s)",
167 idata.str().c_str(), idata.kdata.encoded().c_str(),
171 if (it == kvs.begin()) {
172 //it is the first object, there is no previous.
177 out_data.kdata.parse(it->first);
178 bufferlist::iterator b = it->second.begin();
184 static int get_prev_idata_op(cls_method_context_t hctx,
185 bufferlist *in, bufferlist *out) {
186 CLS_LOG(20, "get_next_idata_op");
187 idata_from_idata_args op;
188 bufferlist::iterator it = in->begin();
191 } catch (buffer::error& err) {
194 int r = get_prev_idata(hctx, op.idata, op.next_idata);
204 * Read all of the index entries where any keys in the map go
206 static int read_many(cls_method_context_t hctx, const set<string> &keys,
207 map<string, bufferlist> * out) {
210 CLS_ERR("reading from a map of size %d, first key encoded is %s",
211 (int)keys.size(), key_data(*keys.begin()).encoded().c_str());
212 r = cls_cxx_map_get_vals(hctx, key_data(*keys.begin()).encoded().c_str(),
213 "", LONG_MAX, out, &more);
215 CLS_ERR("getting omap vals failed with error %d", r);
218 CLS_ERR("got map of size %d ", (int)out->size());
219 if (out->size() > 1) {
220 out->erase(out->upper_bound(key_data(*keys.rbegin()).encoded().c_str()),
223 CLS_ERR("returning map of size %d", (int)out->size());
227 static int read_many_op(cls_method_context_t hctx, bufferlist *in,
229 CLS_LOG(20, "read_many_op");
231 map<string, bufferlist> outmap;
232 bufferlist::iterator it = in->begin();
235 } catch (buffer::error & err) {
238 int r = read_many(hctx, op, &outmap);
242 encode(outmap, *out);
248 * Checks the unwritable xattr. If it is "1" (i.e., it is unwritable), returns
249 * -EACCES. otherwise, returns 0.
251 static int check_writable(cls_method_context_t hctx) {
253 int r = cls_cxx_getxattr(hctx, "unwritable", &bl);
255 CLS_LOG(20, "error reading xattr %s: %d", "unwritable", r);
258 if (string(bl.c_str(), bl.length()) == "1") {
265 static int check_writable_op(cls_method_context_t hctx,
266 bufferlist *in, bufferlist *out) {
267 CLS_LOG(20, "check_writable_op");
268 return check_writable(hctx);
272 * returns -EKEYREJECTED if size is outside of bound, according to comparator.
274 * @bound: the limit to test
275 * @comparator: should be CEPH_OSD_CMPXATTR_OP_[EQ|GT|LT]
277 static int assert_size_in_bound(cls_method_context_t hctx, int bound,
281 int r = cls_cxx_getxattr(hctx, "size", &size_bl);
283 CLS_LOG(20, "error reading xattr %s: %d", "size", r);
287 int size = atoi(string(size_bl.c_str(), size_bl.length()).c_str());
288 CLS_LOG(20, "size is %d, bound is %d", size, bound);
290 //compare size to comparator
291 switch (comparator) {
292 case CEPH_OSD_CMPXATTR_OP_EQ:
294 return -EKEYREJECTED;
297 case CEPH_OSD_CMPXATTR_OP_LT:
299 return -EKEYREJECTED;
302 case CEPH_OSD_CMPXATTR_OP_GT:
304 return -EKEYREJECTED;
308 CLS_LOG(20, "invalid argument passed to assert_size_in_bound: %d",
315 static int assert_size_in_bound_op(cls_method_context_t hctx,
316 bufferlist *in, bufferlist *out) {
317 CLS_LOG(20, "assert_size_in_bound_op");
319 bufferlist::iterator it = in->begin();
322 } catch (buffer::error& err) {
325 return assert_size_in_bound(hctx, op.bound, op.comparator);
329 * Attempts to insert omap into this object's omap.
332 * if unwritable, returns -EACCES.
333 * if size > bound and key doesn't already exist in the omap, returns -EBALANCE.
334 * if exclusive is true, returns -EEXIST if any keys already exist.
336 * @post: object has omap entries inserted, and size xattr is updated
338 static int omap_insert(cls_method_context_t hctx,
339 const map<string, bufferlist> &omap, int bound, bool exclusive) {
343 int r = cls_cxx_stat(hctx, &size, &time);
347 CLS_LOG(20, "inserting %s", omap.begin()->first.c_str());
348 r = check_writable(hctx);
350 CLS_LOG(20, "omap_insert: this object is unwritable: %d", r);
354 int assert_bound = bound;
356 //if this is an exclusive insert, make sure the key doesn't already exist.
357 for (map<string, bufferlist>::const_iterator it = omap.begin();
358 it != omap.end(); ++it) {
360 r = cls_cxx_map_get_val(hctx, it->first, &bl);
361 if (r == 0 && string(bl.c_str(), bl.length()) != ""){
363 CLS_LOG(20, "error: this is an exclusive insert and %s exists.",
368 CLS_LOG(20, "increased assert_bound to %d", assert_bound);
369 } else if (r != -ENODATA && r != -ENOENT) {
370 CLS_LOG(20, "error reading omap val for %s: %d", it->first.c_str(), r);
376 r = cls_cxx_getxattr(hctx, "size", &old_size);
378 CLS_LOG(20, "error reading xattr %s: %d", "size", r);
382 int old_size_int = atoi(string(old_size.c_str(), old_size.length()).c_str());
384 CLS_LOG(20, "asserting size is less than %d (bound is %d)", assert_bound, bound);
385 if (old_size_int >= assert_bound) {
386 return -EKEYREJECTED;
389 int new_size_int = old_size_int + omap.size() - (assert_bound - bound);
390 CLS_LOG(20, "old size is %d, new size is %d", old_size_int, new_size_int);
394 new_size.append(s.str());
396 r = cls_cxx_map_set_vals(hctx, &omap);
398 CLS_LOG(20, "error setting omap: %d", r);
402 r = cls_cxx_setxattr(hctx, "size", &new_size);
404 CLS_LOG(20, "error setting xattr %s: %d", "size", r);
407 CLS_LOG(20, "successfully inserted %s", omap.begin()->first.c_str());
411 static int omap_insert_op(cls_method_context_t hctx,
412 bufferlist *in, bufferlist *out) {
413 CLS_LOG(20, "omap_insert");
415 bufferlist::iterator it = in->begin();
418 } catch (buffer::error& err) {
421 return omap_insert(hctx, op.omap, op.bound, op.exclusive);
424 static int create_with_omap(cls_method_context_t hctx,
425 const map<string, bufferlist> &omap) {
426 CLS_LOG(20, "creating with omap: %s", omap.begin()->first.c_str());
427 //first make sure the object is writable
428 int r = cls_cxx_create(hctx, true);
430 CLS_LOG(20, "omap create: creating failed: %d", r);
434 int new_size_int = omap.size();
435 CLS_LOG(20, "omap insert: new size is %d", new_size_int);
439 new_size.append(s.str());
441 r = cls_cxx_map_set_vals(hctx, &omap);
443 CLS_LOG(20, "omap create: error setting omap: %d", r);
447 r = cls_cxx_setxattr(hctx, "size", &new_size);
449 CLS_LOG(20, "omap create: error setting xattr %s: %d", "size", r);
455 r = cls_cxx_setxattr(hctx, "unwritable", &u);
457 CLS_LOG(20, "omap create: error setting xattr %s: %d", "unwritable", r);
461 CLS_LOG(20, "successfully created %s", omap.begin()->first.c_str());
465 static int create_with_omap_op(cls_method_context_t hctx,
466 bufferlist *in, bufferlist *out) {
467 CLS_LOG(20, "omap_insert");
468 map<string, bufferlist> omap;
469 bufferlist::iterator it = in->begin();
472 } catch (buffer::error& err) {
475 return create_with_omap(hctx, omap);
479 * Attempts to remove omap from this object's omap.
482 * if unwritable, returns -EACCES.
483 * if size < bound and key doesn't already exist in the omap, returns -EBALANCE.
484 * if any of the keys are not in this object, returns -ENODATA.
486 * @post: object has omap entries removed, and size xattr is updated
488 static int omap_remove(cls_method_context_t hctx,
489 const std::set<string> &omap, int bound) {
493 r = cls_cxx_stat(hctx, &size, &time);
498 //first make sure the object is writable
499 r = check_writable(hctx);
504 //check for existance of the key first
505 for (set<string>::const_iterator it = omap.begin();
506 it != omap.end(); ++it) {
508 r = cls_cxx_map_get_val(hctx, *it, &bl);
509 if (r == -ENOENT || r == -ENODATA
510 || string(bl.c_str(), bl.length()) == ""){
513 CLS_LOG(20, "error reading omap val for %s: %d", it->c_str(), r);
518 //fail if removing from an object with only bound entries.
520 r = cls_cxx_getxattr(hctx, "size", &old_size);
522 CLS_LOG(20, "error reading xattr %s: %d", "size", r);
525 int old_size_int = atoi(string(old_size.c_str(), old_size.length()).c_str());
527 CLS_LOG(20, "asserting size is greater than %d", bound);
528 if (old_size_int <= bound) {
529 return -EKEYREJECTED;
532 int new_size_int = old_size_int - omap.size();
533 CLS_LOG(20, "old size is %d, new size is %d", old_size_int, new_size_int);
537 new_size.append(s.str());
539 r = cls_cxx_setxattr(hctx, "size", &new_size);
541 CLS_LOG(20, "error setting xattr %s: %d", "unwritable", r);
545 for (std::set<string>::const_iterator it = omap.begin();
546 it != omap.end(); ++it) {
547 r = cls_cxx_map_remove_key(hctx, *it);
549 CLS_LOG(20, "error removing omap: %d", r);
556 static int omap_remove_op(cls_method_context_t hctx,
557 bufferlist *in, bufferlist *out) {
558 CLS_LOG(20, "omap_remove");
560 bufferlist::iterator it = in->begin();
563 } catch (buffer::error& err) {
566 return omap_remove(hctx, op.omap, op.bound);
570 * checks to see if this object needs to be split or rebalanced. if so, reads
571 * information about it.
573 * @post: if assert_size_in_bound(hctx, bound, comparator) succeeds,
574 * odata contains the size, omap, and unwritable attributes for this object.
575 * Otherwise, odata contains the size and unwritable attribute.
577 static int maybe_read_for_balance(cls_method_context_t hctx,
578 object_data &odata, int bound, int comparator) {
579 CLS_LOG(20, "rebalance reading");
580 //if unwritable, return
581 int r = check_writable(hctx);
583 odata.unwritable = true;
584 CLS_LOG(20, "rebalance read: error getting xattr %s: %d", "unwritable", r);
587 odata.unwritable = false;
590 //get the size attribute
592 r = cls_cxx_getxattr(hctx, "size", &size);
594 CLS_LOG(20, "rebalance read: error getting xattr %s: %d", "size", r);
597 odata.size = atoi(string(size.c_str(), size.length()).c_str());
599 //check if it needs to be balanced
600 r = assert_size_in_bound(hctx, bound, comparator);
602 CLS_LOG(20, "rebalance read: error on asserting size: %d", r);
606 //if the assert succeeded, it needs to be balanced
608 r = cls_cxx_map_get_vals(hctx, "", "", LONG_MAX, &odata.omap, &more);
610 CLS_LOG(20, "rebalance read: getting kvs failed with error %d", r);
614 CLS_LOG(20, "rebalance read: size xattr is %llu, omap size is %llu",
615 (unsigned long long)odata.size,
616 (unsigned long long)odata.omap.size());
620 static int maybe_read_for_balance_op(cls_method_context_t hctx,
621 bufferlist *in, bufferlist *out) {
622 CLS_LOG(20, "maybe_read_for_balance");
624 bufferlist::iterator it = in->begin();
627 } catch (buffer::error& err) {
630 int r = maybe_read_for_balance(hctx, op.odata, op.bound, op.comparator);
642 CLS_LOG(20, "Loaded assert condition class!");
644 cls_handle_t h_class;
645 cls_method_handle_t h_get_idata_from_key;
646 cls_method_handle_t h_get_next_idata;
647 cls_method_handle_t h_get_prev_idata;
648 cls_method_handle_t h_read_many;
649 cls_method_handle_t h_check_writable;
650 cls_method_handle_t h_assert_size_in_bound;
651 cls_method_handle_t h_omap_insert;
652 cls_method_handle_t h_create_with_omap;
653 cls_method_handle_t h_omap_remove;
654 cls_method_handle_t h_maybe_read_for_balance;
656 cls_register("kvs", &h_class);
657 cls_register_cxx_method(h_class, "get_idata_from_key",
659 get_idata_from_key_op, &h_get_idata_from_key);
660 cls_register_cxx_method(h_class, "get_next_idata",
662 get_next_idata_op, &h_get_next_idata);
663 cls_register_cxx_method(h_class, "get_prev_idata",
665 get_prev_idata_op, &h_get_prev_idata);
666 cls_register_cxx_method(h_class, "read_many",
668 read_many_op, &h_read_many);
669 cls_register_cxx_method(h_class, "check_writable",
670 CLS_METHOD_RD | CLS_METHOD_WR,
671 check_writable_op, &h_check_writable);
672 cls_register_cxx_method(h_class, "assert_size_in_bound",
674 assert_size_in_bound_op, &h_assert_size_in_bound);
675 cls_register_cxx_method(h_class, "omap_insert",
677 omap_insert_op, &h_omap_insert);
678 cls_register_cxx_method(h_class, "create_with_omap",
680 create_with_omap_op, &h_create_with_omap);
681 cls_register_cxx_method(h_class, "omap_remove",
683 omap_remove_op, &h_omap_remove);
684 cls_register_cxx_method(h_class, "maybe_read_for_balance",
686 maybe_read_for_balance_op, &h_maybe_read_for_balance);