silo/txn_btree.h

   1 #ifndef _NDB_TXN_BTREE_H_
   2 #define _NDB_TXN_BTREE_H_
   3
   4 #include "base_txn_btree.h"
   5
   6 // XXX: hacky
   7 extern void txn_btree_test();
   8
   9 struct txn_btree_ {
  10   class key_reader {
  11   public:
  12     inline ALWAYS_INLINE const std::string &
  13     operator()(const std::string &s)
  14     {
  15       return s;
  16     }
  17 #if NDB_MASSTREE
  18     inline ALWAYS_INLINE lcdf::Str operator()(lcdf::Str s) {
  19       return s;
  20     }
  21 #endif
  22   };
  23
  24   class key_writer {
  25   public:
  26     constexpr key_writer(const std::string *k)
  27       : k(k) {}
  28
  29     template <typename StringAllocator>
  30     inline const std::string *
  31     fully_materialize(bool stable_input, StringAllocator &sa)
  32     {
  33       if (stable_input || !k)
  34         return k;
  35       std::string * const ret = sa();
  36       ret->assign(k->data(), k->size());
  37       return ret;
  38     }
  39
  40   private:
  41     const std::string *k;
  42   };
  43
  44   // does not bother to interpret the bytes from a record
  45   class single_value_reader {
  46   public:
  47     typedef std::string value_type;
  48
  49     constexpr single_value_reader(std::string *px, size_t max_bytes_read)
  50       : px(px), max_bytes_read(max_bytes_read) {}
  51
  52     template <typename StringAllocator>
  53     inline bool
  54     operator()(const uint8_t *data, size_t sz, StringAllocator &sa)
  55     {
  56       const size_t readsz = std::min(sz, max_bytes_read);
  57       px->assign((const char *) data, readsz);
  58       return true;
  59     }
  60
  61     inline std::string &
  62     results()
  63     {
  64       return *px;
  65     }
  66
  67     inline const std::string &
  68     results() const
  69     {
  70       return *px;
  71     }
  72
  73     template <typename StringAllocator>
  74     inline void
  75     dup(const std::string &vdup, StringAllocator &sa)
  76     {
  77       *px = vdup;
  78     }
  79
  80   private:
  81     std::string *px;
  82     size_t max_bytes_read;
  83   };
  84
  85   class value_reader {
  86   public:
  87     typedef std::string value_type;
  88
  89     constexpr value_reader(size_t max_bytes_read)
  90       : px(nullptr), max_bytes_read(max_bytes_read) {}
  91
  92     template <typename StringAllocator>
  93     inline bool
  94     operator()(const uint8_t *data, size_t sz, StringAllocator &sa)
  95     {
  96       px = sa();
  97       const size_t readsz = std::min(sz, max_bytes_read);
  98       px->assign((const char *) data, readsz);
  99       return true;
 100     }
 101
 102     inline std::string &
 103     results()
 104     {
 105       return *px;
 106     }
 107
 108     inline const std::string &
 109     results() const
 110     {
 111       return *px;
 112     }
 113
 114     template <typename StringAllocator>
 115     inline void
 116     dup(const std::string &vdup, StringAllocator &sa)
 117     {
 118       px = sa();
 119       *px = vdup;
 120     }
 121
 122   private:
 123     std::string *px;
 124     size_t max_bytes_read;
 125   };
 126
 127   class value_writer {
 128   public:
 129     constexpr value_writer(const std::string *v) : v(v) {}
 130     inline size_t
 131     compute_needed(const uint8_t *buf, size_t sz)
 132     {
 133       return v ? v->size() : 0;
 134     }
 135     template <typename StringAllocator>
 136     inline const std::string *
 137     fully_materialize(bool stable_input, StringAllocator &sa)
 138     {
 139       if (stable_input || !v)
 140         return v;
 141       std::string * const ret = sa();
 142       ret->assign(v->data(), v->size());
 143       return ret;
 144     }
 145
 146     // input [buf, buf+sz) is old value
 147     inline void
 148     operator()(uint8_t *buf, size_t sz)
 149     {
 150       if (!v)
 151         return;
 152       NDB_MEMCPY(buf, v->data(), v->size());
 153     }
 154   private:
 155     const std::string *v;
 156   };
 157
 158   static size_t
 159   tuple_writer(dbtuple::TupleWriterMode mode, const void *v, uint8_t *p, size_t sz)
 160   {
 161     const std::string * const vx = reinterpret_cast<const std::string *>(v);
 162     switch (mode) {
 163     case dbtuple::TUPLE_WRITER_NEEDS_OLD_VALUE:
 164       return 0;
 165     case dbtuple::TUPLE_WRITER_COMPUTE_NEEDED:
 166     case dbtuple::TUPLE_WRITER_COMPUTE_DELTA_NEEDED:
 167       return vx->size();
 168     case dbtuple::TUPLE_WRITER_DO_WRITE:
 169     case dbtuple::TUPLE_WRITER_DO_DELTA_WRITE:
 170       NDB_MEMCPY(p, vx->data(), vx->size());
 171       return 0;
 172     }
 173     ALWAYS_ASSERT(false);
 174     return 0;
 175   }
 176
 177   typedef std::string Key;
 178   typedef key_reader KeyReader;
 179   typedef key_writer KeyWriter;
 180   typedef std::string Value;
 181   typedef single_value_reader SingleValueReader;
 182   typedef value_reader ValueReader;
 183   typedef value_writer ValueWriter;
 184 };
 185
 186 /**
 187  * This class implements a serializable, multi-version b-tree
 188  *
 189  * It presents mostly same interface as the underlying concurrent b-tree,
 190  * but the interface is serializable. The main interface differences are,
 191  * insert() and put() do not return a boolean value to indicate whether or not
 192  * they caused the tree to mutate
 193  *
 194  * A txn_btree does not allow keys to map to NULL records, even though the
 195  * underlying concurrent btree does- this simplifies some of the book-keeping
 196  * Additionally, keys cannot map to zero length records.
 197  *
 198  * Note that the txn_btree *manages* the memory of both keys and values internally.
 199  * See the specific notes on search()/insert() about memory ownership
 200  */
 201 template <template <typename> class Transaction>
 202 class txn_btree : public base_txn_btree<Transaction, txn_btree_> {
 203   typedef base_txn_btree<Transaction, txn_btree_> super_type;
 204 public:
 205
 206   //template <typename Traits>
 207   //struct transaction {
 208   //  typedef Transaction<txn_btree_, Traits> type;
 209   //};
 210
 211   //template <typename Traits>
 212   //  using transaction = Transaction<txn_btree_, Traits>;
 213
 214   typedef typename super_type::string_type string_type;
 215   typedef typename super_type::keystring_type keystring_type;
 216   typedef typename super_type::size_type size_type;
 217
 218   typedef txn_btree_::Key key_type;
 219   typedef txn_btree_::Value value_type;
 220   typedef txn_btree_::KeyReader key_reader_type;
 221   typedef txn_btree_::KeyWriter key_writer_type;
 222   typedef txn_btree_::SingleValueReader single_value_reader_type;
 223   typedef txn_btree_::ValueReader value_reader_type;
 224   typedef txn_btree_::ValueWriter value_writer_type;
 225
 226   struct search_range_callback {
 227   public:
 228     virtual ~search_range_callback() {}
 229     virtual bool invoke(const keystring_type &k, const string_type &v) = 0;
 230   };
 231
 232 private:
 233
 234   template <typename T>
 235   class type_callback_wrapper : public search_range_callback {
 236   public:
 237     constexpr type_callback_wrapper(T *callback)
 238       : callback(callback) {}
 239     virtual bool
 240     invoke(const keystring_type &k, const string_type &v)
 241     {
 242       return callback->operator()(k, v);
 243     }
 244   private:
 245     T *const callback;
 246   };
 247
 248   static inline ALWAYS_INLINE string_type
 249   to_string_type(const varkey &k)
 250   {
 251     return string_type((const char *) k.data(), k.size());
 252   }
 253
 254   template <typename Traits>
 255   static inline const std::string *
 256   stablize(Transaction<Traits> &t, const std::string &s)
 257   {
 258     if (Traits::stable_input_memory)
 259       return &s;
 260     std::string * const px = t.string_allocator()();
 261     *px = s;
 262     return px;
 263   }
 264
 265   template <typename Traits>
 266   static inline const std::string *
 267   stablize(Transaction<Traits> &t, const uint8_t *p, size_t sz)
 268   {
 269     if (!sz)
 270       return nullptr;
 271     std::string * const px = t.string_allocator()();
 272     px->assign((const char *) p, sz);
 273     return px;
 274   }
 275
 276   template <typename Traits>
 277   static inline const std::string *
 278   stablize(Transaction<Traits> &t, const varkey &k)
 279   {
 280     return stablize(t, k.data(), k.size());
 281   }
 282
 283 public:
 284
 285   txn_btree(size_type value_size_hint = 128,
 286             bool mostly_append = false,
 287             const std::string &name = "<unknown>")
 288     : super_type(value_size_hint, mostly_append, name)
 289   {}
 290
 291   template <typename Traits>
 292   inline bool
 293   search(Transaction<Traits> &t,
 294          const varkey &k,
 295          value_type &v,
 296          size_t max_bytes_read = string_type::npos)
 297   {
 298     return search(t, to_string_type(k), v, max_bytes_read);
 299   }
 300
 301   // either returns false or v is set to not-empty with value
 302   // precondition: max_bytes_read > 0
 303   template <typename Traits>
 304   inline bool
 305   search(Transaction<Traits> &t,
 306          const key_type &k,
 307          value_type &v,
 308          size_type max_bytes_read = string_type::npos)
 309   {
 310     single_value_reader_type r(&v, max_bytes_read);
 311     return this->do_search(t, k, r);
 312   }
 313
 314   template <typename Traits>
 315   inline void
 316   search_range_call(Transaction<Traits> &t,
 317                     const key_type &lower,
 318                     const key_type *upper,
 319                     search_range_callback &callback,
 320                     size_type max_bytes_read = string_type::npos)
 321   {
 322     key_reader_type kr;
 323     value_reader_type vr(max_bytes_read);
 324     this->do_search_range_call(t, lower, upper, callback, kr, vr);
 325   }
 326
 327   template <typename Traits>
 328   inline void
 329   rsearch_range_call(Transaction<Traits> &t,
 330                      const key_type &upper,
 331                      const key_type *lower,
 332                      search_range_callback &callback,
 333                      size_type max_bytes_read = string_type::npos)
 334   {
 335     key_reader_type kr;
 336     value_reader_type vr(max_bytes_read);
 337     this->do_rsearch_range_call(t, upper, lower, callback, kr, vr);
 338   }
 339
 340   template <typename Traits>
 341   inline void
 342   search_range_call(Transaction<Traits> &t,
 343                     const varkey &lower,
 344                     const varkey *upper,
 345                     search_range_callback &callback,
 346                     size_type max_bytes_read = string_type::npos)
 347   {
 348     key_type u;
 349     if (upper)
 350       u = to_string_type(*upper);
 351     search_range_call(t, to_string_type(lower),
 352         upper ? &u : nullptr, callback, max_bytes_read);
 353   }
 354
 355   template <typename Traits>
 356   inline void
 357   rsearch_range_call(Transaction<Traits> &t,
 358                      const varkey &upper,
 359                      const varkey *lower,
 360                      search_range_callback &callback,
 361                      size_type max_bytes_read = string_type::npos)
 362   {
 363     key_type l;
 364     if (lower)
 365       l = to_string_type(*lower);
 366     rsearch_range_call(t, to_string_type(upper),
 367         lower ? &l : nullptr, callback, max_bytes_read);
 368   }
 369
 370   template <typename Traits, typename T>
 371   inline void
 372   search_range(Transaction<Traits> &t,
 373                const key_type &lower,
 374                const key_type *upper,
 375                T &callback,
 376                size_type max_bytes_read = string_type::npos)
 377   {
 378     type_callback_wrapper<T> w(&callback);
 379     search_range_call(t, lower, upper, w, max_bytes_read);
 380   }
 381
 382   template <typename Traits, typename T>
 383   inline void
 384   search_range(Transaction<Traits> &t,
 385                const varkey &lower,
 386                const varkey *upper,
 387                T &callback,
 388                size_type max_bytes_read = string_type::npos)
 389   {
 390     key_type u;
 391     if (upper)
 392       u = to_string_type(*upper);
 393     search_range(t, to_string_type(lower),
 394         upper ? &u : nullptr, callback, max_bytes_read);
 395   }
 396
 397   template <typename Traits>
 398   inline void
 399   put(Transaction<Traits> &t, const key_type &k, const value_type &v)
 400   {
 401     INVARIANT(!v.empty());
 402     this->do_tree_put(
 403         t, stablize(t, k), stablize(t, v),
 404         txn_btree_::tuple_writer, false);
 405   }
 406
 407   template <typename Traits>
 408   inline void
 409   put(Transaction<Traits> &t, const varkey &k, const value_type &v)
 410   {
 411     INVARIANT(!v.empty());
 412     this->do_tree_put(
 413         t, stablize(t, k), stablize(t, v),
 414         txn_btree_::tuple_writer, false);
 415   }
 416
 417   template <typename Traits>
 418   inline void
 419   insert(Transaction<Traits> &t, const key_type &k, const value_type &v)
 420   {
 421     INVARIANT(!v.empty());
 422     this->do_tree_put(
 423         t, stablize(t, k), stablize(t, v),
 424         txn_btree_::tuple_writer, true);
 425   }
 426
 427   // insert() methods below are for legacy use
 428
 429   template <typename Traits>
 430   inline void
 431   insert(Transaction<Traits> &t, const key_type &k, const uint8_t *v, size_type sz)
 432   {
 433     INVARIANT(v);
 434     INVARIANT(sz);
 435     this->do_tree_put(
 436         t, stablize(t, k), stablize(t, v, sz),
 437         txn_btree_::tuple_writer, true);
 438   }
 439
 440   template <typename Traits>
 441   inline void
 442   insert(Transaction<Traits> &t, const varkey &k, const uint8_t *v, size_type sz)
 443   {
 444     INVARIANT(v);
 445     INVARIANT(sz);
 446     this->do_tree_put(
 447         t, stablize(t, k), stablize(t, v, sz),
 448         txn_btree_::tuple_writer, true);
 449   }
 450
 451   template <typename Traits, typename T>
 452   inline void
 453   insert_object(Transaction<Traits> &t, const varkey &k, const T &obj)
 454   {
 455     insert(t, k, (const uint8_t *) &obj, sizeof(obj));
 456   }
 457
 458   template <typename Traits, typename T>
 459   inline void
 460   insert_object(Transaction<Traits> &t, const key_type &k, const T &obj)
 461   {
 462     insert(t, k, (const uint8_t *) &obj, sizeof(obj));
 463   }
 464
 465   template <typename Traits>
 466   inline void
 467   remove(Transaction<Traits> &t, const key_type &k)
 468   {
 469     this->do_tree_put(t, stablize(t, k), nullptr, txn_btree_::tuple_writer, false);
 470   }
 471
 472   template <typename Traits>
 473   inline void
 474   remove(Transaction<Traits> &t, const varkey &k)
 475   {
 476     this->do_tree_put(t, stablize(t, k), nullptr, txn_btree_::tuple_writer, false);
 477   }
 478
 479   static void Test();
 480
 481 };
 482
 483 #endif /* _NDB_TXN_BTREE_H_ */