首页 文章 精选 留言 我的

精选列表

搜索[源码学习]，共10000篇文章
优秀的个人博客,低调大师

LevelDB源码分析-Compact

Compaction compact由背景线程完成，代码中触发背景线程的函数为： void DBImpl::MaybeScheduleCompaction() { mutex_.AssertHeld(); if (background_compaction_scheduled_) { // Already scheduled } else if (shutting_down_.Acquire_Load()) { // DB is being deleted; no more background compactions } else if (!bg_error_.ok()) { // Already got an error; no more changes } else if (imm_ == nullptr && manual_compaction_ == nullptr && !versions_->NeedsCompaction()) { // No work to be done } else { background_compaction_scheduled_ = true; env_->Schedule(&DBImpl::BGWork, this); } } 背景线程调用的函数为： void DBImpl::BGWork(void *db) { reinterpret_cast<DBImpl *>(db)->BackgroundCall(); } 这个函数是调用BackgroundCall函数实现的： void DBImpl::BackgroundCall() { MutexLock l(&mutex_); assert(background_compaction_scheduled_); if (shutting_down_.Acquire_Load()) { // No more background work when shutting down. } else if (!bg_error_.ok()) { // No more background work after a background error. } else { BackgroundCompaction(); } background_compaction_scheduled_ = false; // Previous compaction may have produced too many files in a level, // so reschedule another compaction if needed. MaybeScheduleCompaction(); background_work_finished_signal_.SignalAll(); } 这个函数主要调用BackgroundCompaction实现相应逻辑： void DBImpl::BackgroundCompaction() 如果immutable table存在则先调用CompactMemTable函数： mutex_.AssertHeld(); if (imm_ != nullptr) { CompactMemTable(); return; } 然后调用PickCompaction函数，PickCompaction函数选取合适的level中需要compact的文件，然后封装一个Compact对象： else { c = versions_->PickCompaction(); } 通过IsTrivialMove判断选出的sstable是否只有一个且在较低的level，并且该sstable与下下一level不会有太多的overlap，如果是的话，简单的将这个sstable移动到下一level。 Status status; if (c == nullptr) { // Nothing to do } else if (!is_manual && c->IsTrivialMove()) { // Move file to next level assert(c->num_input_files(0) == 1); FileMetaData *f = c->input(0, 0); c->edit()->DeleteFile(c->level(), f->number); c->edit()->AddFile(c->level() + 1, f->number, f->file_size, f->smallest, f->largest); status = versions_->LogAndApply(c->edit(), &mutex_); if (!status.ok()) { RecordBackgroundError(status); } VersionSet::LevelSummaryStorage tmp; Log(options_.info_log, "Moved #%lld to level-%d %lld bytes %s: %s\n", static_cast<unsigned long long>(f->number), c->level() + 1, static_cast<unsigned long long>(f->file_size), status.ToString().c_str(), versions_->LevelSummary(&tmp)); } 如果当前level中需要compact的sstable和level+2中的overlap没有超过阈值时，调用DoCompactionWork进行compact，然后调用CleanupCompaction清除compactstate对象，调用DeleteObsoleteFiles删除旧文件，回收内存和磁盘空间： else { CompactionState *compact = new CompactionState(c); status = DoCompactionWork(compact); if (!status.ok()) { RecordBackgroundError(status); } CleanupCompaction(compact); c->ReleaseInputs(); DeleteObsoleteFiles(); } delete c; 其中调用的PickCompaction函数为： Compaction *VersionSet::PickCompaction() 如果有level达到compact的阈值，优先将这个level进行compact。将最大的key大于上一次compact的最大的key的sstable加入inputs_[0]的数组中作为compact的sstable，如果compact_pointer_[level]中没有记录，则从头开始取sstable，如果不存在最大的key大于上一次compact的最大的key的sstable，则将key最小的sstable作为需要compact的sstable： Compaction *c; int level; // We prefer compactions triggered by too much data in a level over // the compactions triggered by seeks. const bool size_compaction = (current_->compaction_score_ >= 1); const bool seek_compaction = (current_->file_to_compact_ != nullptr); if (size_compaction) { level = current_->compaction_level_; assert(level >= 0); assert(level + 1 < config::kNumLevels); c = new Compaction(options_, level); // Pick the first file that comes after compact_pointer_[level] for (size_t i = 0; i < current_->files_[level].size(); i++) { FileMetaData *f = current_->files_[level][i]; if (compact_pointer_[level].empty() || icmp_.Compare(f->largest.Encode(), compact_pointer_[level]) > 0) { c->inputs_[0].push_back(f); break; } } if (c->inputs_[0].empty()) { // Wrap-around to the beginning of the key space c->inputs_[0].push_back(current_->files_[level][0]); } } 如果没有level达到需要compact的阈值，但是有文件因为seek的命中率不够而达到了compact的阈值，则将这个文件compact： else if (seek_compaction) { level = current_->file_to_compact_level_; c = new Compaction(options_, level); c->inputs_[0].push_back(current_->file_to_compact_); } 如果compact的是level0，则还需要调用GetOverlappingInputs函数求出有overlap的文件，然后再调用SetupOtherInputs函数完成compact对象的构建： c->input_version_ = current_; c->input_version_->Ref(); // Files in level 0 may overlap each other, so pick up all overlapping ones if (level == 0) { InternalKey smallest, largest; GetRange(c->inputs_[0], &smallest, &largest); // Note that the next call will discard the file we placed in // c->inputs_[0] earlier and replace it with an overlapping set // which will include the picked file. current_->GetOverlappingInputs(0, &smallest, &largest, &c->inputs_[0]); assert(!c->inputs_[0].empty()); } SetupOtherInputs(c); return c; CompactMemTable函数 void DBImpl::CompactMemTable() 调用WriteLevel0Table将immutable mmtable写入sstable： mutex_.AssertHeld(); assert(imm_ != nullptr); // Save the contents of the memtable as a new Table VersionEdit edit; Version *base = versions_->current(); base->Ref(); Status s = WriteLevel0Table(imm_, &edit, base); base->Unref(); if (s.ok() && shutting_down_.Acquire_Load()) { s = Status::IOError("Deleting DB during memtable compaction"); } 更新当前lognumber，旧日志不再需要，生效新的version： // Replace immutable memtable with the generated Table if (s.ok()) { edit.SetPrevLogNumber(0); edit.SetLogNumber(logfile_number_); // Earlier logs no longer needed s = versions_->LogAndApply(&edit, &mutex_); } 删除旧文件： if (s.ok()) { // Commit to the new state imm_->Unref(); imm_ = nullptr; has_imm_.Release_Store(nullptr); DeleteObsoleteFiles(); } else { RecordBackgroundError(s); } WriteLevel0Table函数为： Status DBImpl::WriteLevel0Table(MemTable *mem, VersionEdit *edit, Version *base) 调用BuildTable接口将immutable memtable写入sstable： mutex_.AssertHeld(); const uint64_t start_micros = env_->NowMicros(); FileMetaData meta; meta.number = versions_->NewFileNumber(); pending_outputs_.insert(meta.number); Iterator *iter = mem->NewIterator(); Log(options_.info_log, "Level-0 table #%llu: started", (unsigned long long)meta.number); Status s; { mutex_.Unlock(); s = BuildTable(dbname_, env_, options_, table_cache_, iter, &meta); mutex_.Lock(); } Log(options_.info_log, "Level-0 table #%llu: %lld bytes %s", (unsigned long long)meta.number, (unsigned long long)meta.file_size, s.ToString().c_str()); delete iter; pending_outputs_.erase(meta.number); 调用PickLevelForMemTableOutput函数选择合适的level将新的sstable加入： // Note that if file_size is zero, the file has been deleted and // should not be added to the manifest. int level = 0; if (s.ok() && meta.file_size > 0) { const Slice min_user_key = meta.smallest.user_key(); const Slice max_user_key = meta.largest.user_key(); if (base != nullptr) { level = base->PickLevelForMemTableOutput(min_user_key, max_user_key); } edit->AddFile(level, meta.number, meta.file_size, meta.smallest, meta.largest); } CompactionStats stats; stats.micros = env_->NowMicros() - start_micros; stats.bytes_written = meta.file_size; stats_[level].Add(stats); return s; PickLevelForMemTableOutput函数： int Version::PickLevelForMemTableOutput( const Slice &smallest_user_key, const Slice &largest_user_key) { int level = 0; if (!OverlapInLevel(0, &smallest_user_key, &largest_user_key)) { // Push to next level if there is no overlap in next level, // and the #bytes overlapping in the level after that are limited. InternalKey start(smallest_user_key, kMaxSequenceNumber, kValueTypeForSeek); InternalKey limit(largest_user_key, 0, static_cast<ValueType>(0)); std::vector<FileMetaData *> overlaps; while (level < config::kMaxMemCompactLevel) { if (OverlapInLevel(level + 1, &smallest_user_key, &largest_user_key)) { break; } if (level + 2 < config::kNumLevels) { // Check that file does not overlap too many grandparent bytes. GetOverlappingInputs(level + 2, &start, &limit, &overlaps); const int64_t sum = TotalFileSize(overlaps); if (sum > MaxGrandParentOverlapBytes(vset_->options_)) { break; } } level++; } } return level; } 选择level的策略为：如果level0和sstable的key有重合，则插入level0，否则循环搜索之后的level，如果下一level和sstable的key有重合，则直接返回，如果没有重合，考虑下下个level和sstable的重合的key的量，如果超过了设定的阈值，则返回（为了保证之后选取该sstable与下一level进行compact的时候，涉及下一level的文件不会太多，减小开销）。 DoCompactionWork函数 Status DBImpl::DoCompactionWork(CompactionState *compact) 将smallest_snapshot赋值为最老的snapshots的sequencenumber，如果不存在snapshots则赋值为当前数据库的sequencenumber： const uint64_t start_micros = env_->NowMicros(); int64_t imm_micros = 0; // Micros spent doing imm_ compactions Log(options_.info_log, "Compacting %d@%d + %d@%d files", compact->compaction->num_input_files(0), compact->compaction->level(), compact->compaction->num_input_files(1), compact->compaction->level() + 1); assert(versions_->NumLevelFiles(compact->compaction->level()) > 0); assert(compact->builder == nullptr); assert(compact->outfile == nullptr); if (snapshots_.empty()) { compact->smallest_snapshot = versions_->LastSequence(); } else { compact->smallest_snapshot = snapshots_.oldest()->sequence_number(); } 调用MakeInputIterator获取需要compact的文件上的迭代器，可以通过这个迭代器直接遍历所有需要compact的文件。 // Release mutex while we're actually doing the compaction work mutex_.Unlock(); Iterator *input = versions_->MakeInputIterator(compact->compaction); input->SeekToFirst(); Status status; ParsedInternalKey ikey; std::string current_user_key; bool has_current_user_key = false; SequenceNumber last_sequence_for_key = kMaxSequenceNumber; 开始通过迭代器遍历需要compact的文件。首先，如果有immutable memtable存在，首先调用CompactMemTable函数进行compact。 for (; input->Valid() && !shutting_down_.Acquire_Load();) { // Prioritize immutable compaction work if (has_imm_.NoBarrier_Load() != nullptr) { const uint64_t imm_start = env_->NowMicros(); mutex_.Lock(); if (imm_ != nullptr) { CompactMemTable(); // Wake up MakeRoomForWrite() if necessary. background_work_finished_signal_.SignalAll(); } mutex_.Unlock(); imm_micros += (env_->NowMicros() - imm_start); } 获取迭代器当前指向的entry的key值，调用ShouldStopBefore函数判断是否需要停止当前sstable的构建，如果需要，则调用FinishCompactionOutputFile函数将包含当前sstable的文件输出。ShouldStopBefore函数主要是判断当前的sstable中的key是否与下下一level中的key的overlap过大，如果过大则停止当前sstable的构建： Slice key = input->key(); if (compact->compaction->ShouldStopBefore(key) && compact->builder != nullptr) { status = FinishCompactionOutputFile(compact, input); if (!status.ok()) { break; } } 将key解码，key和之前的key重复且之前的key的sequencenumber比smallest_snapshot要小则丢弃这个key，因为既然之前重复的key的sequencenumber都比smallest_snapshot要小，当前key也肯定小，key标记为删除且之后的level中不存在这个key（也就是说这个key的删除不会影响到之后）也丢弃这个key。 // Handle key/value, add to state, etc. bool drop = false; if (!ParseInternalKey(key, &ikey)) { // Do not hide error keys current_user_key.clear(); has_current_user_key = false; last_sequence_for_key = kMaxSequenceNumber; } else { if (!has_current_user_key || user_comparator()->Compare(ikey.user_key, Slice(current_user_key)) != 0) { // First occurrence of this user key current_user_key.assign(ikey.user_key.data(), ikey.user_key.size()); has_current_user_key = true; last_sequence_for_key = kMaxSequenceNumber; } if (last_sequence_for_key <= compact->smallest_snapshot) { // Hidden by an newer entry for same user key drop = true; // (A) } else if (ikey.type == kTypeDeletion && ikey.sequence <= compact->smallest_snapshot && compact->compaction->IsBaseLevelForKey(ikey.user_key)) { // For this user key: // (1) there is no data in higher levels // (2) data in lower levels will have larger sequence numbers // (3) data in layers that are being compacted here and have // smaller sequence numbers will be dropped in the next // few iterations of this loop (by rule (A) above). // Therefore this deletion marker is obsolete and can be dropped. drop = true; } last_sequence_for_key = ikey.sequence; } 如果不丢弃key的话，当前的写入sstable的文件不存在则调用OpenCompactionOutputFile函数创建一个文件，然后将KV写入，如果写入后sstable大小达到阈值则调用FinishCompactionOutputFile函数写出这个文件。 if (!drop) { // Open output file if necessary if (compact->builder == nullptr) { status = OpenCompactionOutputFile(compact); if (!status.ok()) { break; } } if (compact->builder->NumEntries() == 0) { compact->current_output()->smallest.DecodeFrom(key); } compact->current_output()->largest.DecodeFrom(key); compact->builder->Add(key, input->value()); // Close output file if it is big enough if (compact->builder->FileSize() >= compact->compaction->MaxOutputFileSize()) { status = FinishCompactionOutputFile(compact, input); if (!status.ok()) { break; } } } 循环以上过程： input->Next(); } 结束遍历之后，将没有写出的文件写出： if (status.ok() && shutting_down_.Acquire_Load()) { status = Status::IOError("Deleting DB during compaction"); } if (status.ok() && compact->builder != nullptr) { status = FinishCompactionOutputFile(compact, input); } if (status.ok()) { status = input->status(); } delete input; input = nullptr; 记录统计信息： CompactionStats stats; stats.micros = env_->NowMicros() - start_micros - imm_micros; for (int which = 0; which < 2; which++) { for (int i = 0; i < compact->compaction->num_input_files(which); i++) { stats.bytes_read += compact->compaction->input(which, i)->file_size; } } for (size_t i = 0; i < compact->outputs.size(); i++) { stats.bytes_written += compact->outputs[i].file_size; } mutex_.Lock(); stats_[compact->compaction->level() + 1].Add(stats); 将此次compaction生效： if (status.ok()) { status = InstallCompactionResults(compact); } if (!status.ok()) { RecordBackgroundError(status); } VersionSet::LevelSummaryStorage tmp; Log(options_.info_log, "compacted to: %s", versions_->LevelSummary(&tmp)); return status; 230 Love u
2019-04-05
优秀的个人博客,低调大师

LevelDB源码分析-MemTable

MemTable（db/memtable.h db/memtable.cc db/skiplist.h） LevelDB中存储在内存中的那部分KV数据都存储在memtable中，而memtable中的数据实际是用跳表来存储的。MemTable使用Arena进行内存管理，并提供了添加、查找、迭代器的接口，而实际上这些接口都是调用SkipList的添加和迭代器接口来实现的。 MemTable类 MemTable类的声明： class MemTable { public: // ... // 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; } } // ... // 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: // ... struct KeyComparator { const InternalKeyComparator comparator; explicit KeyComparator(const InternalKeyComparator &c) : comparator(c) {} int operator()(const char *a, const char *b) const; }; friend class MemTableIterator; friend class MemTableBackwardIterator; typedef SkipList<const char *, KeyComparator> Table; KeyComparator comparator_; int refs_; Arena arena_; Table table_; // ... }; MemTable的成员变量 table_变量即为实际存储KV数据的跳表，arena_用于内存管理，refs_用于引用计数，comparator_用于key之间的比较。 其中最简单的是ref_，调用Ref()函数则增加引用计数，调用Unref()函数减少引用次数，并且当引用次数为0时，Unref()函数会将对象自身delete。 Arena类 然后是arena_。Arena类为： class Arena { public: // ... // Return a pointer to a newly allocated memory block of "bytes" bytes. char *Allocate(size_t bytes); // Allocate memory with the normal alignment guarantees provided by malloc char *AllocateAligned(size_t bytes); // ... private: char *AllocateFallback(size_t bytes); char *AllocateNewBlock(size_t block_bytes); // Allocation state char *alloc_ptr_; size_t alloc_bytes_remaining_; // Array of new[] allocated memory blocks std::vector<char *> blocks_; // Total memory usage of the arena. port::AtomicPointer memory_usage_; // ... }; alloc_ptr_指向当前可以被分配的内存起始位置（在某一个block中），alloc_bytes_remaining_为当前可以被分配的内存的大小，blocks_中包含了指向每一个block的指针，memory_usage_为使用的内存总量，并且使用了AtomicPointer，可以通过内存屏障或者c++的atomic接口实现原子操作。 Arena类提供的方法主要有两个，一个是Allocate(size_t bytes)： inline char *Arena::Allocate(size_t bytes) char *Arena::AllocateFallback(size_t bytes) char *Arena::AllocateNewBlock(size_t block_bytes) { char *result = new char[block_bytes]; blocks_.push_back(result); memory_usage_.NoBarrier_Store( reinterpret_cast<void *>(MemoryUsage() + block_bytes + sizeof(char *))); return result; } Allocate(size_t bytes)函数分配的内存不保证对齐。首先检查当前可以被分配的内存空间是否足够，如果够的话，直接将从alloc_ptr_指向的位置开始的bytes个字节分配给调用者，并移动alloc_ptr_指向新的位置，调整alloc_bytes_remaining_： // The semantics of what to return are a bit messy if we allow // 0-byte allocations, so we disallow them here (we don't need // them for our internal use). assert(bytes > 0); if (bytes <= alloc_bytes_remaining_) { char *result = alloc_ptr_; alloc_ptr_ += bytes; alloc_bytes_remaining_ -= bytes; return result; } 如果不够，就调用AllocateFallback(size_t bytes)函数： return AllocateFallback(bytes); AllocateFallback(size_t bytes)函数首先检查bytes大小是否超过1/4个block，如果超过，则直接调用AllocateNewBlock(size_t block_bytes)函数分配一个新的block给调用者，并且这个新的block中除了当前的bytes个字节，其余部分将不会被使用，而alloc_ptr_的指向却不变，也就是说当前可以被分配的内存保持不变，因为此时剩余空间至少为1/4个block，这样设计就可以减少大块的内存碎片： if (bytes > kBlockSize / 4) { // Object is more than a quarter of our block size. Allocate it separately // to avoid wasting too much space in leftover bytes. char *result = AllocateNewBlock(bytes); return result; } 如果没有超过1/4个block，就分配一个新的block，并且将alloc_ptr_指向新的block中的剩余空间： // We waste the remaining space in the current block. alloc_ptr_ = AllocateNewBlock(kBlockSize); alloc_bytes_remaining_ = kBlockSize; char *result = alloc_ptr_; alloc_ptr_ += bytes; alloc_bytes_remaining_ -= bytes; return result; 还有一个内存分配函数是AllocateAligned(size_t bytes)： char *Arena::AllocateAligned(size_t bytes) 这个函数保证内存对齐。首先，以下代码会调整alloc_ptr_的位置使其对齐： const int align = (sizeof(void *) > 8) ? sizeof(void *) : 8; assert((align & (align - 1)) == 0); // Pointer size should be a power of 2 size_t current_mod = reinterpret_cast<uintptr_t>(alloc_ptr_) & (align - 1); size_t slop = (current_mod == 0 ? 0 : align - current_mod); size_t needed = bytes + slop; char *result; 剩下的代码和Allocate(size_t bytes)中实现一致： if (needed <= alloc_bytes_remaining_) { result = alloc_ptr_ + slop; alloc_ptr_ += needed; alloc_bytes_remaining_ -= needed; } else { // AllocateFallback always returned aligned memory result = AllocateFallback(bytes); } assert((reinterpret_cast<uintptr_t>(result) & (align - 1)) == 0); return result; SkipList类 最后是最重要的跳表，跳表由SkipList类实现： template <typename Key, class Comparator> class SkipList { private: struct Node; public: // ... // Insert key into the list. // REQUIRES: nothing that compares equal to key is currently in the list. void Insert(const Key &key); // Returns true iff an entry that compares equal to key is in the list. bool Contains(const Key &key) const; // Iteration over the contents of a skip list class Iterator { // ... }; private: // ... // Immutable after construction Comparator const compare_; Arena *const arena_; // Arena used for allocations of nodes Node *const head_; // Modified only by Insert(). Read racily by readers, but stale // values are ok. port::AtomicPointer max_height_; // Height of the entire list // ... Node *NewNode(const Key &key, int height); // ... // Return true if key is greater than the data stored in "n" bool KeyIsAfterNode(const Key &key, Node *n) const; // Return the earliest node that comes at or after key. // Return nullptr if there is no such node. // // If prev is non-null, fills prev[level] with pointer to previous // node at "level" for every level in [0..max_height_-1]. Node *FindGreaterOrEqual(const Key &key, Node **prev) const; // Return the latest node with a key < key. // Return head_ if there is no such node. Node *FindLessThan(const Key &key) const; // Return the last node in the list. // Return head_ if list is empty. Node *FindLast() const; // ... }; 跳表中主要有四个成员，一个是compare_用于key的比较，一个是arena_用于内存管理，一个是head_指向跳表开头的节点，最后一个是max_height_为跳表最大的高度。 当然跳表中还需要两个类和结构，一个是Iterator类，用于跳表的遍历，另一个是Node结构，用于表示跳表中的节点。 Node结构 首先是Node结构： template <typename Key, class Comparator> struct SkipList<Key, Comparator>::Node { // ... Key const key; // Accessors/mutators for links. Wrapped in methods so we can // add the appropriate barriers as necessary. Node *Next(int n) // ... void SetNext(int n, Node *x) // ... // No-barrier variants that can be safely used in a few locations. Node *NoBarrier_Next(int n) // ... void NoBarrier_SetNext(int n, Node *x) // ... private: // Array of length equal to the node height. next_[0] is lowest level link. port::AtomicPointer next_[1]; }; Node中key存储了节点中实际存储的KV值，next_为一个变长数组（struct最后一个元素如果是数组，则长度可变），用于存储该节点在不同高度的链表上指向下一个节点的指针。 创建Node使用了一个很神奇的函数： template <typename Key, class Comparator> typename SkipList<Key, Comparator>::Node * SkipList<Key, Comparator>::NewNode(const Key &key, int height) { char *mem = arena_->AllocateAligned( sizeof(Node) + sizeof(port::AtomicPointer) * (height - 1)); return new (mem) Node(key); } 这个函数首先分配了一块对齐的内存，然后通过new (mem) Node(key)这个语句创建一个节点，我也不是很懂。 Iterator类 其次是iterator类： // Iteration over the contents of a skip list class Iterator { public: // ... // Returns true iff the iterator is positioned at a valid node. bool Valid() const; // Returns the key at the current position. // REQUIRES: Valid() const Key &key() const; // Advances to the next position. // REQUIRES: Valid() void Next(); // Advances to the previous position. // REQUIRES: Valid() void Prev(); // Advance to the first entry with a key >= target void Seek(const Key &target); // Position at the first entry in list. // Final state of iterator is Valid() iff list is not empty. void SeekToFirst(); // Position at the last entry in list. // Final state of iterator is Valid() iff list is not empty. void SeekToLast(); private: const SkipList *list_; Node *node_; // ... }; 这个类主要有两个成员，一个list_指向迭代的跳表，node_指向当前迭代的节点。 对跳表的遍历操作需要获取一个Iterator类型的迭代器，然后通过调用Iterator类提供的接口函数对跳表进行遍历。而Iterator类提供的接口函数实际上通过调用SkipList类提供的私有接口函数实现。 SkipList类的私有接口函数如下： template <typename Key, class Comparator> typename SkipList<Key, Comparator>::Node *SkipList<Key, Comparator>::FindGreaterOrEqual(const Key &key, Node **prev) const { Node *x = head_; int level = GetMaxHeight() - 1; while (true) { Node *next = x->Next(level); if (KeyIsAfterNode(key, next)) { // Keep searching in this list x = next; } else { if (prev != nullptr) prev[level] = x; if (level == 0) { return next; } else { // Switch to next list level--; } } } } template <typename Key, class Comparator> typename SkipList<Key, Comparator>::Node * SkipList<Key, Comparator>::FindLessThan(const Key &key) const { Node *x = head_; int level = GetMaxHeight() - 1; while (true) { assert(x == head_ || compare_(x->key, key) < 0); Node *next = x->Next(level); if (next == nullptr || compare_(next->key, key) >= 0) { if (level == 0) { return x; } else { // Switch to next list level--; } } else { x = next; } } } template <typename Key, class Comparator> typename SkipList<Key, Comparator>::Node *SkipList<Key, Comparator>::FindLast() const { Node *x = head_; int level = GetMaxHeight() - 1; while (true) { Node *next = x->Next(level); if (next == nullptr) { if (level == 0) { return x; } else { // Switch to next list level--; } } else { x = next; } } } 这三个函数的实现比较简单，类似于链表的遍历，就不做具体分析了。 另外，跳表还有两个接口函数用于插入和检查是否包含某个key，这两个函数的实现也使用了跳表的私有接口函数： template <typename Key, class Comparator> void SkipList<Key, Comparator>::Insert(const Key &key) template <typename Key, class Comparator> bool SkipList<Key, Comparator>::Contains(const Key &key) const MemTable成员函数 分析完以上这些，我们开始分析MemTable提供的接口函数： 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()] // 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 *buf = arena_.Allocate(encoded_len); char *p = EncodeVarint32(buf, internal_key_size); memcpy(p, key.data(), key_size); p += key_size; EncodeFixed64(p, (s << 8) | type); p += 8; p = EncodeVarint32(p, val_size); memcpy(p, value.data(), val_size); assert(p + val_size == buf + encoded_len); table_.Insert(buf); } Add函数将key和value组成key_size | sequencenumber | type | key | value_size | value这样的形式，然后调用SkipList类提供的Insert函数插入跳表中。 bool MemTable::Get(const LookupKey &key, std::string *value, Status *s) Get函数先用跳表的迭代器找到包含key的节点： Slice memkey = key.memtable_key(); Table::Iterator iter(&table_); iter.Seek(memkey.data()); 然后将节点中的数据解码后封装为Slice存入value： 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 (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<ValueType>(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; } } } return false; MemTableIterator类 MemTable还封装了一个MemTableIterator类用于MemTable的迭代： class MemTableIterator : public Iterator { public: explicit MemTableIterator(MemTable::Table *table) : iter_(table) {} virtual bool Valid() const { return iter_.Valid(); } virtual void Seek(const Slice &k) { iter_.Seek(EncodeKey(&tmp_, k)); } virtual void SeekToFirst() { iter_.SeekToFirst(); } virtual void SeekToLast() { iter_.SeekToLast(); } virtual void Next() { iter_.Next(); } virtual void Prev() { iter_.Prev(); } virtual Slice key() const { return GetLengthPrefixedSlice(iter_.key()); } virtual Slice value() const { Slice key_slice = GetLengthPrefixedSlice(iter_.key()); return GetLengthPrefixedSlice(key_slice.data() + key_slice.size()); } virtual Status status() const { return Status::OK(); } private: MemTable::Table::Iterator iter_; std::string tmp_; // For passing to EncodeKey // No copying allowed MemTableIterator(const MemTableIterator &); void operator=(const MemTableIterator &); }; 遍历MemTable就需要获得一个MemTableIterator类的迭代器，然后通过调用MemTableIterator类提供的接口函数进行遍历。MemTableIterator类其实就是实现了对跳表迭代器的封装，成员为一个跳表的迭代器iter_，接口函数通过调用跳表迭代器的接口函数实现。 227 Love u
2019-04-05
优秀的个人博客,低调大师

LevelDB源码分析-编码

编码（util/coding.h util/coding.cc） LevelDB将整型编码为二进制字符串的形式，同时又能够和ASCII字符区分。 首先是定长编码： void EncodeFixed32(char *buf, uint32_t value) { if (port::kLittleEndian) { memcpy(buf, &value, sizeof(value)); } else { buf[0] = value & 0xff; buf[1] = (value >> 8) & 0xff; buf[2] = (value >> 16) & 0xff; buf[3] = (value >> 24) & 0xff; } } void EncodeFixed64(char *buf, uint64_t value) { if (port::kLittleEndian) { memcpy(buf, &value, sizeof(value)); } else { buf[0] = value & 0xff; buf[1] = (value >> 8) & 0xff; buf[2] = (value >> 16) & 0xff; buf[3] = (value >> 24) & 0xff; buf[4] = (value >> 32) & 0xff; buf[5] = (value >> 40) & 0xff; buf[6] = (value >> 48) & 0xff; buf[7] = (value >> 56) & 0xff; } } 这里根据机器区分大端和小端，LevelDB编码后的字符串为小端存储。在编码时，只是简单的将8位二进制码存储在一个char字符的位置上。因为定长，所以可以和ASCII字符区分。 接下来是定长编码的一些接口函数： void PutFixed32(std::string *dst, uint32_t value) // 将一个32位整型值定长编码后存入dst void PutFixed64(std::string *dst, uint64_t value) // 将一个64位整型值定长编码后存入dst 然后是变长编码： char *EncodeVarint32(char *dst, uint32_t v) { // Operate on characters as unsigneds unsigned char *ptr = reinterpret_cast<unsigned char *>(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<char *>(ptr); } char *EncodeVarint64(char *dst, uint64_t v) { static const int B = 128; unsigned char *ptr = reinterpret_cast<unsigned char *>(dst); while (v >= B) { *(ptr++) = (v & (B - 1)) | B; v >>= 7; } *(ptr++) = static_cast<unsigned char>(v); return reinterpret_cast<char *>(ptr); } LevelDB的变长编码设计的十分巧妙，它以7个二进制bit为一个单位，存入一个char中，同时为了和ASCII码进行区分，将char的最高位设为1（ASCII码为0-127），同样采用小端存储的形式。但是变长编码的最后一个char的最高位是0，以此作为变长编码后的字符串的结束标志。 例如，11001011101111001会被编码为11111001 10101110 00000110。 接下来是一些变长编码的接口函数： void PutVarint32(std::string *dst, uint32_t v) // 将一个32位整型值变长编码后存入dst void PutVarint64(std::string *dst, uint64_t v) // 将一个64位整型值变长编码后存入dst int VarintLength(uint64_t v) // 获取变长编码后的字符串长度(以字节计数) const char *GetVarint32PtrFallback(const char *p, const char *limit, uint32_t *value) // 将以p到limit之间的变长编码字符串解码为32位整型值 bool GetVarint32(Slice *input, uint32_t *value) // 将以p到limit之间的变长编码字符串解码为32位整型值并封装入Slice中 const char *GetVarint64Ptr(const char *p, const char *limit, uint64_t *value) // 将以p到limit之间的变长编码字符串解码为64位整型值 bool GetVarint64(Slice *input, uint64_t *value) // 将以p到limit之间的变长编码字符串解码为64位整型值并封装入Slice中 227 Love u
2019-04-05
优秀的个人博客,低调大师

Erlang PoolBoy 源码分析

问题存疑： 1.如何释放使用过的进程，被再次利用 答：通过try after 的方式，try checkout锁定一个进程，after 用完之后释放被锁定的进程 2.如果出现初始进程不够了，如果处理 答：如果不够用，判定max_overflow参数，确定是否需要重新创建进程，或者将此使用者放入等待列表中，一旦有新释放的，就马上给他使用，受超时器的影响 3.创建新的进程，是如何释放的 答：如果此进程是弹性创建的，则直接释放此进程，如果是work中的进程，则释放使用权 4.如果达到最大的创建进程数，又是如何处理的 答：根据block参数确定，是否忽略或者加入等待列表中 5.如果创建的进程 答：poolboy_sup 通过sample_one_by_one参数形式，动态调用supervisor:start_child创建进程池中的进程 6.为什么需要将初始创建的进程和poolboy进程link起来，有何用意 答：这样就可以将所有的逻辑全部集中在poolboy进程，只是将poolboy_sup当作一个中间件来使用创建进程池中的进程 queue 队列 支持先进先出或者后进先出模式 基本思路 1.通过一个外部sup启动一个poolboy进程，poolboy进程再启动一个poolboy_sup进程 2.poolboy_sup 就是这个进程池，然后启动若干个进程，但是poolboy进程操纵poolboy_sup进程创建很多个进程，然后将这些进程全部保存在poolboy进程 poolboy进程并且与poolboy_sup进程创建的进程进行link 当外部进程需要进程时直接从poolboy进程入口，然后使用，并且poolboy进程还监控所有使用的外部进程 当外部进程down掉后，poolboy进行善后操作，每一个外部进程独享poolboy进程中的一个进程 通过try after 方法，先独享进程，用完后，再释放进程 link的还有一个操作，就是处理poolboy进程池中进程挂掉的情况，先关闭此进程，并进行善后操作，然后，重新启用一个新的进程，带起挂掉的进程 主要的接口就是checkout和checkin的两个接口创建 数据结构采用queue，有两种方式可选，fifo(先进先出),fofi(后进先出)的方式，遍历整个进程池 还采用了一种动态创建进程的机制，就是说，如果发现现有创建的进程不够用了，那就需要创建一个临时的进程，满足使用，当使用完之后，再释放此进程 如果在释放的时候，发现还有等待的进程，则继续使用释放的进程，进行执行 启动 1.创建两个数据结构，wait:queue:new 保存等待的进程吃中分配的进程,monitors=ets:new()创建一个监控使用进程池中进程的数据 2.创建poolboy_sup 监督进程,采用sample_one_by_one 启动，后面采用supervisor:start_child(Sup,Args)启动进程 3.设置启动参数: worker_module:启动进程池的进程模块名 size:进程池中常见size个进程 max_overflow:最多可以弹性创建进程的数量 starategy:lifo|fifo 后进先出|先进先出模式 (一版来说，进程池中的进程没有区别的) Block:true 允许进程获取等待 false 不允许进程获取等待，在checkout接口中设置 4.启动进程池进程 supervisor:start_child(Sup,Args)启动子进程，填充进程池 使用 调用poolboy:transaction(PoolName,Query) 工作机制： 1.首先调用checkout获取进程池中的一个进程，调用的是call同步的方法执行，并设置超时时间 2.使用完毕之后，调用checkin释放进程使用 3.如果出现进程池没有进程，就可以通过max_overflow参数配置，确认是否可以生成新的进程以供使用 4.如果没有，则可以通过Block参数，确认是否可以将当前请求放入waiting中，只要有进程释放，就将此进程返回 5.进程使用完毕，则释放此进程的使用权，清理监控数据列表，然后判定是否弹性创建，决定是否释放，如果有等待进程，则将此进程使用权转移使用 异常处理 1.使用进程挂掉：调用down的handle_info，清理监控列表和工作列表，以及等待列表 2.进程池中进程挂掉：删除监控，重新生成一个进程 执行函数 checkout 获取一个可用进程 checkin 释放一个使用进程 transaction 一个完整的获取和释放进程过程 可能的不足之处 短暂 I/O 任务（比如数据库访问）的大型 worker 池会因为太多迁入和迁出活动而拖垮单个 poolboy 管理器。 参考： Erlang 池管理 maestro Erlang：https://www.oschina.net/p/maestro-erlang
2019-01-21

资源下载

更多资源
Nacos

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

时间: 2026-09-24 大小: 189MB 下载: 9次
Rocky Linux

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。

时间: 2026-08-21 大小: 1.42GB 下载: 6次
Sublime Text

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。

时间: 2026-08-21 大小: 9.84MB 下载: 40次
WebStorm

WebStorm

WebStorm 是jetbrains公司旗下一款JavaScript 开发工具。目前已经被广大中国JS开发者誉为“Web前端开发神器”、“最强大的HTML5编辑器”、“最智能的JavaScript IDE”等。与IntelliJ IDEA同源，继承了IntelliJ IDEA强大的JS部分的功能。

时间: 2025-12-13 大小: 249.54MB 下载: 22次
热门标签
DeepSeek Jdk25 OpenAi HarmonyOS6 Nacos3 SpringBoot4 SpringCloud Docker Kubernetes Service Mesh Redis Gemini3 Rocky

欢迎您来访！

登录后，您将获得更多功能！
热门文章
联系我们

有任何问题或合作意向欢迎联系我们

Email: 99873273@qq.com

QQ: 99873273

用户登录
用户注册