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
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
18 #include "kvthread.hh"
24 typedef short index_type;
25 typedef O offset_type;
26 typedef lcdf::Str Str;
27 typedef lcdf::Json Json;
39 static const char *name() { return "Bag"; }
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;
49 inline Str row_string() const;
51 template <typename ALLOC> inline void deallocate(ALLOC& ti);
52 template <typename ALLOC> inline void deallocate_rcu(ALLOC& ti);
54 template <typename ALLOC>
55 value_bag<O>* update(const Json* first, const Json* last, kvtimestamp_t ts,
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);
70 template <typename PARSER, typename ALLOC>
71 static value_bag<O>* checkpoint_read(PARSER& par, kvtimestamp_t ts,
73 template <typename UNPARSER>
74 inline void checkpoint_write(UNPARSER& unpar) const;
76 void print(FILE* f, const char* prefix, int indent, Str key,
77 kvtimestamp_t initial_ts, const char* suffix = "");
86 inline value_bag<O>::value_bag()
89 d_.pos_[0] = sizeof(bagdata);
93 inline kvtimestamp_t value_bag<O>::timestamp() const {
98 inline size_t value_bag<O>::size() const {
99 return sizeof(kvtimestamp_t) + d_.pos_[d_.ncol_];
102 template <typename O>
103 inline int value_bag<O>::ncol() const {
107 template <typename O>
108 inline O value_bag<O>::column_length(int i) const {
109 return d_.pos_[i + 1] - d_.pos_[i];
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));
120 template <typename O>
121 inline lcdf::Str value_bag<O>::row_string() const {
122 return Str(d_.s_, d_.pos_[d_.ncol_]);
125 template <typename O> template <typename ALLOC>
126 inline void value_bag<O>::deallocate(ALLOC& ti) {
127 ti.deallocate(this, size(), memtag_value);
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);
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
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();
148 sz -= column_length(idx);
153 sz += (ncol - d_.ncol_) * sizeof(offset_type);
155 value_bag<O>* row = (value_bag<O>*) ti.allocate(sz, memtag_value);
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());
167 // Otherwise need to do more work
168 row->d_.ncol_ = ncol;
169 sz = sizeof(bagdata) + ncol * sizeof(offset_type);
172 unsigned this_col = (first != last ? first[0].as_u() : ncol);
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];
179 memcpy(row->d_.pos_ + col, d_.pos_ + col,
180 sizeof(offset_type) * (end_col - col));
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);
190 // mark empty columns if we're extending
191 while (col != this_col) {
192 row->d_.pos_[col] = sz;
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());
207 row->d_.pos_[ncol] = sz;
211 template <typename O> template <typename ALLOC>
212 inline value_bag<O>* value_bag<O>::update(int col, Str value, kvtimestamp_t ts,
214 Json change[2] = {Json(col), Json(value)};
215 return update(&change[0], &change[2], ts, ti);
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) {
222 return empty.update(first, last, ts, ti);
225 template <typename O> template <typename ALLOC>
226 inline value_bag<O>* value_bag<O>::create1(Str str, kvtimestamp_t ts,
228 value_bag<O>* row = (value_bag<O>*) ti.allocate(sizeof(kvtimestamp_t) + sizeof(bagdata) + sizeof(O) + str.length(), memtag_value);
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());
237 template <typename O> template <typename ALLOC>
238 inline void value_bag<O>::deallocate_rcu_after_update(const Json*, const Json*, ALLOC& ti) {
242 template <typename O> template <typename ALLOC>
243 inline void value_bag<O>::deallocate_after_failed_update(const Json*, const Json*, ALLOC& ti) {
247 template <typename O> template <typename PARSER, typename ALLOC>
248 inline value_bag<O>* value_bag<O>::checkpoint_read(PARSER& par,
253 value_bag<O>* row = (value_bag<O>*) ti.allocate(sizeof(kvtimestamp_t) + value.length(), memtag_value);
255 memcpy(row->d_.s_, value.data(), value.length());
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_]);
264 template <typename O>
265 void value_bag<O>::print(FILE *f, const char *prefix, int indent,
266 Str key, kvtimestamp_t initial_ts,
269 kvtimestamp_t adj_ts = timestamp_sub(ts_, initial_ts);
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);
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);
280 fprintf(f, "] @" PRIKVTSPARTS "%s\n",
281 KVTS_HIGHPART(adj_ts), KVTS_LOWPART(adj_ts), suffix);