Skip to content

M
MariaDB
Commit 3b923630 authored by Yoni Fogel's avatar Yoni Fogel
Browse files
Options

Refs Tokutek/ft-index#46 Minor optimizations, added some todos for code review. WILL NOT COMPILE

parent 2041d749
Show whitespace changes
Inline Side-by-side
Showing with 175 additions and 226 deletions
46.code-review
View file @ 3b923630
Notes during 2014-01-08 Leif/Yoni
-Should verify (dmt?omt?bndata?) crash or return error on failed verify
DECISIONS:
Replace dmt_functor with implicit interface only. Instead of (for data type x) requiring the name to be dmt_functor<x> just pass the writer's class name into the dmt's template as a new parameter.
Replace dmt_functor<default> with comments explaining the "interface"
-==========================================-
See wiki: See wiki:
https://github.com/Tokutek/ft-index/wiki/Improving-in-memory-query-performance---Design https://github.com/Tokutek/ft-index/wiki/Improving-in-memory-query-performance---Design
......
util/dmt.cc
View file @ 3b923630
...@@ -98,11 +98,23 @@ namespace toku { ...@@ -98,11 +98,23 @@ namespace toku {
template<typename dmtdata_t, typename dmtdataout_t> template<typename dmtdata_t, typename dmtdataout_t>
void dmt<dmtdata_t, dmtdataout_t>::create(void) { void dmt<dmtdata_t, dmtdataout_t>::create(void) {
this->create_internal_no_alloc(false); toku_mempool_zero(&this->mp);
this->values_same_size = true;
this->value_length = 0;
this->is_array = true;
this->d.a.num_values = 0;
//TODO: maybe allocate enough space for something by default? //TODO: maybe allocate enough space for something by default?
// We may be relying on not needing to allocate space the first time (due to limited time spent while a lock is held) // We may be relying on not needing to allocate space the first time (due to limited time spent while a lock is held)
} }
/**
* Note: create_from_sorted_memory_of_fixed_size_elements does not take ownership of 'mem'.
* Owner is still responsible for freeing it.
* While in the OMT a similar function would steal ownership, this doesn't make sense for the DMT because
* we (usually) have to add padding for alignment (mem has all of the elements PACKED).
* Also all current uses (as of Jan 12, 2014) of this function would require mallocing a new array
* in order to allow stealing.
*/
template<typename dmtdata_t, typename dmtdataout_t> template<typename dmtdata_t, typename dmtdataout_t>
void dmt<dmtdata_t, dmtdataout_t>::create_from_sorted_memory_of_fixed_size_elements( void dmt<dmtdata_t, dmtdataout_t>::create_from_sorted_memory_of_fixed_size_elements(
const void *mem, const void *mem,
...@@ -112,20 +124,21 @@ void dmt<dmtdata_t, dmtdataout_t>::create_from_sorted_memory_of_fixed_size_eleme ...@@ -112,20 +124,21 @@ void dmt<dmtdata_t, dmtdataout_t>::create_from_sorted_memory_of_fixed_size_eleme
this->values_same_size = true; this->values_same_size = true;
this->value_length = fixed_value_length; this->value_length = fixed_value_length;
this->is_array = true; this->is_array = true;
this->d.a.start_idx = 0;
this->d.a.num_values = numvalues; this->d.a.num_values = numvalues;
const uint8_t pad_bytes = get_fixed_length_alignment_overhead(); const uint8_t pad_bytes = get_fixed_length_alignment_overhead();
uint32_t aligned_memsize = mem_length + numvalues * pad_bytes; uint32_t aligned_memsize = mem_length + numvalues * pad_bytes;
toku_mempool_construct(&this->mp, aligned_memsize); toku_mempool_construct(&this->mp, aligned_memsize);
if (aligned_memsize > 0) { if (aligned_memsize > 0) {
paranoid_invariant(numvalues > 0);
void *ptr = toku_mempool_malloc(&this->mp, aligned_memsize, 1); void *ptr = toku_mempool_malloc(&this->mp, aligned_memsize, 1);
paranoid_invariant_notnull(ptr); paranoid_invariant_notnull(ptr);
uint8_t *CAST_FROM_VOIDP(dest, ptr); uint8_t * const CAST_FROM_VOIDP(dest, ptr);
const uint8_t *CAST_FROM_VOIDP(src, mem); const uint8_t * const CAST_FROM_VOIDP(src, mem);
if (pad_bytes == 0) { if (pad_bytes == 0) {
paranoid_invariant(aligned_memsize == mem_length); paranoid_invariant(aligned_memsize == mem_length);
memcpy(dest, src, aligned_memsize); memcpy(dest, src, aligned_memsize);
} else { } else {
// TODO(leif): check what vectorizes best: multiplying like this or adding to offsets
const uint32_t fixed_len = this->value_length; const uint32_t fixed_len = this->value_length;
const uint32_t fixed_aligned_len = align(this->value_length); const uint32_t fixed_aligned_len = align(this->value_length);
paranoid_invariant(this->d.a.num_values*fixed_len == mem_length); paranoid_invariant(this->d.a.num_values*fixed_len == mem_length);
...@@ -136,25 +149,6 @@ void dmt<dmtdata_t, dmtdataout_t>::create_from_sorted_memory_of_fixed_size_eleme ...@@ -136,25 +149,6 @@ void dmt<dmtdata_t, dmtdataout_t>::create_from_sorted_memory_of_fixed_size_eleme
} }
} }
template<typename dmtdata_t, typename dmtdataout_t>
void dmt<dmtdata_t, dmtdataout_t>::create_no_array(void) {
this->create_internal_no_alloc(false);
}
template<typename dmtdata_t, typename dmtdataout_t>
void dmt<dmtdata_t, dmtdataout_t>::create_internal_no_alloc(bool as_tree) {
toku_mempool_zero(&this->mp);
this->values_same_size = true;
this->value_length = 0;
this->is_array = !as_tree;
if (as_tree) {
this->d.t.root.set_to_null();
} else {
this->d.a.start_idx = 0;
this->d.a.num_values = 0;
}
}
template<typename dmtdata_t, typename dmtdataout_t> template<typename dmtdata_t, typename dmtdataout_t>
void dmt<dmtdata_t, dmtdataout_t>::clone(const dmt &src) { void dmt<dmtdata_t, dmtdataout_t>::clone(const dmt &src) {
*this = src; *this = src;
...@@ -164,11 +158,12 @@ void dmt<dmtdata_t, dmtdataout_t>::clone(const dmt &src) { ...@@ -164,11 +158,12 @@ void dmt<dmtdata_t, dmtdataout_t>::clone(const dmt &src) {
template<typename dmtdata_t, typename dmtdataout_t> template<typename dmtdata_t, typename dmtdataout_t>
void dmt<dmtdata_t, dmtdataout_t>::clear(void) { void dmt<dmtdata_t, dmtdataout_t>::clear(void) {
this->is_array = true; this->is_array = true;
this->d.a.start_idx = 0;
this->d.a.num_values = 0; this->d.a.num_values = 0;
this->values_same_size = true; // Reset state this->values_same_size = true; // Reset state
this->value_length = 0; this->value_length = 0;
toku_mempool_destroy(&this->mp); //TODO(leif): Note that this can mess with our memory_footprint calculation (we may touch past what is marked as 'used' in the mempool)
// One 'fix' is for mempool to also track what was touched, and reset() shouldn't reset that, though realloc() might.
toku_mempool_reset(&this->mp);
} }
template<typename dmtdata_t, typename dmtdataout_t> template<typename dmtdata_t, typename dmtdataout_t>
...@@ -220,28 +215,17 @@ int dmt<dmtdata_t, dmtdataout_t>::insert_at(const dmtdatain_t &value, const uint ...@@ -220,28 +215,17 @@ int dmt<dmtdata_t, dmtdataout_t>::insert_at(const dmtdatain_t &value, const uint
if (idx > this->size()) { return EINVAL; } if (idx > this->size()) { return EINVAL; }
bool same_size = this->values_same_size && (this->size() == 0 || value.get_dmtdatain_t_size() == this->value_length); bool same_size = this->values_same_size && (this->size() == 0 || value.get_dmtdatain_t_size() == this->value_length);
if (same_size) {
if (this->is_array) { if (this->is_array) {
if (idx == this->d.a.num_values) { if (same_size && idx == this->d.a.num_values) {
return this->insert_at_array_end<true>(value); return this->insert_at_array_end<true>(value);
} }
#if 0
//TODO: enable if we support delete_at with array
if (idx == 0 && this->d.a.start_idx > 0) {
paranoid_invariant(false); // Should not be possible (yet)
return this->insert_at_array_beginning(value);
}
#endif
}
}
if (this->is_array) {
this->convert_from_array_to_tree(); this->convert_from_array_to_tree();
} }
// Is a tree.
paranoid_invariant(!is_array);
if (!same_size) { if (!same_size) {
this->values_same_size = false; this->values_same_size = false;
} }
paranoid_invariant(!is_array);
// Is a tree.
this->maybe_resize_tree(&value); this->maybe_resize_tree(&value);
subtree *rebalance_subtree = nullptr; subtree *rebalance_subtree = nullptr;
...@@ -263,27 +247,13 @@ int dmt<dmtdata_t, dmtdataout_t>::insert_at_array_end(const dmtdatain_t& value_i ...@@ -263,27 +247,13 @@ int dmt<dmtdata_t, dmtdataout_t>::insert_at_array_end(const dmtdatain_t& value_i
paranoid_invariant(this->value_length == value_in.get_dmtdatain_t_size()); paranoid_invariant(this->value_length == value_in.get_dmtdatain_t_size());
if (with_resize) { if (with_resize) {
this->maybe_resize_array(+1); this->maybe_resize_array_for_insert();
} }
dmtdata_t *dest = this->alloc_array_value_end(); dmtdata_t *dest = this->alloc_array_value_end();
value_in.write_dmtdata_t_to(dest); value_in.write_dmtdata_t_to(dest);
return 0; return 0;
} }
template<typename dmtdata_t, typename dmtdataout_t>
int dmt<dmtdata_t, dmtdataout_t>::insert_at_array_beginning(const dmtdatain_t& value_in) {
invariant(false); //TODO: enable this later
paranoid_invariant(this->is_array);
paranoid_invariant(this->values_same_size);
paranoid_invariant(this->d.a.num_values > 0);
//TODO: when deleting last element, should set start_idx to 0
this->maybe_resize_array(+1); // +1 or 0? Depends on how memory management works
dmtdata_t *dest = this->alloc_array_value_beginning();
value_in.write_dmtdata_t_to(dest);
return 0;
}
template<typename dmtdata_t, typename dmtdataout_t> template<typename dmtdata_t, typename dmtdataout_t>
dmtdata_t * dmt<dmtdata_t, dmtdataout_t>::alloc_array_value_end(void) { dmtdata_t * dmt<dmtdata_t, dmtdataout_t>::alloc_array_value_end(void) {
paranoid_invariant(this->is_array); paranoid_invariant(this->is_array);
...@@ -298,55 +268,37 @@ dmtdata_t * dmt<dmtdata_t, dmtdataout_t>::alloc_array_value_end(void) { ...@@ -298,55 +268,37 @@ dmtdata_t * dmt<dmtdata_t, dmtdataout_t>::alloc_array_value_end(void) {
return n; return n;
} }
template<typename dmtdata_t, typename dmtdataout_t>
dmtdata_t * dmt<dmtdata_t, dmtdataout_t>::alloc_array_value_beginning(void) {
invariant(false); //TODO: enable this later
paranoid_invariant(this->is_array);
paranoid_invariant(this->values_same_size);
paranoid_invariant(this->d.a.start_idx > 0);
const uint32_t real_idx = --this->d.a.start_idx;
this->d.a.num_values++;
//TODO: figure out how to keep mempool correct here.. do we free during delete_at (begin)? If so how do we re'malloc' from beginning? Alternatively never free from beginning?
return get_array_value_internal(&this->mp, real_idx);
}
template<typename dmtdata_t, typename dmtdataout_t> template<typename dmtdata_t, typename dmtdataout_t>
dmtdata_t * dmt<dmtdata_t, dmtdataout_t>::get_array_value(const uint32_t idx) const { dmtdata_t * dmt<dmtdata_t, dmtdataout_t>::get_array_value(const uint32_t idx) const {
paranoid_invariant(this->is_array); paranoid_invariant(this->is_array);
paranoid_invariant(this->values_same_size); paranoid_invariant(this->values_same_size);
//TODO: verify initial create always set is_array and values_same_size
paranoid_invariant(idx < this->d.a.num_values); paranoid_invariant(idx < this->d.a.num_values);
const uint32_t real_idx = idx + this->d.a.start_idx; return get_array_value_internal(&this->mp, idx);
return get_array_value_internal(&this->mp, real_idx);
} }
template<typename dmtdata_t, typename dmtdataout_t> template<typename dmtdata_t, typename dmtdataout_t>
dmtdata_t * dmt<dmtdata_t, dmtdataout_t>::get_array_value_internal(const struct mempool *mempool, const uint32_t real_idx) const { dmtdata_t * dmt<dmtdata_t, dmtdataout_t>::get_array_value_internal(const struct mempool *mempool, const uint32_t idx) const {
void* ptr = toku_mempool_get_pointer_from_base_and_offset(mempool, real_idx * align(this->value_length)); void* ptr = toku_mempool_get_pointer_from_base_and_offset(mempool, idx * align(this->value_length));
dmtdata_t *CAST_FROM_VOIDP(value, ptr); dmtdata_t *CAST_FROM_VOIDP(value, ptr);
return value; return value;
} }
//TODO(leif) write microbenchmarks to compare growth factor. Note: growth factor here is actually 2.5 because of mempool_construct
template<typename dmtdata_t, typename dmtdataout_t> template<typename dmtdata_t, typename dmtdataout_t>
void dmt<dmtdata_t, dmtdataout_t>::maybe_resize_array(const int change) { void dmt<dmtdata_t, dmtdataout_t>::maybe_resize_array_for_insert(void) {
paranoid_invariant(change == -1 || change == 1); bool space_available = toku_mempool_get_free_space(&this->mp) >= align(this->value_length);
paranoid_invariant(change == 1); //TODO: go over change == -1.. may or may not be ok
bool space_available = change < 0 || toku_mempool_get_free_space(&this->mp) >= align(this->value_length); if (!space_available) {
const uint32_t n = this->d.a.num_values + 1;
const uint32_t n = this->d.a.num_values + change; const uint32_t new_n = n <=2 ? 4 : 2*n;
const uint32_t new_n = n<=2 ? 4 : 2*n;
const uint32_t new_space = align(this->value_length) * new_n; const uint32_t new_space = align(this->value_length) * new_n;
bool too_much_space = new_space <= toku_mempool_get_size(&this->mp) / 2;
if (!space_available || too_much_space) {
struct mempool new_kvspace; struct mempool new_kvspace;
toku_mempool_construct(&new_kvspace, new_space); toku_mempool_construct(&new_kvspace, new_space);
size_t copy_bytes = this->d.a.num_values * align(this->value_length); size_t copy_bytes = this->d.a.num_values * align(this->value_length);
invariant(copy_bytes + align(this->value_length) <= new_space); invariant(copy_bytes + align(this->value_length) <= new_space);
paranoid_invariant(copy_bytes <= toku_mempool_get_used_space(&this->mp));
// Copy over to new mempool // Copy over to new mempool
if (this->d.a.num_values > 0) { if (this->d.a.num_values > 0) {
void* dest = toku_mempool_malloc(&new_kvspace, copy_bytes, 1); void* dest = toku_mempool_malloc(&new_kvspace, copy_bytes, 1);
...@@ -355,7 +307,6 @@ void dmt<dmtdata_t, dmtdataout_t>::maybe_resize_array(const int change) { ...@@ -355,7 +307,6 @@ void dmt<dmtdata_t, dmtdataout_t>::maybe_resize_array(const int change) {
} }
toku_mempool_destroy(&this->mp); toku_mempool_destroy(&this->mp);
this->mp = new_kvspace; this->mp = new_kvspace;
this->d.a.start_idx = 0;
} }
} }
...@@ -378,16 +329,16 @@ void dmt<dmtdata_t, dmtdataout_t>::convert_from_tree_to_array(void) { ...@@ -378,16 +329,16 @@ void dmt<dmtdata_t, dmtdataout_t>::convert_from_tree_to_array(void) {
const uint32_t num_values = this->size(); const uint32_t num_values = this->size();
node_idx *tmp_array; node_offset *tmp_array;
bool malloced = false; bool malloced = false;
tmp_array = alloc_temp_node_idxs(num_values); tmp_array = alloc_temp_node_offsets(num_values);
if (!tmp_array) { if (!tmp_array) {
malloced = true; malloced = true;
XMALLOC_N(num_values, tmp_array); XMALLOC_N(num_values, tmp_array);
} }
this->fill_array_with_subtree_idxs(tmp_array, this->d.t.root); this->fill_array_with_subtree_offsets(tmp_array, this->d.t.root);
struct mempool new_mp = this->mp; struct mempool new_mp;
const uint32_t fixed_len = this->value_length; const uint32_t fixed_len = this->value_length;
const uint32_t fixed_aligned_len = align(this->value_length); const uint32_t fixed_aligned_len = align(this->value_length);
size_t mem_needed = num_values * fixed_aligned_len; size_t mem_needed = num_values * fixed_aligned_len;
...@@ -401,7 +352,6 @@ void dmt<dmtdata_t, dmtdataout_t>::convert_from_tree_to_array(void) { ...@@ -401,7 +352,6 @@ void dmt<dmtdata_t, dmtdataout_t>::convert_from_tree_to_array(void) {
toku_mempool_destroy(&this->mp); toku_mempool_destroy(&this->mp);
this->mp = new_mp; this->mp = new_mp;
this->is_array = true; this->is_array = true;
this->d.a.start_idx = 0;
this->d.a.num_values = num_values; this->d.a.num_values = num_values;
if (malloced) toku_free(tmp_array); if (malloced) toku_free(tmp_array);
...@@ -414,12 +364,10 @@ void dmt<dmtdata_t, dmtdataout_t>::convert_from_array_to_tree(void) { ...@@ -414,12 +364,10 @@ void dmt<dmtdata_t, dmtdataout_t>::convert_from_array_to_tree(void) {
//save array-format information to locals //save array-format information to locals
const uint32_t num_values = this->d.a.num_values; const uint32_t num_values = this->d.a.num_values;
const uint32_t offset = this->d.a.start_idx;
paranoid_invariant_zero(offset); //TODO: remove this
node_idx *tmp_array; node_offset *tmp_array;
bool malloced = false; bool malloced = false;
tmp_array = alloc_temp_node_idxs(num_values); tmp_array = alloc_temp_node_offsets(num_values);
if (!tmp_array) { if (!tmp_array) {
malloced = true; malloced = true;
XMALLOC_N(num_values, tmp_array); XMALLOC_N(num_values, tmp_array);
...@@ -430,11 +378,11 @@ void dmt<dmtdata_t, dmtdataout_t>::convert_from_array_to_tree(void) { ...@@ -430,11 +378,11 @@ void dmt<dmtdata_t, dmtdataout_t>::convert_from_array_to_tree(void) {
toku_mempool_construct(&this->mp, mem_needed); toku_mempool_construct(&this->mp, mem_needed);
for (uint32_t i = 0; i < num_values; i++) { for (uint32_t i = 0; i < num_values; i++) {
dmtdatain_t functor(this->value_length, get_array_value_internal(&old_mp, i+offset)); dmtdatain_t functor(this->value_length, get_array_value_internal(&old_mp, i));
tmp_array[i] = node_malloc_and_set_value(functor); tmp_array[i] = node_malloc_and_set_value(functor);
} }
this->is_array = false; this->is_array = false;
this->rebuild_subtree_from_idxs(&this->d.t.root, tmp_array, num_values); this->rebuild_subtree_from_offsets(&this->d.t.root, tmp_array, num_values);
if (malloced) toku_free(tmp_array); if (malloced) toku_free(tmp_array);
toku_mempool_destroy(&old_mp); toku_mempool_destroy(&old_mp);
...@@ -450,21 +398,15 @@ int dmt<dmtdata_t, dmtdataout_t>::delete_at(const uint32_t idx) { ...@@ -450,21 +398,15 @@ int dmt<dmtdata_t, dmtdataout_t>::delete_at(const uint32_t idx) {
return 0; return 0;
} }
if (this->is_array) { if (this->is_array) {
//TODO: support array delete
//TODO: If/when we implement array delete, we must update verify() (and others?) w.r.t. mempool fragmentation
this->convert_from_array_to_tree(); this->convert_from_array_to_tree();
} }
paranoid_invariant(!is_array); paranoid_invariant(!is_array);
if (this->is_array) {
paranoid_invariant(false);
} else {
subtree *rebalance_subtree = nullptr; subtree *rebalance_subtree = nullptr;
this->delete_internal(&this->d.t.root, idx, nullptr, &rebalance_subtree); this->delete_internal(&this->d.t.root, idx, nullptr, &rebalance_subtree);
if (rebalance_subtree != nullptr) { if (rebalance_subtree != nullptr) {
this->rebalance(rebalance_subtree); this->rebalance(rebalance_subtree);
} }
}
this->maybe_resize_tree(nullptr); this->maybe_resize_tree(nullptr);
return 0; return 0;
} }
...@@ -488,6 +430,7 @@ int dmt<dmtdata_t, dmtdataout_t>::iterate_on_range(const uint32_t left, const ui ...@@ -488,6 +430,7 @@ int dmt<dmtdata_t, dmtdataout_t>::iterate_on_range(const uint32_t left, const ui
return this->iterate_internal<iterate_extra_t, f>(left, right, this->d.t.root, 0, iterate_extra); return this->iterate_internal<iterate_extra_t, f>(left, right, this->d.t.root, 0, iterate_extra);
} }
//TODO(yoni) determine where this is used and if it should crash or return an error on bad verify
template<typename dmtdata_t, typename dmtdataout_t> template<typename dmtdata_t, typename dmtdataout_t>
void dmt<dmtdata_t, dmtdataout_t>::verify(void) const { void dmt<dmtdata_t, dmtdataout_t>::verify(void) const {
uint32_t num_values = this->size(); uint32_t num_values = this->size();
...@@ -504,7 +447,7 @@ void dmt<dmtdata_t, dmtdataout_t>::verify(void) const { ...@@ -504,7 +447,7 @@ void dmt<dmtdata_t, dmtdataout_t>::verify(void) const {
invariant(pool_used == num_values * align(this->value_length)); invariant(pool_used == num_values * align(this->value_length));
// Array form must have 0 fragmentation in mempool. // Array form must have 0 fragmentation in mempool.
invariant(pool_frag == 0); //TODO: if/when we implement array delete this invariant may need to change. invariant(pool_frag == 0);
} else { } else {
if (this->values_same_size) { if (this->values_same_size) {
// We know exactly how much memory should be used. // We know exactly how much memory should be used.
...@@ -624,13 +567,11 @@ size_t dmt<dmtdata_t, dmtdataout_t>::memory_size(void) { ...@@ -624,13 +567,11 @@ size_t dmt<dmtdata_t, dmtdataout_t>::memory_size(void) {
template<typename dmtdata_t, typename dmtdataout_t> template<typename dmtdata_t, typename dmtdataout_t>
dmt_node_templated<dmtdata_t> & dmt<dmtdata_t, dmtdataout_t>::get_node(const subtree &subtree) const { dmt_node_templated<dmtdata_t> & dmt<dmtdata_t, dmtdataout_t>::get_node(const subtree &subtree) const {
paranoid_invariant(!subtree.is_null()); paranoid_invariant(!subtree.is_null());
return get_node(subtree.get_index()); return get_node(subtree.get_offset());
} }
template<typename dmtdata_t, typename dmtdataout_t> template<typename dmtdata_t, typename dmtdataout_t>
dmt_node_templated<dmtdata_t> & dmt<dmtdata_t, dmtdataout_t>::get_node(const node_idx offset) const { dmt_node_templated<dmtdata_t> & dmt<dmtdata_t, dmtdataout_t>::get_node(const node_offset offset) const {
//TODO: implement
//Need to decide what to do with regards to cnode/dnode
void* ptr = toku_mempool_get_pointer_from_base_and_offset(&this->mp, offset); void* ptr = toku_mempool_get_pointer_from_base_and_offset(&this->mp, offset);
dmt_node *CAST_FROM_VOIDP(node, ptr); dmt_node *CAST_FROM_VOIDP(node, ptr);
return *node; return *node;
...@@ -643,7 +584,7 @@ void dmt<dmtdata_t, dmtdataout_t>::node_set_value(dmt_node * n, const dmtdatain_ ...@@ -643,7 +584,7 @@ void dmt<dmtdata_t, dmtdataout_t>::node_set_value(dmt_node * n, const dmtdatain_
} }
template<typename dmtdata_t, typename dmtdataout_t> template<typename dmtdata_t, typename dmtdataout_t>
node_idx dmt<dmtdata_t, dmtdataout_t>::node_malloc_and_set_value(const dmtdatain_t &value) { node_offset dmt<dmtdata_t, dmtdataout_t>::node_malloc_and_set_value(const dmtdatain_t &value) {
size_t val_size = value.get_dmtdatain_t_size(); size_t val_size = value.get_dmtdatain_t_size();
size_t size_to_alloc = __builtin_offsetof(dmt_node, value) + val_size; size_t size_to_alloc = __builtin_offsetof(dmt_node, value) + val_size;
size_to_alloc = align(size_to_alloc); size_to_alloc = align(size_to_alloc);
...@@ -676,29 +617,31 @@ void dmt<dmtdata_t, dmtdataout_t>::maybe_resize_tree(const dmtdatain_t * value) ...@@ -676,29 +617,31 @@ void dmt<dmtdata_t, dmtdataout_t>::maybe_resize_tree(const dmtdatain_t * value)
const ssize_t need_size = curr_used + add_size; const ssize_t need_size = curr_used + add_size;
paranoid_invariant(need_size <= UINT32_MAX); paranoid_invariant(need_size <= UINT32_MAX);
//TODO(leif) consider different growth rates
const ssize_t new_size = 2*need_size; const ssize_t new_size = 2*need_size;
paranoid_invariant(new_size <= UINT32_MAX); paranoid_invariant(new_size <= UINT32_MAX);
//const uint32_t num_nodes = this->nweight(this->d.t.root);
if ((curr_capacity / 2 >= new_size) || // Way too much allocated if ((curr_capacity / 2 >= new_size) || // Way too much allocated
(curr_free < add_size)) { // No room in mempool (curr_free < add_size)) { // No room in mempool
// Copy all memory and reconstruct dmt in new mempool. // Copy all memory and reconstruct dmt in new mempool.
if (curr_free < add_size && toku_mempool_get_frag_size(&this->mp) == 0) {
// TODO(yoni) or TODO(leif) consider doing this not just when frag size is zero, but also when it is a small percentage of the total mempool size
// Offsets remain the same in the new mempool so we can just realloc.
toku_mempool_realloc_larger(&this->mp, new_size);
} else if (!this->d.t.root.is_null()) {
struct mempool new_kvspace; struct mempool new_kvspace;
struct mempool old_kvspace;
toku_mempool_construct(&new_kvspace, new_size); toku_mempool_construct(&new_kvspace, new_size);
if (!this->d.t.root.is_null()) {
const dmt_node &n = get_node(this->d.t.root); const dmt_node &n = get_node(this->d.t.root);
node_idx *tmp_array; node_offset *tmp_array;
bool malloced = false; bool malloced = false;
tmp_array = alloc_temp_node_idxs(n.weight); tmp_array = alloc_temp_node_offsets(n.weight);
if (!tmp_array) { if (!tmp_array) {
malloced = true; malloced = true;
XMALLOC_N(n.weight, tmp_array); XMALLOC_N(n.weight, tmp_array);
} }
this->fill_array_with_subtree_idxs(tmp_array, this->d.t.root); this->fill_array_with_subtree_offsets(tmp_array, this->d.t.root);
for (node_idx i = 0; i < n.weight; i++) { for (node_offset i = 0; i < n.weight; i++) {
dmt_node &node = get_node(tmp_array[i]); dmt_node &node = get_node(tmp_array[i]);
const size_t bytes_to_copy = __builtin_offsetof(dmt_node, value) + node.value_length; const size_t bytes_to_copy = __builtin_offsetof(dmt_node, value) + node.value_length;
const size_t bytes_to_alloc = align(bytes_to_copy); const size_t bytes_to_alloc = align(bytes_to_copy);
...@@ -707,15 +650,14 @@ void dmt<dmtdata_t, dmtdataout_t>::maybe_resize_tree(const dmtdatain_t * value) ...@@ -707,15 +650,14 @@ void dmt<dmtdata_t, dmtdataout_t>::maybe_resize_tree(const dmtdatain_t * value)
tmp_array[i] = toku_mempool_get_offset_from_pointer_and_base(&new_kvspace, newdata); tmp_array[i] = toku_mempool_get_offset_from_pointer_and_base(&new_kvspace, newdata);
} }
old_kvspace = this->mp; struct mempool old_kvspace = this->mp;
this->mp = new_kvspace; this->mp = new_kvspace;
this->rebuild_subtree_from_idxs(&this->d.t.root, tmp_array, n.weight); this->rebuild_subtree_from_offsets(&this->d.t.root, tmp_array, n.weight);
if (malloced) toku_free(tmp_array); if (malloced) toku_free(tmp_array);
toku_mempool_destroy(&old_kvspace); toku_mempool_destroy(&old_kvspace);
} } else {
else {
toku_mempool_destroy(&this->mp); toku_mempool_destroy(&this->mp);
this->mp = new_kvspace; toku_mempool_construct(&this->mp, new_size);
} }
} }
} }
...@@ -737,12 +679,12 @@ template<typename dmtdata_t, typename dmtdataout_t> ...@@ -737,12 +679,12 @@ template<typename dmtdata_t, typename dmtdataout_t>
void dmt<dmtdata_t, dmtdataout_t>::insert_internal(subtree *const subtreep, const dmtdatain_t &value, const uint32_t idx, subtree **const rebalance_subtree) { void dmt<dmtdata_t, dmtdataout_t>::insert_internal(subtree *const subtreep, const dmtdatain_t &value, const uint32_t idx, subtree **const rebalance_subtree) {
if (subtreep->is_null()) { if (subtreep->is_null()) {
paranoid_invariant_zero(idx); paranoid_invariant_zero(idx);
const node_idx newidx = this->node_malloc_and_set_value(value); const node_offset newoffset = this->node_malloc_and_set_value(value);
dmt_node &newnode = get_node(newidx); dmt_node &newnode = get_node(newoffset);
newnode.weight = 1; newnode.weight = 1;
newnode.left.set_to_null(); newnode.left.set_to_null();
newnode.right.set_to_null(); newnode.right.set_to_null();
subtreep->set_index(newidx); subtreep->set_offset(newoffset);
} else { } else {
dmt_node &n = get_node(*subtreep); dmt_node &n = get_node(*subtreep);
n.weight++; n.weight++;
...@@ -777,12 +719,13 @@ void dmt<dmtdata_t, dmtdataout_t>::delete_internal(subtree *const subtreep, cons ...@@ -777,12 +719,13 @@ void dmt<dmtdata_t, dmtdataout_t>::delete_internal(subtree *const subtreep, cons
} else if (idx == leftweight) { } else if (idx == leftweight) {
// Found the correct index. // Found the correct index.
if (n.left.is_null()) { if (n.left.is_null()) {
paranoid_invariant_zero(idx);
// Delete n and let parent point to n.right // Delete n and let parent point to n.right
subtree ptr_this = *subtreep; subtree ptr_this = *subtreep;
*subtreep = n.right; *subtreep = n.right;
subtree to_free; subtree to_free;
if (subtree_replace != nullptr) { if (subtree_replace != nullptr) {
// Swap self with the other node. // Swap self with the other node. Taking over all responsibility.
to_free = *subtree_replace; to_free = *subtree_replace;
dmt_node &ancestor = get_node(*subtree_replace); dmt_node &ancestor = get_node(*subtree_replace);
if (*rebalance_subtree == &ancestor.right) { if (*rebalance_subtree == &ancestor.right) {
...@@ -801,6 +744,7 @@ void dmt<dmtdata_t, dmtdataout_t>::delete_internal(subtree *const subtreep, cons ...@@ -801,6 +744,7 @@ void dmt<dmtdata_t, dmtdataout_t>::delete_internal(subtree *const subtreep, cons
// Delete n and let parent point to n.left // Delete n and let parent point to n.left
subtree to_free = *subtreep; subtree to_free = *subtreep;
*subtreep = n.left; *subtreep = n.left;
paranoid_invariant(idx>0);
paranoid_invariant_null(subtree_replace); // To be recursive, we're looking for index 0. n is index > 0 here. paranoid_invariant_null(subtree_replace); // To be recursive, we're looking for index 0. n is index > 0 here.
this->node_free(to_free); this->node_free(to_free);
} else { } else {
...@@ -913,36 +857,37 @@ void dmt<dmtdata_t, dmtdataout_t>::fetch_internal(const subtree &subtree, const ...@@ -913,36 +857,37 @@ void dmt<dmtdata_t, dmtdataout_t>::fetch_internal(const subtree &subtree, const
} }
template<typename dmtdata_t, typename dmtdataout_t> template<typename dmtdata_t, typename dmtdataout_t>
void dmt<dmtdata_t, dmtdataout_t>::fill_array_with_subtree_idxs(node_idx *const array, const subtree &subtree) const { void dmt<dmtdata_t, dmtdataout_t>::fill_array_with_subtree_offsets(node_offset *const array, const subtree &subtree) const {
if (!subtree.is_null()) { if (!subtree.is_null()) {
const dmt_node &tree = get_node(subtree); const dmt_node &tree = get_node(subtree);
this->fill_array_with_subtree_idxs(&array[0], tree.left); this->fill_array_with_subtree_offsets(&array[0], tree.left);
array[this->nweight(tree.left)] = subtree.get_index(); array[this->nweight(tree.left)] = subtree.get_offset();
this->fill_array_with_subtree_idxs(&array[this->nweight(tree.left) + 1], tree.right); this->fill_array_with_subtree_offsets(&array[this->nweight(tree.left) + 1], tree.right);
} }
} }
template<typename dmtdata_t, typename dmtdataout_t> template<typename dmtdata_t, typename dmtdataout_t>
void dmt<dmtdata_t, dmtdataout_t>::rebuild_subtree_from_idxs(subtree *const subtree, const node_idx *const idxs, const uint32_t numvalues) { void dmt<dmtdata_t, dmtdataout_t>::rebuild_subtree_from_offsets(subtree *const subtree, const node_offset *const offsets, const uint32_t numvalues) {
if (numvalues==0) { if (numvalues==0) {
subtree->set_to_null(); subtree->set_to_null();
} else { } else {
uint32_t halfway = numvalues/2; uint32_t halfway = numvalues/2;
subtree->set_index(idxs[halfway]); subtree->set_offset(offsets[halfway]);
dmt_node &newnode = get_node(idxs[halfway]); dmt_node &newnode = get_node(offsets[halfway]);
newnode.weight = numvalues; newnode.weight = numvalues;
// value is already in there. // value is already in there.
this->rebuild_subtree_from_idxs(&newnode.left, &idxs[0], halfway); this->rebuild_subtree_from_offsets(&newnode.left, &offsets[0], halfway);
this->rebuild_subtree_from_idxs(&newnode.right, &idxs[halfway+1], numvalues-(halfway+1)); this->rebuild_subtree_from_offsets(&newnode.right, &offsets[halfway+1], numvalues-(halfway+1));
} }
} }
//TODO(leif): Note that this can mess with our memory_footprint calculation (we may touch past what is marked as 'used' in the mempool)
template<typename dmtdata_t, typename dmtdataout_t> template<typename dmtdata_t, typename dmtdataout_t>
node_idx* dmt<dmtdata_t, dmtdataout_t>::alloc_temp_node_idxs(uint32_t num_idxs) { node_offset* dmt<dmtdata_t, dmtdataout_t>::alloc_temp_node_offsets(uint32_t num_offsets) {
size_t mem_needed = num_idxs * sizeof(node_idx); size_t mem_needed = num_offsets * sizeof(node_offset);
size_t mem_free; size_t mem_free;
mem_free = toku_mempool_get_free_space(&this->mp); mem_free = toku_mempool_get_free_space(&this->mp);
node_idx* CAST_FROM_VOIDP(tmp, toku_mempool_get_next_free_ptr(&this->mp)); node_offset* CAST_FROM_VOIDP(tmp, toku_mempool_get_next_free_ptr(&this->mp));
if (mem_free >= mem_needed) { if (mem_free >= mem_needed) {
return tmp; return tmp;
} }
...@@ -951,36 +896,28 @@ node_idx* dmt<dmtdata_t, dmtdataout_t>::alloc_temp_node_idxs(uint32_t num_idxs) ...@@ -951,36 +896,28 @@ node_idx* dmt<dmtdata_t, dmtdataout_t>::alloc_temp_node_idxs(uint32_t num_idxs)
template<typename dmtdata_t, typename dmtdataout_t> template<typename dmtdata_t, typename dmtdataout_t>
void dmt<dmtdata_t, dmtdataout_t>::rebalance(subtree *const subtree) { void dmt<dmtdata_t, dmtdataout_t>::rebalance(subtree *const subtree) {
(void) subtree;
paranoid_invariant(!subtree->is_null()); paranoid_invariant(!subtree->is_null());
node_idx idx = subtree->get_index();
//TODO: restore optimization, maybe only if values_same_size // There is a possible "optimization" here:
#if 0 // if (this->values_same_size && subtree == &this->d.t.root) {
if (!dynamic && idx==this->d.t.root.get_index()) { // this->convert_from_tree_to_array();
//Try to convert to an array. // return;
//If this fails, (malloc) nothing will have changed. // }
//In the failure case we continue on to the standard rebalance // but we don't want to do it because it involves actually copying values around
//algorithm. // as opposed to stopping in the middle of rebalancing (like in the OMT)
this->convert_to_array();
if (supports_marks) { node_offset offset = subtree->get_offset();
this->convert_to_tree(); const dmt_node &n = get_node(offset);
} node_offset *tmp_array;
} else {
#endif
const dmt_node &n = get_node(idx);
node_idx *tmp_array;
bool malloced = false; bool malloced = false;
tmp_array = alloc_temp_node_idxs(n.weight); tmp_array = alloc_temp_node_offsets(n.weight);
if (!tmp_array) { if (!tmp_array) {
malloced = true; malloced = true;
XMALLOC_N(n.weight, tmp_array); XMALLOC_N(n.weight, tmp_array);
} }
this->fill_array_with_subtree_idxs(tmp_array, *subtree); this->fill_array_with_subtree_offsets(tmp_array, *subtree);
this->rebuild_subtree_from_idxs(subtree, tmp_array, n.weight); this->rebuild_subtree_from_offsets(subtree, tmp_array, n.weight);
if (malloced) toku_free(tmp_array); if (malloced) toku_free(tmp_array);
#if 0
}
#endif
} }
template<typename dmtdata_t, typename dmtdataout_t> template<typename dmtdata_t, typename dmtdataout_t>
...@@ -1207,22 +1144,22 @@ bool dmt<dmtdata_t, dmtdataout_t>::value_length_is_fixed(void) const { ...@@ -1207,22 +1144,22 @@ bool dmt<dmtdata_t, dmtdataout_t>::value_length_is_fixed(void) const {
template<typename dmtdata_t, typename dmtdataout_t> template<typename dmtdata_t, typename dmtdataout_t>
void dmt<dmtdata_t, dmtdataout_t>::serialize_values(uint32_t expected_unpadded_memory, struct wbuf *wb) const { void dmt<dmtdata_t, dmtdataout_t>::serialize_values(uint32_t expected_unpadded_memory, struct wbuf *wb) const {
invariant(this->is_array); invariant(this->is_array);
invariant(this->values_same_size);
const uint8_t pad_bytes = get_fixed_length_alignment_overhead(); const uint8_t pad_bytes = get_fixed_length_alignment_overhead();
const uint32_t fixed_len = this->value_length; const uint32_t fixed_len = this->value_length;
const uint32_t fixed_aligned_len = align(this->value_length); const uint32_t fixed_aligned_len = align(this->value_length);
paranoid_invariant(expected_unpadded_memory == this->d.a.num_values * this->value_length); paranoid_invariant(expected_unpadded_memory == this->d.a.num_values * this->value_length);
paranoid_invariant(toku_mempool_get_used_space(&this->mp) >= paranoid_invariant(toku_mempool_get_used_space(&this->mp) >=
expected_unpadded_memory + pad_bytes * this->d.a.num_values + expected_unpadded_memory + pad_bytes * this->d.a.num_values);
this->d.a.start_idx * fixed_aligned_len);
if (this->d.a.num_values == 0) { if (this->d.a.num_values == 0) {
// Nothing to serialize // Nothing to serialize
} else if (pad_bytes == 0) { } else if (pad_bytes == 0) {
// Basically a memcpy // Basically a memcpy
wbuf_nocrc_literal_bytes(wb, get_array_value(0), expected_unpadded_memory); wbuf_nocrc_literal_bytes(wb, get_array_value(0), expected_unpadded_memory);
} else { } else {
uint8_t* dest = wbuf_nocrc_reserve_literal_bytes(wb, expected_unpadded_memory); uint8_t* const dest = wbuf_nocrc_reserve_literal_bytes(wb, expected_unpadded_memory);
uint8_t* src = reinterpret_cast<uint8_t*>(get_array_value(0)); const uint8_t* const src = reinterpret_cast<uint8_t*>(get_array_value(0));
paranoid_invariant(this->d.a.num_values*fixed_len == expected_unpadded_memory); //TODO(leif) maybe look at vectorization here
for (uint32_t i = 0; i < this->d.a.num_values; i++) { for (uint32_t i = 0; i < this->d.a.num_values; i++) {
memcpy(&dest[i*fixed_len], &src[i*fixed_aligned_len], fixed_len); memcpy(&dest[i*fixed_len], &src[i*fixed_aligned_len], fixed_len);
} }
...@@ -1233,9 +1170,10 @@ template<typename dmtdata_t, typename dmtdataout_t> ...@@ -1233,9 +1170,10 @@ template<typename dmtdata_t, typename dmtdataout_t>
void dmt<dmtdata_t, dmtdataout_t>::builder::create(uint32_t _max_values, uint32_t _max_value_bytes) { void dmt<dmtdata_t, dmtdataout_t>::builder::create(uint32_t _max_values, uint32_t _max_value_bytes) {
this->max_values = _max_values; this->max_values = _max_values;
this->max_value_bytes = _max_value_bytes; this->max_value_bytes = _max_value_bytes;
this->temp.create_no_array(); this->temp.create();
paranoid_invariant_null(toku_mempool_get_base(&this->temp.mp));
this->temp_valid = true; this->temp_valid = true;
this->sorted_nodes = nullptr; this->sorted_node_offsets = nullptr;
// Include enough space for alignment padding // Include enough space for alignment padding
size_t initial_space = (ALIGNMENT - 1) * _max_values + _max_value_bytes; size_t initial_space = (ALIGNMENT - 1) * _max_values + _max_value_bytes;
...@@ -1251,8 +1189,8 @@ void dmt<dmtdata_t, dmtdataout_t>::builder::append(const dmtdatain_t &value) { ...@@ -1251,8 +1189,8 @@ void dmt<dmtdata_t, dmtdataout_t>::builder::append(const dmtdatain_t &value) {
return; return;
} }
if (this->temp.is_array) { if (this->temp.is_array) {
// Convert to tree format // Convert to tree format (without weights and linkage)
XMALLOC_N(this->max_values, this->sorted_nodes); XMALLOC_N(this->max_values, this->sorted_node_offsets);
// Include enough space for alignment padding // Include enough space for alignment padding
size_t mem_needed = (ALIGNMENT - 1 + __builtin_offsetof(dmt_node, value)) * max_values + max_value_bytes; size_t mem_needed = (ALIGNMENT - 1 + __builtin_offsetof(dmt_node, value)) * max_values + max_value_bytes;
...@@ -1261,19 +1199,18 @@ void dmt<dmtdata_t, dmtdataout_t>::builder::append(const dmtdatain_t &value) { ...@@ -1261,19 +1199,18 @@ void dmt<dmtdata_t, dmtdataout_t>::builder::append(const dmtdatain_t &value) {
const uint32_t num_values = this->temp.d.a.num_values; const uint32_t num_values = this->temp.d.a.num_values;
toku_mempool_construct(&this->temp.mp, mem_needed); toku_mempool_construct(&this->temp.mp, mem_needed);
// Copy over and get node_idxs // Copy over and get node_offsets
for (uint32_t i = 0; i < num_values; i++) { for (uint32_t i = 0; i < num_values; i++) {
dmtdatain_t functor(this->temp.value_length, this->temp.get_array_value_internal(&old_mp, i)); dmtdatain_t functor(this->temp.value_length, this->temp.get_array_value_internal(&old_mp, i));
this->sorted_nodes[i] = this->temp.node_malloc_and_set_value(functor); this->sorted_node_offsets[i] = this->temp.node_malloc_and_set_value(functor);
} }
this->temp.is_array = false; this->temp.is_array = false;
this->temp.values_same_size = false; this->temp.values_same_size = false;
this->temp.value_length = 0;
toku_mempool_destroy(&old_mp); toku_mempool_destroy(&old_mp);
} }
paranoid_invariant(!this->temp.is_array); paranoid_invariant(!this->temp.is_array);
this->temp.values_same_size = false; this->sorted_node_offsets[this->temp.d.a.num_values++] = this->temp.node_malloc_and_set_value(value);
// Insert dynamic.
this->sorted_nodes[this->temp.d.a.num_values++] = this->temp.node_malloc_and_set_value(value);
} }
template<typename dmtdata_t, typename dmtdataout_t> template<typename dmtdata_t, typename dmtdataout_t>
...@@ -1286,22 +1223,30 @@ template<typename dmtdata_t, typename dmtdataout_t> ...@@ -1286,22 +1223,30 @@ template<typename dmtdata_t, typename dmtdataout_t>
void dmt<dmtdata_t, dmtdataout_t>::builder::build(dmt<dmtdata_t, dmtdataout_t> *dest) { void dmt<dmtdata_t, dmtdataout_t>::builder::build(dmt<dmtdata_t, dmtdataout_t> *dest) {
invariant(this->temp_valid); invariant(this->temp_valid);
//NOTE: Always use d.a.num_values for size because we have not yet created root. //NOTE: Always use d.a.num_values for size because we have not yet created root.
invariant(this->temp.d.a.num_values == this->max_values); // Optionally make it <= invariant(this->temp.d.a.num_values <= this->max_values);
// Memory invariant is taken care of incrementally // Memory invariant is taken care of incrementally (during append())
if (!this->temp.is_array) { if (!this->temp.is_array) {
invariant_notnull(this->sorted_nodes); invariant_notnull(this->sorted_node_offsets);
this->temp.rebuild_subtree_from_idxs(&this->temp.d.t.root, this->sorted_nodes, this->temp.d.a.num_values); this->temp.rebuild_subtree_from_offsets(&this->temp.d.t.root, this->sorted_node_offsets, this->temp.d.a.num_values);
toku_free(this->sorted_nodes); toku_free(this->sorted_node_offsets);
this->sorted_nodes = nullptr; this->sorted_node_offsets = nullptr;
} }
paranoid_invariant_null(this->sorted_nodes); paranoid_invariant_null(this->sorted_node_offsets);
size_t used = toku_mempool_get_used_space(&this->temp.mp); const size_t used = toku_mempool_get_used_space(&this->temp.mp);
size_t allocated = toku_mempool_get_size(&this->temp.mp); const size_t allocated = toku_mempool_get_size(&this->temp.mp);
size_t max_allowed = used + used / 4; // We want to use no more than (about) the actual used space + 25% overhead for mempool growth.
size_t footprint = toku_mempool_footprint(&this->temp.mp); // When we know the elements are fixed-length, we use the better dmt constructor.
if (allocated > max_allowed && footprint > max_allowed) { // In practice, as of Jan 2014, we use the builder in two cases:
// - When we know the elements are not fixed-length.
// - During upgrade of a pre version 25 basement node.
// During upgrade, we will probably wildly overallocate because we don't account for the values that aren't stored in the dmt, so here we want to shrink the mempool.
// When we know the elements are not fixed-length, we still know how much memory they occupy in total, modulo alignment, so we want to allow for mempool overhead and worst-case alignment overhead, and not shrink the mempool.
const size_t max_allowed = used + (ALIGNMENT-1) * this->temp.size();
const size_t max_allowed_with_mempool_overhead = max_allowed + max_allowed / 4;
//TODO(leif): get footprint calculation correct (under jemalloc) and add some form of footprint constraint
if (allocated > max_allowed_with_mempool_overhead) {
// Reallocate smaller mempool to save memory // Reallocate smaller mempool to save memory
invariant_zero(toku_mempool_get_frag_size(&this->temp.mp)); invariant_zero(toku_mempool_get_frag_size(&this->temp.mp));
struct mempool new_mp; struct mempool new_mp;
......
util/dmt.h
View file @ 3b923630
...@@ -98,7 +98,7 @@ PATENT RIGHTS GRANT: ...@@ -98,7 +98,7 @@ PATENT RIGHTS GRANT:
#include <vector> #include <vector>
namespace toku { namespace toku {
typedef uint32_t node_idx; typedef uint32_t node_offset;
/** /**
...@@ -170,14 +170,14 @@ class subtree { ...@@ -170,14 +170,14 @@ class subtree {
} }
inline bool is_null(void) const { inline bool is_null(void) const {
return NODE_NULL == this->get_index(); return NODE_NULL == this->get_offset();
} }
inline node_idx get_index(void) const { inline node_offset get_offset(void) const {
return m_index; return m_index;
} }
inline void set_index(node_idx index) { inline void set_offset(node_offset index) {
paranoid_invariant(index != NODE_NULL); paranoid_invariant(index != NODE_NULL);
m_index = index; m_index = index;
} }
...@@ -231,7 +231,7 @@ class dmt { ...@@ -231,7 +231,7 @@ class dmt {
private: private:
uint32_t max_values; uint32_t max_values;
uint32_t max_value_bytes; uint32_t max_value_bytes;
node_idx *sorted_nodes; node_offset *sorted_node_offsets;
bool temp_valid; bool temp_valid;
dmt<dmtdata_t, dmtdataout_t> temp; dmt<dmtdata_t, dmtdataout_t> temp;
}; };
...@@ -242,13 +242,6 @@ class dmt { ...@@ -242,13 +242,6 @@ class dmt {
*/ */
void create(void); void create(void);
/**
* Effect: Create an empty DMT with no internal allocated space.
* Performance: constant time.
* Rationale: In some cases we need a valid dmt but don't want to malloc.
*/
void create_no_array(void);
/** /**
* Effect: Create a DMT containing values. The number of values is in numvalues. * Effect: Create a DMT containing values. The number of values is in numvalues.
* Requires: this has not been created yet * Requires: this has not been created yet
...@@ -517,7 +510,6 @@ class dmt { ...@@ -517,7 +510,6 @@ class dmt {
ENSURE_POD(dmt_node); ENSURE_POD(dmt_node);
struct dmt_array { struct dmt_array {
uint32_t start_idx;
uint32_t num_values; uint32_t num_values;
}; };
...@@ -525,6 +517,7 @@ class dmt { ...@@ -525,6 +517,7 @@ class dmt {
subtree root; subtree root;
}; };
// TODO(yoni): comment relationship between values_same_size, num values, value_length, note about tree still keeping track (until lost once)
bool values_same_size; bool values_same_size;
uint32_t value_length; uint32_t value_length;
struct mempool mp; struct mempool mp;
...@@ -538,21 +531,19 @@ class dmt { ...@@ -538,21 +531,19 @@ class dmt {
void verify_internal(const subtree &subtree, std::vector<bool> *touched) const; void verify_internal(const subtree &subtree, std::vector<bool> *touched) const;
void create_internal_no_alloc(bool as_tree);
dmt_node & get_node(const subtree &subtree) const; dmt_node & get_node(const subtree &subtree) const;
dmt_node & get_node(const node_idx offset) const; dmt_node & get_node(const node_offset offset) const;
uint32_t nweight(const subtree &subtree) const; uint32_t nweight(const subtree &subtree) const;
node_idx node_malloc_and_set_value(const dmtdatain_t &value); node_offset node_malloc_and_set_value(const dmtdatain_t &value);
void node_set_value(dmt_node *n, const dmtdatain_t &value); void node_set_value(dmt_node *n, const dmtdatain_t &value);
void node_free(const subtree &st); void node_free(const subtree &st);
void maybe_resize_array(const int change); void maybe_resize_array_for_insert(void);
void convert_to_tree(void); void convert_to_tree(void);
...@@ -566,15 +557,11 @@ class dmt { ...@@ -566,15 +557,11 @@ class dmt {
template<bool with_resize> template<bool with_resize>
int insert_at_array_end(const dmtdatain_t& value_in); int insert_at_array_end(const dmtdatain_t& value_in);
int insert_at_array_beginning(const dmtdatain_t& value_in);
dmtdata_t * alloc_array_value_end(void); dmtdata_t * alloc_array_value_end(void);
dmtdata_t * alloc_array_value_beginning(void);
dmtdata_t * get_array_value(const uint32_t idx) const; dmtdata_t * get_array_value(const uint32_t idx) const;
dmtdata_t * get_array_value_internal(const struct mempool *mempool, const uint32_t real_idx) const; dmtdata_t * get_array_value_internal(const struct mempool *mempool, const uint32_t idx) const;
void convert_from_array_to_tree(void); void convert_from_array_to_tree(void);
...@@ -610,10 +597,10 @@ class dmt { ...@@ -610,10 +597,10 @@ class dmt {
void fetch_internal(const subtree &subtree, const uint32_t i, uint32_t *const value_len, dmtdataout_t *const value) const; void fetch_internal(const subtree &subtree, const uint32_t i, uint32_t *const value_len, dmtdataout_t *const value) const;
__attribute__((nonnull)) __attribute__((nonnull))
void fill_array_with_subtree_idxs(node_idx *const array, const subtree &subtree) const; void fill_array_with_subtree_offsets(node_offset *const array, const subtree &subtree) const;
__attribute__((nonnull)) __attribute__((nonnull))
void rebuild_subtree_from_idxs(subtree *const subtree, const node_idx *const idxs, const uint32_t numvalues); void rebuild_subtree_from_offsets(subtree *const subtree, const node_offset *const offsets, const uint32_t numvalues);
__attribute__((nonnull)) __attribute__((nonnull))
void rebalance(subtree *const subtree); void rebalance(subtree *const subtree);
...@@ -654,7 +641,7 @@ class dmt { ...@@ -654,7 +641,7 @@ class dmt {
int (*h)(const uint32_t, const dmtdata_t &, const dmtcmp_t &)> int (*h)(const uint32_t, const dmtdata_t &, const dmtcmp_t &)>
int find_internal_minus(const subtree &subtree, const dmtcmp_t &extra, uint32_t *const value_len, dmtdataout_t *const value, uint32_t *const idxp) const; int find_internal_minus(const subtree &subtree, const dmtcmp_t &extra, uint32_t *const value_len, dmtdataout_t *const value, uint32_t *const idxp) const;
node_idx* alloc_temp_node_idxs(uint32_t num_idxs); node_offset* alloc_temp_node_offsets(uint32_t num_idxs);
uint32_t align(const uint32_t x) const; uint32_t align(const uint32_t x) const;
}; };
......
util/mempool.cc
View file @ 3b923630
...@@ -142,6 +142,11 @@ void toku_mempool_construct(struct mempool *mp, size_t data_size) { ...@@ -142,6 +142,11 @@ void toku_mempool_construct(struct mempool *mp, size_t data_size) {
} }
} }
void toku_mempool_reset(struct mempool *mp) {
mp->free_offset = 0;
mp->frag_size = 0;
}
void toku_mempool_realloc_larger(struct mempool *mp, size_t data_size) { void toku_mempool_realloc_larger(struct mempool *mp, size_t data_size) {
invariant(data_size > mp->free_offset); invariant(data_size > mp->free_offset);
size_t mpsize = data_size + (data_size/4); // allow 1/4 room for expansion (would be wasted if read-only) size_t mpsize = data_size + (data_size/4); // allow 1/4 room for expansion (would be wasted if read-only)
......
util/mempool.h
View file @ 3b923630
...@@ -123,6 +123,10 @@ void toku_mempool_init(struct mempool *mp, void *base, size_t free_offset, size_ ...@@ -123,6 +123,10 @@ void toku_mempool_init(struct mempool *mp, void *base, size_t free_offset, size_
*/ */
void toku_mempool_construct(struct mempool *mp, size_t data_size); void toku_mempool_construct(struct mempool *mp, size_t data_size);
/* treat mempool as if it has just been created; ignore any frag and start allocating from beginning again.
*/
void toku_mempool_reset(struct mempool *mp);
/* reallocate memory for construct mempool /* reallocate memory for construct mempool
*/ */
void toku_mempool_realloc_larger(struct mempool *mp, size_t data_size); void toku_mempool_realloc_larger(struct mempool *mp, size_t data_size);
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7