From 32ce8b311269df205e192c2b61dc47faaa2c5971 Mon Sep 17 00:00:00 2001 From: KunoiSayami Date: Sun, 20 Feb 2022 15:41:14 +0800 Subject: feat(memtable): Add kernel function in cuda code Signed-off-by: KunoiSayami --- CMakeLists.txt | 8 ++- db/memtable.cu | 203 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++ db/memtable.cuh | 90 +++++++++++++++++++++++++ db/skiplist.cuh | 2 +- util/arena.cu | 4 +- util/arena.cuh | 4 +- util/coding.cu | 157 +++++++++++++++++++++++++++++++++++++++++++ util/coding.cuh | 123 ++++++++++++++++++++++++++++++++++ 8 files changed, 583 insertions(+), 8 deletions(-) create mode 100644 db/memtable.cu create mode 100644 db/memtable.cuh create mode 100644 util/coding.cu create mode 100644 util/coding.cuh diff --git a/CMakeLists.txt b/CMakeLists.txt index b6d0d01..aa46c4b 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -144,7 +144,8 @@ target_sources(leveldb "db/log_reader.h" "db/log_writer.cc" "db/log_writer.h" - "db/memtable.cc" + "db/memtable.cu" + "db/memtable.cuh" "db/memtable.h" "db/repair.cc" "db/skiplist.h" @@ -182,8 +183,9 @@ target_sources(leveldb "util/arena.cuh" "util/bloom.cc" "util/cache.cc" - "util/coding.cc" - "util/coding.h" + "util/coding.cu" + "util/coding.cuh" + #"util/coding.h" "util/comparator.cc" "util/crc32c.cc" "util/crc32c.h" diff --git a/db/memtable.cu b/db/memtable.cu new file mode 100644 index 0000000..c4c3a75 --- /dev/null +++ b/db/memtable.cu @@ -0,0 +1,203 @@ +// Copyright (c) 2011 The LevelDB Authors. All rights reserved. +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. See the AUTHORS file for names of contributors. + +#include "db/memtable.cuh" +#include "db/dbformat.h" +#include "leveldb/comparator.h" +#include "leveldb/env.h" +#include "leveldb/iterator.h" +#include "util/coding.cuh" + +namespace leveldb { + +static Slice GetLengthPrefixedSlice(const char* data) { + uint32_t len; + const char* p = data; + p = GetVarint32Ptr(p, p + 5, &len); // +5: we assume "p" is not corrupted + return Slice(p, len); +} + +__device__ MemTable::MemTable(const InternalKeyComparator& comparator) + : comparator_(comparator), refs_(0), table_(comparator_, &arena_) {} + +MemTable::~MemTable() { assert(refs_ == 0); } + +size_t MemTable::ApproximateMemoryUsage() { return arena_.MemoryUsage(); } + +int MemTable::KeyComparator::operator()(const char* aptr, + const char* bptr) const { + // Internal keys are encoded as length-prefixed strings. + Slice a = GetLengthPrefixedSlice(aptr); + Slice b = GetLengthPrefixedSlice(bptr); + return comparator.Compare(a, b); +} + +// Encode a suitable internal key target for "target" and return it. +// Uses *scratch as scratch space, and the returned pointer will point +// into this scratch space. +static const char* EncodeKey(std::string* scratch, const Slice& target) { + scratch->clear(); + PutVarint32(scratch, target.size()); + scratch->append(target.data(), target.size()); + return scratch->data(); +} + +class MemTableIterator : public Iterator { + public: + explicit MemTableIterator(MemTable::Table* table) : iter_(table) {} + + MemTableIterator(const MemTableIterator&) = delete; + MemTableIterator& operator=(const MemTableIterator&) = delete; + + ~MemTableIterator() override = default; + + bool Valid() const override { return iter_.Valid(); } + void Seek(const Slice& k) override { iter_.Seek(EncodeKey(&tmp_, k)); } + void SeekToFirst() override { iter_.SeekToFirst(); } + void SeekToLast() override { iter_.SeekToLast(); } + void Next() override { iter_.Next(); } + void Prev() override { iter_.Prev(); } + Slice key() const override { return GetLengthPrefixedSlice(iter_.key()); } + Slice value() const override { + Slice key_slice = GetLengthPrefixedSlice(iter_.key()); + return GetLengthPrefixedSlice(key_slice.data() + key_slice.size()); + } + + Status status() const override { return Status::OK(); } + + private: + MemTable::Table::Iterator iter_; + std::string tmp_; // For passing to EncodeKey +}; + +Iterator* MemTable::NewIterator() { return new MemTableIterator(&table_); } + +__device__ char* EncodeVarint32Device(char* dst, uint32_t v) { + // Operate on characters as unsigneds + uint8_t* ptr = reinterpret_cast(dst); + static const int B = 128; + if (v < (1 << 7)) { + *(ptr++) = v; + } else if (v < (1 << 14)) { + *(ptr++) = v | B; + *(ptr++) = v >> 7; + } else if (v < (1 << 21)) { + *(ptr++) = v | B; + *(ptr++) = (v >> 7) | B; + *(ptr++) = v >> 14; + } else if (v < (1 << 28)) { + *(ptr++) = v | B; + *(ptr++) = (v >> 7) | B; + *(ptr++) = (v >> 14) | B; + *(ptr++) = v >> 21; + } else { + *(ptr++) = v | B; + *(ptr++) = (v >> 7) | B; + *(ptr++) = (v >> 14) | B; + *(ptr++) = (v >> 21) | B; + *(ptr++) = v >> 28; + } + return reinterpret_cast(ptr); +} + +__device__ inline void EncodeFixed64Device(char* dst, uint64_t value) { + uint8_t* const buffer = reinterpret_cast(dst); + + // Recent clang and gcc optimize this to a single mov / str instruction. + buffer[0] = static_cast(value); + buffer[1] = static_cast(value >> 8); + buffer[2] = static_cast(value >> 16); + buffer[3] = static_cast(value >> 24); + buffer[4] = static_cast(value >> 32); + buffer[5] = static_cast(value >> 40); + buffer[6] = static_cast(value >> 48); + buffer[7] = static_cast(value >> 56); +} + +__global__ void Add_(MemTable * mtb, SequenceNumber s, ValueType type, + size_t encoded_len, size_t internal_key_size, + char * key_data, size_t key_size, char * value_data, size_t val_size) { + char* buf = mtb->arena_.Allocate(encoded_len); + char* p = EncodeVarint32Device(buf, internal_key_size); + memcpy(p, key_data, key_size); + //std::memcpy(p, key_data, key_size); + p += key_size; + EncodeFixed64Device(p, (s << 8) | type); + p += 8; + p = EncodeVarint32Device(p, val_size); + //std::memcpy(p, value.data(), val_size); + memcpy(p, value_data, val_size); + assert(p + val_size == buf + encoded_len); + mtb->table_.Insert(buf); +} + +void MemTable::Add(SequenceNumber s, ValueType type, const Slice& key, + const Slice& value) { + // Format of an entry is concatenation of: + // key_size : varint32 of internal_key.size() + // key bytes : char[internal_key.size()] + // tag : uint64((sequence << 8) | type) + // value_size : varint32 of value.size() + // value bytes : char[value.size()] + size_t key_size = key.size(); + size_t val_size = value.size(); + size_t internal_key_size = key_size + 8; + const size_t encoded_len = VarintLength(internal_key_size) + + internal_key_size + VarintLength(val_size) + + val_size; + char * key_mem = nullptr; + cudaMalloc((void**)&key_mem, key_size); + cudaMemcpy(key_mem, key.data(), key_size, cudaMemcpyHostToDevice); + char * value_mem = nullptr; + cudaMalloc((void**)&value_mem, val_size); + cudaMemcpy(value_mem, value.data(), val_size, cudaMemcpyHostToDevice); + + Add_<<<1, 1>>>(this, s, type, encoded_len, internal_key_size, key_mem, key_size, value_mem, val_size); + cudaDeviceSynchronize(); + + cudaFree(key_mem); + cudaFree(value_mem); +} + +__global__ void Get_(MemTable * met, bool * ret) { + + Slice memkey = key.memtable_key(); + met->Table::Iterator iter(&met->table_); + iter.Seek(memkey.data()); + if (iter.Valid()) { + // entry format is: + // klength varint32 + // userkey char[klength] + // tag uint64 + // vlength varint32 + // value char[vlength] + // Check that it belongs to same user key. We do not check the + // sequence number since the Seek() call above should have skipped + // all entries with overly large sequence numbers. + const char* entry = iter.key(); + uint32_t key_length; + const char* key_ptr = GetVarint32Ptr(entry, entry + 5, &key_length); + if (met->comparator_.comparator.user_comparator()->Compare( + Slice(key_ptr, key_length - 8), key.user_key()) == 0) { + // Correct user key + const uint64_t tag = DecodeFixed64(key_ptr + key_length - 8); + switch (static_cast(tag & 0xff)) { + case kTypeValue: { + Slice v = GetLengthPrefixedSlice(key_ptr + key_length); + value->assign(v.data(), v.size()); + return true; + } + case kTypeDeletion: + *s = Status::NotFound(Slice()); + return true; + } + } + } +} + +bool MemTable::Get(const LookupKey& key, std::string* value, Status* s) { + return false; +} + +} // namespace leveldb diff --git a/db/memtable.cuh b/db/memtable.cuh new file mode 100644 index 0000000..ac3c2b6 --- /dev/null +++ b/db/memtable.cuh @@ -0,0 +1,90 @@ +// Copyright (c) 2011 The LevelDB Authors. All rights reserved. +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. See the AUTHORS file for names of contributors. + +#ifndef STORAGE_LEVELDB_DB_MEMTABLE_H_ +#define STORAGE_LEVELDB_DB_MEMTABLE_H_ + +#include + +#include "db/dbformat.h" +#include "db/skiplist.cuh" +#include "leveldb/db.h" +#include "util/arena.cuh" + +namespace leveldb { + +class InternalKeyComparator; +class MemTableIterator; + +class MemTable { + public: + // MemTables are reference counted. The initial reference count + // is zero and the caller must call Ref() at least once. + __device__ explicit MemTable(const InternalKeyComparator& comparator); + + MemTable(const MemTable&) = delete; + MemTable& operator=(const MemTable&) = delete; + + // Increase reference count. + void Ref() { ++refs_; } + + // Drop reference count. Delete if no more references exist. + void Unref() { + --refs_; + assert(refs_ >= 0); + if (refs_ <= 0) { + delete this; + } + } + + // Returns an estimate of the number of bytes of data in use by this + // data structure. It is safe to call when MemTable is being modified. + size_t ApproximateMemoryUsage(); + + // Return an iterator that yields the contents of the memtable. + // + // The caller must ensure that the underlying MemTable remains live + // while the returned iterator is live. The keys returned by this + // iterator are internal keys encoded by AppendInternalKey in the + // db/format.{h,cc} module. + Iterator* NewIterator(); + + // Add an entry into memtable that maps key to value at the + // specified sequence number and with the specified type. + // Typically value will be empty if type==kTypeDeletion. + void Add(SequenceNumber seq, ValueType type, const Slice& key, + const Slice& value); + + // If memtable contains a value for key, store it in *value and return true. + // If memtable contains a deletion for key, store a NotFound() error + // in *status and return true. + // Else, return false. + bool Get(const LookupKey& key, std::string* value, Status* s); + + private: + friend class MemTableIterator; + friend class MemTableBackwardIterator; + friend __global__ void Add_(MemTable *, SequenceNumber, ValueType, size_t, size_t, char*, size_t, char *, size_t); + friend __global__ void Get_(MemTable *, bool * ret); + + struct KeyComparator { + const InternalKeyComparator comparator; + __device__ explicit KeyComparator(const InternalKeyComparator& c) : comparator(c) {} + __device__ int operator()(const char* a, const char* b) const; + __device__ ~KeyComparator() = default; + }; + + typedef SkipList Table; + + ~MemTable(); // Private since only Unref() should be used to delete it + + KeyComparator comparator_; + int refs_; + Arena arena_; + Table table_; +}; + +} // namespace leveldb + +#endif // STORAGE_LEVELDB_DB_MEMTABLE_H_ diff --git a/db/skiplist.cuh b/db/skiplist.cuh index 1fd7091..125a790 100644 --- a/db/skiplist.cuh +++ b/db/skiplist.cuh @@ -218,7 +218,7 @@ struct SkipList::Node { // version of the returned Node. return next_[n].load(cuda::memory_order_acquire); } - __device__ void SetNext(int n, Node* x) { + __device__ __host__ void SetNext(int n, Node* x) { assert(n >= 0); // Use a 'release store' so that anybody who reads through this // pointer observes a fully initialized version of the inserted node. diff --git a/util/arena.cu b/util/arena.cu index f954577..e6c8c04 100644 --- a/util/arena.cu +++ b/util/arena.cu @@ -8,11 +8,11 @@ namespace leveldb { static const int kBlockSize = 4096; -__device__ Arena::Arena() +__device__ __host__ Arena::Arena() : alloc_ptr_(nullptr), alloc_bytes_remaining_(0), memory_usage_(0), head_(nullptr), blocks_(nullptr) {} -__device__ Arena::~Arena() { +__host__ __device__ Arena::~Arena() { ArenaNode * current = this->head_; while (current != nullptr) { ArenaNode * next = current->next; diff --git a/util/arena.cuh b/util/arena.cuh index b70dcb9..e9abd31 100644 --- a/util/arena.cuh +++ b/util/arena.cuh @@ -17,12 +17,12 @@ namespace leveldb { class Arena { public: - explicit __device__ Arena(); + explicit __device__ __host__ Arena(); Arena(const Arena&) = delete; Arena& operator=(const Arena&) = delete; - __device__ ~Arena(); + __host__ __device__ ~Arena(); // Return a pointer to a newly allocated memory block of "bytes" bytes. __device__ char* Allocate(size_t bytes); diff --git a/util/coding.cu b/util/coding.cu new file mode 100644 index 0000000..63ddf28 --- /dev/null +++ b/util/coding.cu @@ -0,0 +1,157 @@ +// Copyright (c) 2011 The LevelDB Authors. All rights reserved. +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. See the AUTHORS file for names of contributors. + +#include "util/coding.cuh" + +namespace leveldb { + +void PutFixed32(std::string* dst, uint32_t value) { + char buf[sizeof(value)]; + EncodeFixed32(buf, value); + dst->append(buf, sizeof(buf)); +} + +void PutFixed64(std::string* dst, uint64_t value) { + char buf[sizeof(value)]; + EncodeFixed64(buf, value); + dst->append(buf, sizeof(buf)); +} + +char* EncodeVarint32(char* dst, uint32_t v) { + // Operate on characters as unsigneds + uint8_t* ptr = reinterpret_cast(dst); + static const int B = 128; + if (v < (1 << 7)) { + *(ptr++) = v; + } else if (v < (1 << 14)) { + *(ptr++) = v | B; + *(ptr++) = v >> 7; + } else if (v < (1 << 21)) { + *(ptr++) = v | B; + *(ptr++) = (v >> 7) | B; + *(ptr++) = v >> 14; + } else if (v < (1 << 28)) { + *(ptr++) = v | B; + *(ptr++) = (v >> 7) | B; + *(ptr++) = (v >> 14) | B; + *(ptr++) = v >> 21; + } else { + *(ptr++) = v | B; + *(ptr++) = (v >> 7) | B; + *(ptr++) = (v >> 14) | B; + *(ptr++) = (v >> 21) | B; + *(ptr++) = v >> 28; + } + return reinterpret_cast(ptr); +} + + +void PutVarint32(std::string* dst, uint32_t v) { + char buf[5]; + char* ptr = EncodeVarint32(buf, v); + dst->append(buf, ptr - buf); +} + +char* EncodeVarint64(char* dst, uint64_t v) { + static const int B = 128; + uint8_t* ptr = reinterpret_cast(dst); + while (v >= B) { + *(ptr++) = v | B; + v >>= 7; + } + *(ptr++) = static_cast(v); + return reinterpret_cast(ptr); +} + +void PutVarint64(std::string* dst, uint64_t v) { + char buf[10]; + char* ptr = EncodeVarint64(buf, v); + dst->append(buf, ptr - buf); +} + +void PutLengthPrefixedSlice(std::string* dst, const Slice& value) { + PutVarint32(dst, value.size()); + dst->append(value.data(), value.size()); +} + +int VarintLength(uint64_t v) { + int len = 1; + while (v >= 128) { + v >>= 7; + len++; + } + return len; +} + +const char* GetVarint32PtrFallback(const char* p, const char* limit, + uint32_t* value) { + uint32_t result = 0; + for (uint32_t shift = 0; shift <= 28 && p < limit; shift += 7) { + uint32_t byte = *(reinterpret_cast(p)); + p++; + if (byte & 128) { + // More bytes are present + result |= ((byte & 127) << shift); + } else { + result |= (byte << shift); + *value = result; + return reinterpret_cast(p); + } + } + return nullptr; +} + +bool GetVarint32(Slice* input, uint32_t* value) { + const char* p = input->data(); + const char* limit = p + input->size(); + const char* q = GetVarint32Ptr(p, limit, value); + if (q == nullptr) { + return false; + } else { + *input = Slice(q, limit - q); + return true; + } +} + +const char* GetVarint64Ptr(const char* p, const char* limit, uint64_t* value) { + uint64_t result = 0; + for (uint32_t shift = 0; shift <= 63 && p < limit; shift += 7) { + uint64_t byte = *(reinterpret_cast(p)); + p++; + if (byte & 128) { + // More bytes are present + result |= ((byte & 127) << shift); + } else { + result |= (byte << shift); + *value = result; + return reinterpret_cast(p); + } + } + return nullptr; +} + +bool GetVarint64(Slice* input, uint64_t* value) { + const char* p = input->data(); + const char* limit = p + input->size(); + const char* q = GetVarint64Ptr(p, limit, value); + if (q == nullptr) { + return false; + } else { + *input = Slice(q, limit - q); + return true; + } +} + +bool GetLengthPrefixedSlice(Slice* input, Slice* result) { + uint32_t len; + if (GetVarint32(input, &len) && input->size() >= len) { + *result = Slice(input->data(), len); + input->remove_prefix(len); + return true; + } else { + return false; + } +} + +} // namespace leveldb diff --git a/util/coding.cuh b/util/coding.cuh new file mode 100644 index 0000000..f89177b --- /dev/null +++ b/util/coding.cuh @@ -0,0 +1,123 @@ +// Copyright (c) 2011 The LevelDB Authors. All rights reserved. +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. See the AUTHORS file for names of contributors. +// +// Endian-neutral encoding: +// * Fixed-length numbers are encoded with least-significant byte first +// * In addition we support variable length "varint" encoding +// * Strings are encoded prefixed by their length in varint format + +#ifndef STORAGE_LEVELDB_UTIL_CODING_H_ +#define STORAGE_LEVELDB_UTIL_CODING_H_ + +#include +#include +#include + +#include "leveldb/slice.h" +#include "port/port.h" + +namespace leveldb { + +// Standard Put... routines append to a string +void PutFixed32(std::string* dst, uint32_t value); +void PutFixed64(std::string* dst, uint64_t value); +void PutVarint32(std::string* dst, uint32_t value); +void PutVarint64(std::string* dst, uint64_t value); +void PutLengthPrefixedSlice(std::string* dst, const Slice& value); + +// Standard Get... routines parse a value from the beginning of a Slice +// and advance the slice past the parsed value. +bool GetVarint32(Slice* input, uint32_t* value); +bool GetVarint64(Slice* input, uint64_t* value); +bool GetLengthPrefixedSlice(Slice* input, Slice* result); + +// Pointer-based variants of GetVarint... These either store a value +// in *v and return a pointer just past the parsed value, or return +// nullptr on error. These routines only look at bytes in the range +// [p..limit-1] +const char* GetVarint32Ptr(const char* p, const char* limit, uint32_t* v); +const char* GetVarint64Ptr(const char* p, const char* limit, uint64_t* v); + +// Returns the length of the varint32 or varint64 encoding of "v" +int VarintLength(uint64_t v); + +// Lower-level versions of Put... that write directly into a character buffer +// and return a pointer just past the last byte written. +// REQUIRES: dst has enough space for the value being written +char* EncodeVarint32(char* dst, uint32_t value); +__device__ char* EncodeVarint32Device(char* dst, uint32_t v); +char* EncodeVarint64(char* dst, uint64_t value); + +// Lower-level versions of Put... that write directly into a character buffer +// REQUIRES: dst has enough space for the value being written + +inline void EncodeFixed32(char* dst, uint32_t value) { + uint8_t* const buffer = reinterpret_cast(dst); + + // Recent clang and gcc optimize this to a single mov / str instruction. + buffer[0] = static_cast(value); + buffer[1] = static_cast(value >> 8); + buffer[2] = static_cast(value >> 16); + buffer[3] = static_cast(value >> 24); +} + +__host__ __device__ inline void EncodeFixed64(char* dst, uint64_t value) { + uint8_t* const buffer = reinterpret_cast(dst); + + // Recent clang and gcc optimize this to a single mov / str instruction. + buffer[0] = static_cast(value); + buffer[1] = static_cast(value >> 8); + buffer[2] = static_cast(value >> 16); + buffer[3] = static_cast(value >> 24); + buffer[4] = static_cast(value >> 32); + buffer[5] = static_cast(value >> 40); + buffer[6] = static_cast(value >> 48); + buffer[7] = static_cast(value >> 56); +} + +// Lower-level versions of Get... that read directly from a character buffer +// without any bounds checking. + +inline uint32_t DecodeFixed32(const char* ptr) { + const uint8_t* const buffer = reinterpret_cast(ptr); + + // Recent clang and gcc optimize this to a single mov / ldr instruction. + return (static_cast(buffer[0])) | + (static_cast(buffer[1]) << 8) | + (static_cast(buffer[2]) << 16) | + (static_cast(buffer[3]) << 24); +} + +inline uint64_t DecodeFixed64(const char* ptr) { + const uint8_t* const buffer = reinterpret_cast(ptr); + + // Recent clang and gcc optimize this to a single mov / ldr instruction. + return (static_cast(buffer[0])) | + (static_cast(buffer[1]) << 8) | + (static_cast(buffer[2]) << 16) | + (static_cast(buffer[3]) << 24) | + (static_cast(buffer[4]) << 32) | + (static_cast(buffer[5]) << 40) | + (static_cast(buffer[6]) << 48) | + (static_cast(buffer[7]) << 56); +} + +// Internal routine for use by fallback path of GetVarint32Ptr +const char* GetVarint32PtrFallback(const char* p, const char* limit, + uint32_t* value); +inline const char* GetVarint32Ptr(const char* p, const char* limit, + uint32_t* value) { + if (p < limit) { + uint32_t result = *(reinterpret_cast(p)); + if ((result & 128) == 0) { + *value = result; + return p + 1; + } + } + return GetVarint32PtrFallback(p, limit, value); +} + +} // namespace leveldb + +#endif // STORAGE_LEVELDB_UTIL_CODING_H_ -- cgit v1.3.1