Hawkin
/
arangodb
mirror of https://gitee.com/bigwinds/arangodb
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
arangodb/arangod/RocksDBEngine/RocksDBCuckooIndexEstimator.h

949 lines
32 KiB
C++
Raw Permalink Blame History

////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2017 ArangoDB GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
/// you may not use this file except in compliance with the License.
/// You may obtain a copy of the License at
///
/// http://www.apache.org/licenses/LICENSE-2.0
///
/// Unless required by applicable law or agreed to in writing, software
/// distributed under the License is distributed on an "AS IS" BASIS,
/// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
/// See the License for the specific language governing permissions and
/// limitations under the License.
///
/// Copyright holder is ArangoDB GmbH, Cologne, Germany
///
/// @author Daniel Larkin
/// @author Michael Hackstein
////////////////////////////////////////////////////////////////////////////////
#ifndef ARANGOD_ROCKSDB_ROCKSDB_INDEX_ESTIMATOR_H
#define ARANGOD_ROCKSDB_ROCKSDB_INDEX_ESTIMATOR_H 1
#include <set>
#include <map>
#include "Basics/Exceptions.h"
#include "Basics/ReadLocker.h"
#include "Basics/ReadWriteLock.h"
#include "Basics/WriteLocker.h"
#include "Basics/fasthash.h"
#include "Logger/Logger.h"
#include "RocksDBEngine/RocksDBFormat.h"
#include <rocksdb/types.h>
#include <velocypack/StringRef.h>
// In the following template:
// Key is the key type, it must be copyable and movable, furthermore, Key
// must be default constructible (without arguments) as empty and
// must have an empty() method to indicate that the instance is
// empty.
// If using fasthash64 on all bytes of the object is not
// a suitable hash function, one has to instanciate the template
// with two hash function types as 3rd and 4th argument. If
// std::equal_to<Key> is not implemented or does not behave correctly,
// one has to supply a comparison class as well.
// This class is not thread-safe!
namespace arangodb {
// C++ wrapper for the hash function:
template <class T, uint64_t Seed>
class HashWithSeed {
public:
uint64_t operator()(T const& t) const noexcept {
// Some implementation like Fnv or xxhash looking at bytes in type T,
// taking the seed into account.
auto p = reinterpret_cast<void const*>(&t);
return fasthash64(p, sizeof(T), Seed);
}
};
template <class Key, class HashKey = HashWithSeed<Key, 0xdeadbeefdeadbeefULL>,
class Fingerprint = HashWithSeed<Key, 0xabcdefabcdef1234ULL>,
class HashShort = HashWithSeed<uint16_t, 0xfedcbafedcba4321ULL>, class CompKey = std::equal_to<Key>>
class RocksDBCuckooIndexEstimator {
// Note that the following has to be a power of two and at least 4!
static constexpr uint32_t kSlotsPerBucket = 4;
// total size of a slot
static constexpr size_t kSlotSize = sizeof(uint16_t);
// total size of a counter
static constexpr size_t kCounterSize = sizeof(uint32_t);
// maximum number of cuckoo rounds on insertion
static constexpr unsigned kMaxRounds = 16;
private:
// Helper class to hold the finger prints.
// Fingerprints are 64bit aligned and we
// have 4 slots per 64bit bucket.
struct Slot {
private:
uint16_t* _data;
uint32_t* _counter;
public:
explicit Slot(uint16_t* data) : _data(data), _counter(nullptr) {}
~Slot() {
// Not responsible for anything
}
bool operator==(const Slot& other) const { return _data == other._data; }
uint16_t* fingerprint() const {
TRI_ASSERT(_data != nullptr);
return _data;
}
uint32_t* counter() const {
TRI_ASSERT(_counter != nullptr);
return _counter;
}
void reset() {
*fingerprint() = 0;
*counter() = 0;
}
bool isEqual(uint16_t fp) const { return ((*fingerprint()) == fp); }
bool isEmpty() const { return (*fingerprint()) == 0; }
// If this returns FALSE we have removed the
// last element => we need to remove the fingerprint as well.
bool decrease() {
if (*counter() > 1) {
(*counter())--;
return true;
}
return false;
}
void increase() {
if ((*counter()) < UINT32_MAX) {
(*counter())++;
}
}
void init(uint16_t fp) {
// This is the first element
*fingerprint() = fp;
*counter() = 1;
}
void swap(uint16_t& fp, uint32_t& cnt) {
std::swap(*fingerprint(), fp);
std::swap(*counter(), cnt);
}
void injectCounter(uint32_t* cnt) { _counter = cnt; }
};
enum SerializeFormat : char {
// To describe this format we use | as a seperator for readability, but it
// is NOT a printed character in the serialized string
// NOCOMPRESSION:
// type|length|size|nrUsed|nrCuckood|nrTotal|niceSize|logSize|base|counters
NOCOMPRESSION = '1'
};
public:
static bool isFormatSupported(arangodb::velocypack::StringRef serialized) {
TRI_ASSERT(serialized.size() > sizeof(_appliedSeq) + sizeof(char));
switch (serialized[sizeof(_appliedSeq)]) {
case SerializeFormat::NOCOMPRESSION:
return true;
}
return false;
}
RocksDBCuckooIndexEstimator(uint64_t size)
: _randState(0x2636283625154737ULL),
_logSize(0),
_size(0),
_niceSize(0),
_sizeMask(0),
_sizeShift(0),
_slotAllocSize(0),
_counterAllocSize(0),
_base(nullptr),
_slotBase(nullptr),
_counters(nullptr),
_counterBase(nullptr),
_nrUsed(0),
_nrCuckood(0),
_nrTotal(0),
_appliedSeq(0),
_needToPersist(false) {
// Inflate size so that we have some padding to avoid failure
size *= 2;
size = (size >= 1024) ? size : 1024; // want 256 buckets minimum
// First find the smallest power of two that is not smaller than size:
size /= kSlotsPerBucket;
_size = size;
initializeDefault();
}
RocksDBCuckooIndexEstimator(arangodb::velocypack::StringRef const serialized)
: _randState(0x2636283625154737ULL),
_logSize(0),
_size(0),
_niceSize(0),
_sizeMask(0),
_sizeShift(0),
_slotAllocSize(0),
_counterAllocSize(0),
_base(nullptr),
_slotBase(nullptr),
_counters(nullptr),
_counterBase(nullptr),
_nrUsed(0),
_nrCuckood(0),
_nrTotal(0),
_appliedSeq(0),
_needToPersist(false) {
switch (serialized[sizeof(_appliedSeq)]) {
case SerializeFormat::NOCOMPRESSION: {
deserializeUncompressed(serialized);
break;
}
default: {
LOG_TOPIC("bcd09", WARN, arangodb::Logger::ENGINES)
<< "unable to restore index estimates: invalid format found";
// Do not construct from serialization, use other constructor instead
THROW_ARANGO_EXCEPTION_MESSAGE(
TRI_ERROR_INTERNAL,
"unable to restore index estimates: invalid format found");
}
}
}
~RocksDBCuckooIndexEstimator() {
delete[] _slotBase;
delete[] _counterBase;
}
RocksDBCuckooIndexEstimator(RocksDBCuckooIndexEstimator const&) = delete;
RocksDBCuckooIndexEstimator(RocksDBCuckooIndexEstimator&&) = delete;
RocksDBCuckooIndexEstimator& operator=(RocksDBCuckooIndexEstimator const&) = delete;
RocksDBCuckooIndexEstimator& operator=(RocksDBCuckooIndexEstimator&&) = delete;
/**
* @brief Serialize estimator for persistence, applying any buffered updates
*
* Format is hard-coded, and must support older formats for backwards
* compatibility. The first 8 bytes consist of a sequence number S. All
* updates prior to and including S are reflected in the serialization. Any
* updates after S must be kept in the WALs and "replayed" during recovery.
*
* Applies any buffered updates and updates the "committed" seq/tick state.
*
* @param serialized String for output
* @param commitSeq Above that are still uncommited operations
*/
void serialize(std::string& serialized, rocksdb::SequenceNumber maxCommitSeq) {
// We always have to start with the commit seq, type and then the length
// commit seq, above that is an uncommited operations
// rocksdb::SequenceNumber commitSeq = committableSeq();
// must apply updates first to be valid, WAL needs to preserve
rocksdb::SequenceNumber appliedSeq = applyUpdates(maxCommitSeq);
TRI_ASSERT(appliedSeq <= maxCommitSeq);
{
// Sorry we need a consistent state, so we have to read-lock
READ_LOCKER(locker, _lock);
appliedSeq = std::max(appliedSeq, _appliedSeq.load(std::memory_order_acquire));
TRI_ASSERT(appliedSeq != std::numeric_limits<rocksdb::SequenceNumber>::max());
rocksutils::uint64ToPersistent(serialized, appliedSeq);
// type
serialized += SerializeFormat::NOCOMPRESSION;
// length
uint64_t serialLength =
(sizeof(SerializeFormat) + sizeof(uint64_t) + sizeof(_size) +
sizeof(_nrUsed) + sizeof(_nrCuckood) + sizeof(_nrTotal) + sizeof(_niceSize) +
sizeof(_logSize) + (_size * kSlotSize * kSlotsPerBucket)) +
(_size * kCounterSize * kSlotsPerBucket);
serialized.reserve(sizeof(uint64_t) + serialLength);
// We always prepend the length, so parsing is easier
rocksutils::uint64ToPersistent(serialized, serialLength);
// Add all member variables
rocksutils::uint64ToPersistent(serialized, _size);
rocksutils::uint64ToPersistent(serialized, _nrUsed);
rocksutils::uint64ToPersistent(serialized, _nrCuckood);
rocksutils::uint64ToPersistent(serialized, _nrTotal);
rocksutils::uint64ToPersistent(serialized, _niceSize);
rocksutils::uint64ToPersistent(serialized, _logSize);
// Add the data blob
// Size is as follows: nrOfBuckets * kSlotsPerBucket * SlotSize
TRI_ASSERT((_size * kSlotSize * kSlotsPerBucket) <= _slotAllocSize);
for (uint64_t i = 0; i < (_size * kSlotSize * kSlotsPerBucket); i += kSlotSize) {
rocksutils::uint16ToPersistent(serialized,
*(reinterpret_cast<uint16_t*>(_base + i)));
}
TRI_ASSERT((_size * kCounterSize * kSlotsPerBucket) <= _counterAllocSize);
for (uint64_t i = 0; i < (_size * kCounterSize * kSlotsPerBucket); i += kCounterSize) {
rocksutils::uint32ToPersistent(serialized,
*(reinterpret_cast<uint32_t*>(_counters + i)));
}
bool havePendingUpdates = !_insertBuffers.empty() || !_removalBuffers.empty() ||
!_truncateBuffer.empty();
_needToPersist.store(havePendingUpdates, std::memory_order_release);
}
_appliedSeq.store(appliedSeq, std::memory_order_release);
}
/// @brief only call directly during startup/recovery; otherwise buffer
void clear() {
WRITE_LOCKER(locker, _lock);
// Reset Stats
_nrTotal = 0;
_nrCuckood = 0;
_nrUsed = 0;
// Reset filter content
// Now initialize all slots in all buckets with zero data:
for (uint32_t b = 0; b < _size; ++b) {
for (size_t i = 0; i < kSlotsPerBucket; ++i) {
Slot f = findSlot(b, i);
f.injectCounter(findCounter(b, i));
f.reset();
}
}
_needToPersist.store(true, std::memory_order_release);
}
Result bufferTruncate(rocksdb::SequenceNumber seq) {
Result res = basics::catchVoidToResult([&]() -> void {
WRITE_LOCKER(locker, _lock);
_truncateBuffer.emplace(seq);
_needToPersist.store(true, std::memory_order_release);
});
return res;
}
double computeEstimate() {
READ_LOCKER(locker, _lock);
if (0 == _nrTotal) {
TRI_ASSERT(0 == _nrUsed);
// If we do not have any documents, we have a rather constant estimate.
return 1.0;
}
TRI_ASSERT(_nrUsed <= _nrTotal);
if (_nrUsed > _nrTotal) {
_nrTotal = _nrUsed; // should never happen, but will keep estimates valid
// for production where the above assert is disabled
}
return (static_cast<double>(_nrUsed) / static_cast<double>(_nrTotal));
}
bool lookup(Key const& k) const {
// look up a key, return either false if no pair with key k is
// found or true.
uint64_t hash1 = _hasherKey(k);
uint64_t pos1 = hashToPos(hash1);
uint16_t fingerprint = keyToFingerprint(k);
// We compute the second hash already here to allow the result to
// survive a mispredicted branch in the first loop. Is this sensible?
uint64_t hash2 = _hasherPosFingerprint(pos1, fingerprint);
uint64_t pos2 = hashToPos(hash2);
bool found = false;
{
READ_LOCKER(guard, _lock);
findSlotNoCuckoo(pos1, pos2, fingerprint, found);
}
return found;
}
/// @brief only call directly during startup/recovery; otherwise buffer
bool insert(Key const& k) {
// insert the key k
//
// The inserted key will have its fingerprint input entered in the table. If
// there is a collision and a fingerprint needs to be cuckooed, a certain
// number of attempts will be made. After that, a given fingerprint may
// simply be expunged. If something is expunged, the function will return
// false, otherwise true.
uint64_t hash1 = _hasherKey(k);
uint64_t pos1 = hashToPos(hash1);
uint16_t fingerprint = keyToFingerprint(k);
// We compute the second hash already here to let it survive a
// mispredicted
// branch in the first loop:
uint64_t hash2 = _hasherPosFingerprint(pos1, fingerprint);
uint64_t pos2 = hashToPos(hash2);
{
WRITE_LOCKER(guard, _lock);
Slot slot = findSlotCuckoo(pos1, pos2, fingerprint);
if (slot.isEmpty()) {
// Free slot insert ourself.
slot.init(fingerprint);
++_nrUsed;
TRI_ASSERT(_nrUsed > 0);
} else {
TRI_ASSERT(slot.isEqual(fingerprint));
slot.increase();
}
++_nrTotal;
_needToPersist.store(true, std::memory_order_release);
}
return true;
}
/// @brief only call directly during startup/recovery; otherwise buffer
bool remove(Key const& k) {
// remove one element with key k, if one is in the table. Return true if
// a key was removed and false otherwise.
// look up a key, return either false if no pair with key k is
// found or true.
uint64_t hash1 = _hasherKey(k);
uint64_t pos1 = hashToPos(hash1);
uint16_t fingerprint = keyToFingerprint(k);
// We compute the second hash already here to allow the result to
// survive a mispredicted branch in the first loop. Is this sensible?
uint64_t hash2 = _hasherPosFingerprint(pos1, fingerprint);
uint64_t pos2 = hashToPos(hash2);
bool found = false;
{
WRITE_LOCKER(guard, _lock);
Slot slot = findSlotNoCuckoo(pos1, pos2, fingerprint, found);
if (found) {
// only decrease the total if we actually found it
--_nrTotal;
if (!slot.decrease()) {
// Removed last element. Have to remove
slot.reset();
--_nrUsed;
}
_needToPersist.store(true, std::memory_order_release);
return true;
}
// If we get here we assume that the element was once inserted, but
// removed by cuckoo
// Reduce nrCuckood;
if (_nrCuckood > 0) {
// not included in _nrTotal, just decrease here
--_nrCuckood;
}
_needToPersist.store(true, std::memory_order_release);
}
return false;
}
uint64_t capacity() const { return _size * kSlotsPerBucket; }
// not thread safe. called only during tests
uint64_t nrTotal() const { return _nrTotal; }
// not thread safe. called only during tests
uint64_t nrUsed() const { return _nrUsed; }
// not thread safe. called only during tests
uint64_t nrCuckood() const { return _nrCuckood; }
bool needToPersist() const {
return _needToPersist.load(std::memory_order_acquire);
}
/**
* @brief Buffer updates to this estimator to be applied when appropriate
*
* Buffers updates associated with a given commit seq/tick. Will hold updates
* until all previous blockers have been removed to ensure a consistent state
* for sync/recovery and avoid any missed updates.
*
* @param seq The seq/tick post-commit, prior to call
* @param inserts Vector of hashes to insert
* @param removals Vector of hashes to remove
* @return May return error if any functions throw (e.g. alloc)
*/
Result bufferUpdates(rocksdb::SequenceNumber seq, std::vector<Key>&& inserts,
std::vector<Key>&& removals) {
TRI_ASSERT(!inserts.empty() || !removals.empty());
Result res = basics::catchVoidToResult([&]() -> void {
WRITE_LOCKER(locker, _lock);
if (!inserts.empty()) {
_insertBuffers.emplace(seq, std::move(inserts));
}
if (!removals.empty()) {
_removalBuffers.emplace(seq, std::move(removals));
}
_needToPersist.store(true, std::memory_order_release);
});
return res;
}
/**
* @brief Fetches the most recently set "committed" seq/tick
*
* This value is set during serialization.
*
* @return The latest seq/tick through which the estimate is valid
*/
rocksdb::SequenceNumber appliedSeq() const {
return _appliedSeq.load(std::memory_order_acquire);
}
/// @brief set the most recently set "committed" seq/tick
/// only set when recalculating the index estimate
void setAppliedSeq(rocksdb::SequenceNumber seq) {
_appliedSeq.store(seq, std::memory_order_release);
_needToPersist.store(true, std::memory_order_release);
}
private: // methods
/// @brief call with output from committableSeq(current), and before serialize
rocksdb::SequenceNumber applyUpdates(rocksdb::SequenceNumber commitSeq) {
rocksdb::SequenceNumber appliedSeq = 0;
Result res = basics::catchVoidToResult([&]() -> void {
std::vector<Key> inserts;
std::vector<Key> removals;
// truncate will increase this sequence
rocksdb::SequenceNumber ignoreSeq = 0;
while (true) {
bool foundTruncate = false;
// find out if we have buffers to apply
{
WRITE_LOCKER(locker, _lock);
{
// check for a truncate marker
auto it = _truncateBuffer.begin(); // sorted ASC
while (it != _truncateBuffer.end() && *it <= commitSeq) {
ignoreSeq = *it;
TRI_ASSERT(ignoreSeq != 0);
foundTruncate = true;
appliedSeq = std::max(appliedSeq, ignoreSeq);
it = _truncateBuffer.erase(it);
}
}
TRI_ASSERT(ignoreSeq <= commitSeq);
// check for inserts
auto it = _insertBuffers.begin(); // sorted ASC
while (it != _insertBuffers.end() && it->first <= commitSeq) {
if (it->first <= ignoreSeq) {
TRI_ASSERT(it->first <= appliedSeq);
it = _insertBuffers.erase(it);
continue;
}
inserts = std::move(it->second);
TRI_ASSERT(!inserts.empty());
appliedSeq = std::max(appliedSeq, it->first);
_insertBuffers.erase(it);
break;
}
// check for removals
it = _removalBuffers.begin(); // sorted ASC
while (it != _removalBuffers.end() && it->first <= commitSeq) {
if (it->first <= ignoreSeq) {
TRI_ASSERT(it->first <= appliedSeq);
it = _removalBuffers.erase(it);
continue;
}
removals = std::move(it->second);
TRI_ASSERT(!removals.empty());
appliedSeq = std::max(appliedSeq, it->first);
_removalBuffers.erase(it);
break;
}
}
if (foundTruncate) {
clear(); // clear estimates
}
// no inserts or removals left to apply, drop out of loop
if (inserts.empty() && removals.empty()) {
break;
}
// apply inserts
for (auto const& key : inserts) {
insert(key);
}
inserts.clear();
// apply removals
for (auto const& key : removals) {
remove(key);
}
removals.clear();
} // </while(true)>
});
return appliedSeq;
}
uint64_t memoryUsage() const {
return sizeof(RocksDBCuckooIndexEstimator) + _slotAllocSize + _counterAllocSize;
}
Slot findSlotNoCuckoo(uint64_t pos1, uint64_t pos2, uint16_t fp, bool& found) const {
found = false;
Slot s = findSlotNoCuckoo(pos1, fp, found);
if (found) {
return s;
}
// Not found by first hash. use second hash.
return findSlotNoCuckoo(pos2, fp, found);
}
// Find a slot for this fingerprint
// This function guarantees the following:
// If this fingerprint is stored already, the slot will be
// pointing to this fingerprint.
// If this fingerprint is NOT storead already the returned slot
// will be empty and can be filled with the fingerprint
// In order to create an empty slot this function tries to
// cuckoo neighboring elements, if that does not succeed
// it deletes a random element occupying a position.
Slot findSlotCuckoo(uint64_t pos1, uint64_t pos2, uint16_t fp) {
Slot firstEmpty(nullptr);
bool foundEmpty = false;
for (uint64_t i = 0; i < kSlotsPerBucket; ++i) {
Slot slot = findSlot(pos1, i);
if (slot.isEqual(fp)) {
// Found we are done, short-circuit.
slot.injectCounter(findCounter(pos1, i));
return slot;
}
if (!foundEmpty && slot.isEmpty()) {
slot.injectCounter(findCounter(pos1, i));
foundEmpty = true;
firstEmpty = slot;
}
}
for (uint64_t i = 0; i < kSlotsPerBucket; ++i) {
Slot slot = findSlot(pos2, i);
if (slot.isEqual(fp)) {
// Found we are done, short-circuit.
slot.injectCounter(findCounter(pos2, i));
return slot;
}
if (!foundEmpty && slot.isEmpty()) {
slot.injectCounter(findCounter(pos2, i));
foundEmpty = true;
firstEmpty = slot;
}
}
// Value not yet inserted.
if (foundEmpty) {
// But we found an empty slot
return firstEmpty;
}
// We also did not find an empty slot, now the cuckoo goes...
// We initially write a 0 in here, because the caller will
// Increase the counter by one
uint32_t counter = 0;
uint8_t r = pseudoRandomChoice();
if ((r & 1) != 0) {
std::swap(pos1, pos2);
}
// Now expunge a random element from any of these slots:
// and place our own into it.
// We have to keep the reference to the cuckood slot here.
r = pseudoRandomChoice();
uint64_t i = r & (kSlotsPerBucket - 1);
firstEmpty = findSlot(pos1, i);
firstEmpty.injectCounter(findCounter(pos1, i));
firstEmpty.swap(fp, counter);
uint64_t hash2 = _hasherPosFingerprint(pos1, fp);
pos2 = hashToPos(hash2);
// Now let the cuckoo fly and find a place for the poor one we just took
// out.
for (uint64_t i = 0; i < kSlotsPerBucket; ++i) {
Slot slot = findSlot(pos2, i);
if (slot.isEmpty()) {
slot.injectCounter(findCounter(pos2, i));
// Yeah we found an empty place already
*slot.fingerprint() = fp;
*slot.counter() = counter;
++_nrUsed;
return firstEmpty;
}
}
// Bad luck, let us try to move to a different slot.
for (unsigned attempt = 1; attempt < kMaxRounds; attempt++) {
std::swap(pos1, pos2);
// Now expunge a random element from any of these slots:
r = pseudoRandomChoice();
uint64_t i = r & (kSlotsPerBucket - 1);
// We expunge the element at position pos1 and slot i:
Slot slot = findSlot(pos1, i);
if (slot == firstEmpty) {
// We have to keep this one in place.
// Take a different one
i = (i + 1) % kSlotsPerBucket;
slot = findSlot(pos1, i);
}
slot.injectCounter(findCounter(pos1, i));
slot.swap(fp, counter);
hash2 = _hasherPosFingerprint(pos1, fp);
pos2 = hashToPos(hash2);
for (uint64_t i = 0; i < kSlotsPerBucket; ++i) {
Slot slot = findSlot(pos2, i);
if (slot.isEmpty()) {
slot.injectCounter(findCounter(pos2, i));
// Finally an empty place
*slot.fingerprint() = fp;
*slot.counter() = counter;
++_nrUsed;
return firstEmpty;
}
}
}
// If we get here we had to remove one of the elements.
// Let's increas the cuckoo counter
_nrCuckood++;
// and let's decrease the total so we don't have to recalculate later
_nrTotal = (_nrTotal >= counter) ? (_nrTotal - counter) : 0;
return firstEmpty;
}
// Do not use the output if found == false
Slot findSlotNoCuckoo(uint64_t pos, uint16_t fp, bool& found) const {
found = false;
for (uint64_t i = 0; i < kSlotsPerBucket; ++i) {
Slot slot = findSlot(pos, i);
if (fp == *slot.fingerprint()) {
slot.injectCounter(findCounter(pos, i));
found = true;
return slot;
}
}
return Slot{nullptr};
}
Slot findSlot(uint64_t pos, uint64_t slot) const {
TRI_ASSERT(kSlotSize * (pos * kSlotsPerBucket + slot) <= _slotAllocSize);
char* address = _base + kSlotSize * (pos * kSlotsPerBucket + slot);
auto ret = reinterpret_cast<uint16_t*>(address);
return Slot(ret);
}
uint32_t* findCounter(uint64_t pos, uint64_t slot) const {
TRI_ASSERT(kCounterSize * (pos * kSlotsPerBucket + slot) <= _counterAllocSize);
char* address = _counters + kCounterSize * (pos * kSlotsPerBucket + slot);
return reinterpret_cast<uint32_t*>(address);
}
uint64_t hashToPos(uint64_t hash) const {
uint64_t relevantBits = (hash >> _sizeShift) & _sizeMask;
return ((relevantBits < _size) ? relevantBits : (relevantBits - _size));
}
uint16_t keyToFingerprint(Key const& k) const {
uint64_t hash = _fingerprint(k);
uint16_t fingerprint =
(uint16_t)((hash ^ (hash >> 16) ^ (hash >> 32) ^ (hash >> 48)) & 0xFFFF);
return (fingerprint ? fingerprint : 1);
}
uint64_t _hasherPosFingerprint(uint64_t pos, uint16_t fingerprint) const {
return ((pos << _sizeShift) ^ _hasherShort(fingerprint));
}
uint8_t pseudoRandomChoice() {
_randState = _randState * 997 + 17; // ignore overflows
return static_cast<uint8_t>((_randState >> 37) & 0xff);
}
void deserializeUncompressed(arangodb::velocypack::StringRef const& serialized) {
// Assert that we have at least the member variables
TRI_ASSERT(serialized.size() >=
(sizeof(_appliedSeq)) +
(sizeof(SerializeFormat) + sizeof(uint64_t) + sizeof(_size) +
sizeof(_nrUsed) + sizeof(_nrCuckood) + sizeof(_nrTotal) +
sizeof(_niceSize) + sizeof(_logSize)));
char const* current = serialized.data();
_appliedSeq = rocksutils::uint64FromPersistent(current);
current += sizeof(_appliedSeq);
TRI_ASSERT(*current == SerializeFormat::NOCOMPRESSION);
current++; // Skip format char
uint64_t length = rocksutils::uint64FromPersistent(current);
current += sizeof(uint64_t);
// Validate that the serialized format is exactly as long as
// we expect it to be
TRI_ASSERT(serialized.size() == length + sizeof(_appliedSeq));
_size = rocksutils::uint64FromPersistent(current);
current += sizeof(uint64_t);
if (_size <= 256) {
THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_INTERNAL,
"unable to unserialize index estimates");
}
_nrUsed = rocksutils::uint64FromPersistent(current);
current += sizeof(uint64_t);
_nrCuckood = rocksutils::uint64FromPersistent(current);
current += sizeof(uint64_t);
_nrTotal = rocksutils::uint64FromPersistent(current);
current += sizeof(uint64_t);
_niceSize = rocksutils::uint64FromPersistent(current);
current += sizeof(uint64_t);
_logSize = rocksutils::uint64FromPersistent(current);
current += sizeof(uint64_t);
deriveSizesAndAlloc();
// Validate that we have enough data in the serialized format.
TRI_ASSERT(serialized.size() ==
(sizeof(_appliedSeq) + sizeof(SerializeFormat) +
sizeof(uint64_t) + sizeof(_size) + sizeof(_nrUsed) +
sizeof(_nrCuckood) + sizeof(_nrTotal) + sizeof(_niceSize) +
sizeof(_logSize) + (_size * kSlotSize * kSlotsPerBucket)) +
(_size * kCounterSize * kSlotsPerBucket));
// Insert the raw data
// Size is as follows: nrOfBuckets * kSlotsPerBucket * SlotSize
TRI_ASSERT((_size * kSlotSize * kSlotsPerBucket) <= _slotAllocSize);
for (uint64_t i = 0; i < (_size * kSlotSize * kSlotsPerBucket); i += kSlotSize) {
*(reinterpret_cast<uint16_t*>(_base + i)) =
rocksutils::uint16FromPersistent(current);
current += kSlotSize;
}
TRI_ASSERT((_size * kCounterSize * kSlotsPerBucket) <= _counterAllocSize);
for (uint64_t i = 0; i < (_size * kCounterSize * kSlotsPerBucket); i += kCounterSize) {
*(reinterpret_cast<uint32_t*>(_counters + i)) =
rocksutils::uint32FromPersistent(current);
current += kCounterSize;
}
}
void initializeDefault() {
_niceSize = 256;
_logSize = 8;
while (_niceSize < _size) {
_niceSize <<= 1;
_logSize += 1;
}
deriveSizesAndAlloc();
// Now initialize all slots in all buckets with zero data:
for (uint32_t b = 0; b < _size; ++b) {
for (size_t i = 0; i < kSlotsPerBucket; ++i) {
Slot f = findSlot(b, i);
f.injectCounter(findCounter(b, i));
f.reset();
}
}
}
void deriveSizesAndAlloc() {
_sizeMask = _niceSize - 1;
_sizeShift = static_cast<uint32_t>((64 - _logSize) / 2);
// give 64 bytes padding to enable 64-byte alignment
_slotAllocSize = _size * kSlotSize * kSlotsPerBucket + 64;
_slotBase = new char[_slotAllocSize];
_base = reinterpret_cast<char*>(
(reinterpret_cast<uintptr_t>(_slotBase) + 63) &
~((uintptr_t)0x3fu)); // to actually implement the 64-byte alignment,
// shift base pointer within allocated space to
// 64-byte boundary
// give 64 bytes padding to enable 64-byte alignment
_counterAllocSize = _size * kCounterSize * kSlotsPerBucket + 64;
_counterBase = new char[_counterAllocSize];
_counters = reinterpret_cast<char*>(
(reinterpret_cast<uintptr_t>(_counterBase) + 63) &
~((uintptr_t)0x3fu)); // to actually implement the 64-byte alignment,
// shift base pointer within allocated space to
// 64-byte boundary
}
private: // member variables
uint64_t _randState; // pseudo random state for expunging
uint64_t _logSize; // logarithm (base 2) of number of buckets
uint64_t _size; // actual number of buckets
uint64_t _niceSize; // smallest power of 2 at least number of buckets, ==
// 2^_logSize
uint64_t _sizeMask; // used to mask out some bits from the hash
uint32_t _sizeShift; // used to shift the bits down to get a position
uint64_t _slotAllocSize; // number of allocated bytes for the slots,
// == _size * kSlotsPerBucket * kSlotSize + 64
uint64_t _counterAllocSize; // number of allocated bytes ofr the counters,
// == _size * kSlotsPerBucket * kCounterSize + 64
char* _base; // pointer to allocated space, 64-byte aligned
char* _slotBase; // base of original allocation
char* _counters; // pointer to allocated space, 64-byte aligned
char* _counterBase; // base of original counter allocation
uint64_t _nrUsed; // number of pairs stored in the table
uint64_t _nrCuckood; // number of elements that have been removed by cuckoo
uint64_t _nrTotal; // number of elements included in total (not cuckood)
std::atomic<rocksdb::SequenceNumber> _appliedSeq;
std::atomic<bool> _needToPersist;
std::multimap<rocksdb::SequenceNumber, std::vector<Key>> _insertBuffers;
std::multimap<rocksdb::SequenceNumber, std::vector<Key>> _removalBuffers;
std::set<rocksdb::SequenceNumber> _truncateBuffer;
HashKey _hasherKey; // Instance to compute the first hash function
Fingerprint _fingerprint; // Instance to compute a fingerprint of a key
HashShort _hasherShort; // Instance to compute the second hash function
arangodb::basics::ReadWriteLock mutable _lock;
}; // namespace arangodb
} // namespace arangodb
#endif
Reference in New Issue View Git Blame Copy Permalink