silo/masstree/value_bag.hh

   1 /* Masstree
   2  * Eddie Kohler, Yandong Mao, Robert Morris
   3  * Copyright (c) 2012-2014 President and Fellows of Harvard College
   4  * Copyright (c) 2012-2014 Massachusetts Institute of Technology
   5  *
   6  * Permission is hereby granted, free of charge, to any person obtaining a
   7  * copy of this software and associated documentation files (the "Software"),
   8  * to deal in the Software without restriction, subject to the conditions
   9  * listed in the Masstree LICENSE file. These conditions include: you must
  10  * preserve this copyright notice, and you cannot mention the copyright
  11  * holders in advertising related to the Software without their permission.
  12  * The Software is provided WITHOUT ANY WARRANTY, EXPRESS OR IMPLIED. This
  13  * notice is a summary of the Masstree LICENSE file; the license in that file
  14  * is legally binding.
  15  */
  16 #ifndef VALUE_BAG_HH
  17 #define VALUE_BAG_HH
  18 #include "kvthread.hh"
  19 #include "json.hh"
  20
  21 template <typename O>
  22 class value_bag {
  23   public:
  24     typedef short index_type;
  25     typedef O offset_type;
  26     typedef lcdf::Str Str;
  27     typedef lcdf::Json Json;
  28
  29   private:
  30     union bagdata {
  31         struct {
  32             offset_type ncol_;
  33             offset_type pos_[1];
  34         };
  35         char s_[0];
  36     };
  37
  38   public:
  39     static const char *name() { return "Bag"; }
  40
  41     inline value_bag();
  42
  43     inline kvtimestamp_t timestamp() const;
  44     inline size_t size() const;
  45     inline int ncol() const;
  46     inline O column_length(int i) const;
  47     inline Str col(int i) const;
  48
  49     inline Str row_string() const;
  50
  51     template <typename ALLOC> inline void deallocate(ALLOC& ti);
  52     template <typename ALLOC> inline void deallocate_rcu(ALLOC& ti);
  53
  54     template <typename ALLOC>
  55     value_bag<O>* update(const Json* first, const Json* last, kvtimestamp_t ts,
  56                          ALLOC& ti) const;
  57     template <typename ALLOC>
  58     inline value_bag<O>* update(int col, Str value,
  59                                 kvtimestamp_t ts, ALLOC& ti) const;
  60     template <typename ALLOC>
  61     static value_bag<O>* create(const Json* first, const Json* last,
  62                                 kvtimestamp_t ts, ALLOC& ti);
  63     template <typename ALLOC>
  64     static value_bag<O>* create1(Str value, kvtimestamp_t ts, ALLOC& ti);
  65     template <typename ALLOC>
  66     inline void deallocate_rcu_after_update(const Json* first, const Json* last, ALLOC& ti);
  67     template <typename ALLOC>
  68     inline void deallocate_after_failed_update(const Json* first, const Json* last, ALLOC& ti);
  69
  70     template <typename PARSER, typename ALLOC>
  71     static value_bag<O>* checkpoint_read(PARSER& par, kvtimestamp_t ts,
  72                                          ALLOC& ti);
  73     template <typename UNPARSER>
  74     inline void checkpoint_write(UNPARSER& unpar) const;
  75
  76     void print(FILE* f, const char* prefix, int indent, Str key,
  77                kvtimestamp_t initial_ts, const char* suffix = "");
  78
  79   private:
  80     kvtimestamp_t ts_;
  81     bagdata d_;
  82 };
  83
  84
  85 template <typename O>
  86 inline value_bag<O>::value_bag()
  87     : ts_(0) {
  88     d_.ncol_ = 0;
  89     d_.pos_[0] = sizeof(bagdata);
  90 }
  91
  92 template <typename O>
  93 inline kvtimestamp_t value_bag<O>::timestamp() const {
  94     return ts_;
  95 }
  96
  97 template <typename O>
  98 inline size_t value_bag<O>::size() const {
  99     return sizeof(kvtimestamp_t) + d_.pos_[d_.ncol_];
 100 }
 101
 102 template <typename O>
 103 inline int value_bag<O>::ncol() const {
 104     return d_.ncol_;
 105 }
 106
 107 template <typename O>
 108 inline O value_bag<O>::column_length(int i) const {
 109     return d_.pos_[i + 1] - d_.pos_[i];
 110 }
 111
 112 template <typename O>
 113 inline lcdf::Str value_bag<O>::col(int i) const {
 114     if (unsigned(i) < unsigned(d_.ncol_))
 115         return Str(d_.s_ + d_.pos_[i], column_length(i));
 116     else
 117         return Str();
 118 }
 119
 120 template <typename O>
 121 inline lcdf::Str value_bag<O>::row_string() const {
 122     return Str(d_.s_, d_.pos_[d_.ncol_]);
 123 }
 124
 125 template <typename O> template <typename ALLOC>
 126 inline void value_bag<O>::deallocate(ALLOC& ti) {
 127     ti.deallocate(this, size(), memtag_value);
 128 }
 129
 130 template <typename O> template <typename ALLOC>
 131 inline void value_bag<O>::deallocate_rcu(ALLOC& ti) {
 132     ti.deallocate_rcu(this, size(), memtag_value);
 133 }
 134
 135 // prerequisite: [first, last) is an array [column, value, column, value, ...]
 136 // each column is unsigned; the columns are strictly increasing;
 137 // each value is a string
 138 template <typename O> template <typename ALLOC>
 139 value_bag<O>* value_bag<O>::update(const Json* first, const Json* last,
 140                                    kvtimestamp_t ts, ALLOC& ti) const
 141 {
 142     size_t sz = size();
 143     unsigned ncol = d_.ncol_;
 144     for (auto it = first; it != last; it += 2) {
 145         unsigned idx = it[0].as_u();
 146         sz += it[1].as_s().length();
 147         if (idx < d_.ncol_)
 148             sz -= column_length(idx);
 149         else
 150             ncol = idx + 1;
 151     }
 152     if (ncol > d_.ncol_)
 153         sz += (ncol - d_.ncol_) * sizeof(offset_type);
 154
 155     value_bag<O>* row = (value_bag<O>*) ti.allocate(sz, memtag_value);
 156     row->ts_ = ts;
 157
 158     // Minor optimization: Replacing one small column without changing length
 159     if (ncol == d_.ncol_ && sz == size() && first + 2 == last
 160         && first[1].as_s().length() <= 16) {
 161         memcpy(row->d_.s_, d_.s_, sz - sizeof(kvtimestamp_t));
 162         memcpy(row->d_.s_ + d_.pos_[first[0].as_u()],
 163                first[1].as_s().data(), first[1].as_s().length());
 164         return row;
 165     }
 166
 167     // Otherwise need to do more work
 168     row->d_.ncol_ = ncol;
 169     sz = sizeof(bagdata) + ncol * sizeof(offset_type);
 170     unsigned col = 0;
 171     while (1) {
 172         unsigned this_col = (first != last ? first[0].as_u() : ncol);
 173
 174         // copy data from old row
 175         if (col != this_col && col < d_.ncol_) {
 176             unsigned end_col = std::min(this_col, unsigned(d_.ncol_));
 177             ssize_t delta = sz - d_.pos_[col];
 178             if (delta == 0)
 179                 memcpy(row->d_.pos_ + col, d_.pos_ + col,
 180                        sizeof(offset_type) * (end_col - col));
 181             else
 182                 for (unsigned i = col; i < end_col; ++i)
 183                     row->d_.pos_[i] = d_.pos_[i] + delta;
 184             size_t amt = d_.pos_[end_col] - d_.pos_[col];
 185             memcpy(row->d_.s_ + sz, d_.s_ + d_.pos_[col], amt);
 186             sz += amt;
 187             col = end_col;
 188         }
 189
 190         // mark empty columns if we're extending
 191         while (col != this_col) {
 192             row->d_.pos_[col] = sz;
 193             ++col;
 194         }
 195
 196         if (col == ncol)
 197             break;
 198
 199         // copy data from change
 200         row->d_.pos_[col] = sz;
 201         Str val = first[1].as_s();
 202         memcpy(row->d_.s_ + sz, val.data(), val.length());
 203         sz += val.length();
 204         first += 2;
 205         ++col;
 206     }
 207     row->d_.pos_[ncol] = sz;
 208     return row;
 209 }
 210
 211 template <typename O> template <typename ALLOC>
 212 inline value_bag<O>* value_bag<O>::update(int col, Str value, kvtimestamp_t ts,
 213                                           ALLOC& ti) const {
 214     Json change[2] = {Json(col), Json(value)};
 215     return update(&change[0], &change[2], ts, ti);
 216 }
 217
 218 template <typename O> template <typename ALLOC>
 219 inline value_bag<O>* value_bag<O>::create(const Json* first, const Json* last,
 220                                           kvtimestamp_t ts, ALLOC& ti) {
 221     value_bag<O> empty;
 222     return empty.update(first, last, ts, ti);
 223 }
 224
 225 template <typename O> template <typename ALLOC>
 226 inline value_bag<O>* value_bag<O>::create1(Str str, kvtimestamp_t ts,
 227                                            ALLOC& ti) {
 228     value_bag<O>* row = (value_bag<O>*) ti.allocate(sizeof(kvtimestamp_t) + sizeof(bagdata) + sizeof(O) + str.length(), memtag_value);
 229     row->ts_ = ts;
 230     row->d_.ncol_ = 1;
 231     row->d_.pos_[0] = sizeof(bagdata) + sizeof(O);
 232     row->d_.pos_[1] = sizeof(bagdata) + sizeof(O) + str.length();
 233     memcpy(row->d_.s_ + row->d_.pos_[0], str.data(), str.length());
 234     return row;
 235 }
 236
 237 template <typename O> template <typename ALLOC>
 238 inline void value_bag<O>::deallocate_rcu_after_update(const Json*, const Json*, ALLOC& ti) {
 239     deallocate_rcu(ti);
 240 }
 241
 242 template <typename O> template <typename ALLOC>
 243 inline void value_bag<O>::deallocate_after_failed_update(const Json*, const Json*, ALLOC& ti) {
 244     deallocate(ti);
 245 }
 246
 247 template <typename O> template <typename PARSER, typename ALLOC>
 248 inline value_bag<O>* value_bag<O>::checkpoint_read(PARSER& par,
 249                                                    kvtimestamp_t ts,
 250                                                    ALLOC& ti) {
 251     Str value;
 252     par >> value;
 253     value_bag<O>* row = (value_bag<O>*) ti.allocate(sizeof(kvtimestamp_t) + value.length(), memtag_value);
 254     row->ts_ = ts;
 255     memcpy(row->d_.s_, value.data(), value.length());
 256     return row;
 257 }
 258
 259 template <typename O> template <typename UNPARSER>
 260 inline void value_bag<O>::checkpoint_write(UNPARSER& unpar) const {
 261     unpar << Str(d_.s_, d_.pos_[d_.ncol_]);
 262 }
 263
 264 template <typename O>
 265 void value_bag<O>::print(FILE *f, const char *prefix, int indent,
 266                          Str key, kvtimestamp_t initial_ts,
 267                          const char *suffix)
 268 {
 269     kvtimestamp_t adj_ts = timestamp_sub(ts_, initial_ts);
 270     if (d_.ncol_ == 1)
 271         fprintf(f, "%s%*s%.*s = %.*s @" PRIKVTSPARTS "%s\n", prefix, indent, "",
 272                 key.len, key.s, d_.pos_[1] - d_.pos_[0], d_.s_ + d_.pos_[0],
 273                 KVTS_HIGHPART(adj_ts), KVTS_LOWPART(adj_ts), suffix);
 274     else {
 275         fprintf(f, "%s%*s%.*s = [", prefix, indent, "", key.len, key.s);
 276         for (int col = 0; col < d_.ncol_; ++col) {
 277             int pos = d_.pos_[col], len = std::min(40, d_.pos_[col + 1] - pos);
 278             fprintf(f, col ? "|%.*s" : "%.*s", len, d_.s_ + pos);
 279         }
 280         fprintf(f, "] @" PRIKVTSPARTS "%s\n",
 281                 KVTS_HIGHPART(adj_ts), KVTS_LOWPART(adj_ts), suffix);
 282     }
 283 }
 284
 285 #endif