10.1.1 cache.h

class Cache {public:Cache() { }//Destroys all existing entries by calling the "deleter"//function that was passed to the constructor.virtual ~Cache();//Opaque handle to an entry stored in the cache.struct Handle { };//Insert a mapping from key->value into the cache and assign it//the specified charge against the total cache capacity.////Returns a handle that corresponds to the mapping.  The caller//must call this->Release(handle) when the returned mapping is no//longer needed.////When the inserted entry is no longer needed, the key and//value will be passed to "deleter".virtual Handle* Insert(const Slice& key, void* value, size_t charge,void (*deleter)(const Slice& key, void* value)) = 0;//If the cache has no mapping for "key", returns NULL.////Else return a handle that corresponds to the mapping.  The caller//must call this->Release(handle) when the returned mapping is no//longer needed.virtual Handle* Lookup(const Slice& key) = 0;//Release a mapping returned by a previous Lookup().//REQUIRES: handle must not have been released yet.//REQUIRES: handle must have been returned by a method on *this.virtual void Release(Handle* handle) = 0;//Return the value encapsulated in a handle returned by a//successful Lookup().//REQUIRES: handle must not have been released yet.//REQUIRES: handle must have been returned by a method on *this.virtual void* Value(Handle* handle) = 0;//If the cache contains entry for key, erase it.  Note that the//underlying entry will be kept around until all existing handles//to it have been released.virtual void Erase(const Slice& key) = 0;//Return a new numeric id.  May be used by multiple clients who are//sharing the same cache to partition the key space.  Typically the//client will allocate a new id at startup and prepend the id to//its cache keys.virtual uint64_t NewId() = 0;private:void LRU_Remove(Handle* e);void LRU_Append(Handle* e);void Unref(Handle* e); //可以看到Handle可能实际上底层有引用计数存在.struct Rep;Rep* rep_; //opaque实现指针的存在.
};

extern Cache* NewLRUCache(size_t capacity);
Cache* NewLRUCache(size_t capacity) {return new ShardedLRUCache(capacity);
}

10.1.2 comparator.h

//Advanced functions: these are used to reduce the space requirements
//for internal data structures like index blocks.//If *start < limit, changes *start to a short string in [start,limit).
//Simple comparator implementations may return with *start unchanged,
//i.e., an implementation of this method that does nothing is correct.
virtual void FindShortestSeparator(std::string* start,const Slice& limit) const = 0;//Changes *key to a short string >= *key.
//Simple comparator implementations may return with *key unchanged,
//i.e., an implementation of this method that does nothing is correct.
virtual void FindShortSuccessor(std::string* key) const = 0;

//Intentionally not destroyed to prevent destructor racing
//with background threads.
static const Comparator* bytewise = new BytewiseComparatorImpl;const Comparator* BytewiseComparator() {return bytewise;
}

10.1.3 db.h

//A DB is a persistent ordered map from keys to values.
//A DB is safe for concurrent access from multiple threads without
//any external synchronization.
class DB {public://Open the database with the specified "name".//Stores a pointer to a heap-allocated database in *dbptr and returns//OK on success.//Stores NULL in *dbptr and returns a non-OK status on error.//Caller should delete *dbptr when it is no longer needed.static Status Open(const Options& options,const std::string& name,DB** dbptr);DB() { }virtual ~DB();//Set the database entry for "key" to "value".  Returns OK on success,//and a non-OK status on error.//Note: consider setting options.sync = true.virtual Status Put(const WriteOptions& options,const Slice& key,const Slice& value) = 0;//Remove the database entry (if any) for "key".  Returns OK on//success, and a non-OK status on error.  It is not an error if "key"//did not exist in the database.//Note: consider setting options.sync = true.virtual Status Delete(const WriteOptions& options, const Slice& key) = 0;//Apply the specified updates to the database.//Returns OK on success, non-OK on failure.//Note: consider setting options.sync = true.virtual Status Write(const WriteOptions& options, WriteBatch* updates) = 0;//If the database contains an entry for "key" store the//corresponding value in *value and return OK.////If there is no entry for "key" leave *value unchanged and return//a status for which Status::IsNotFound() returns true.////May return some other Status on an error.virtual Status Get(const ReadOptions& options,const Slice& key, std::string* value) = 0;//Return a heap-allocated iterator over the contents of the database.//The result of NewIterator() is initially invalid (caller must//call one of the Seek methods on the iterator before using it).////Caller should delete the iterator when it is no longer needed.//The returned iterator should be deleted before this db is deleted.virtual Iterator* NewIterator(const ReadOptions& options) = 0;//Return a handle to the current DB state.  Iterators created with//this handle will all observe a stable snapshot of the current DB//state.  The caller must call ReleaseSnapshot(result) when the//snapshot is no longer needed.virtual const Snapshot* GetSnapshot() = 0;//Release a previously acquired snapshot.  The caller must not//use "snapshot" after this call.virtual void ReleaseSnapshot(const Snapshot* snapshot) = 0;//DB implementations can export properties about their state//via this method.  If "property" is a valid property understood by this//DB implementation, fills "*value" with its current value and returns//true.  Otherwise returns false.//////Valid property names include:////"leveldb.num-files-at-level<N>" - return the number of files at level <N>,//where <N> is an ASCII representation of a level number (e.g. "0").//"leveldb.stats" - returns a multi-line string that describes statistics//about the internal operation of the DB.//"leveldb.sstables" - returns a multi-line string that describes all//of the sstables that make up the db contents.virtual bool GetProperty(const Slice& property, std::string* value) = 0;//For each i in [0,n-1], store in "sizes[i]", the approximate//file system space used by keys in "[range[i].start .. range[i].limit)".////Note that the returned sizes measure file system space usage, so//if the user data compresses by a factor of ten, the returned//sizes will be one-tenth the size of the corresponding user data size.////The results may not include the sizes of recently written data.virtual void GetApproximateSizes(const Range* range, int n,uint64_t* sizes) = 0;//Compact the underlying storage for the key range [*begin,*end].//In particular, deleted and overwritten versions are discarded,//and the data is rearranged to reduce the cost of operations//needed to access the data.  This operation should typically only//be invoked by users who understand the underlying implementation.////begin==NULL is treated as a key before all keys in the database.//end==NULL is treated as a key after all keys in the database.//Therefore the following call will compact the entire database://db->CompactRange(NULL, NULL);virtual void CompactRange(const Slice* begin, const Slice* end) = 0;
};

//Destroy the contents of the specified database.
//Be very careful using this method.
Status DestroyDB(const std::string& name, const Options& options);//If a DB cannot be opened, you may attempt to call this method to
//resurrect as much of the contents of the database as possible.
//Some data may be lost, so be careful when calling this function
//on a database that contains important information.
Status RepairDB(const std::string& dbname, const Options& options);

10.1.4 env.h

1. Env

   class Env {public:Env() { }virtual ~Env();//Return a default environment suitable for the current operating//system.  Sophisticated users may wish to provide their own Env//implementation instead of relying on this default environment.////The result of Default() belongs to leveldb and must never be deleted.static Env* Default();//Create a brand new sequentially-readable file with the specified name.//On success, stores a pointer to the new file in *result and returns OK.//On failure stores NULL in *result and returns non-OK.  If the file does//not exist, returns a non-OK status.////The returned file will only be accessed by one thread at a time.virtual Status NewSequentialFile(const std::string& fname,SequentialFile** result) = 0;//Create a brand new random access read-only file with the//specified name.  On success, stores a pointer to the new file in//*result and returns OK.  On failure stores NULL in *result and//returns non-OK.  If the file does not exist, returns a non-OK//status.////The returned file may be concurrently accessed by multiple threads.virtual Status NewRandomAccessFile(const std::string& fname,RandomAccessFile** result) = 0;//Create an object that writes to a new file with the specified//name.  Deletes any existing file with the same name and creates a//new file.  On success, stores a pointer to the new file in//*result and returns OK.  On failure stores NULL in *result and//returns non-OK.////The returned file will only be accessed by one thread at a time.virtual Status NewWritableFile(const std::string& fname,WritableFile** result) = 0;//Returns true iff the named file exists.virtual bool FileExists(const std::string& fname) = 0;//Store in *result the names of the children of the specified directory.//The names are relative to "dir".//Original contents of *results are dropped.virtual Status GetChildren(const std::string& dir,std::vector<std::string>* result) = 0;//Delete the named file.virtual Status DeleteFile(const std::string& fname) = 0;//Create the specified directory.virtual Status CreateDir(const std::string& dirname) = 0;//Delete the specified directory.virtual Status DeleteDir(const std::string& dirname) = 0;//Store the size of fname in *file_size.virtual Status GetFileSize(const std::string& fname, uint64_t* file_size) = 0;//Rename file src to target.virtual Status RenameFile(const std::string& src,const std::string& target) = 0;//Lock the specified file.  Used to prevent concurrent access to//the same db by multiple processes.  On failure, stores NULL in//*lock and returns non-OK.////On success, stores a pointer to the object that represents the//acquired lock in *lock and returns OK.  The caller should call//UnlockFile(*lock) to release the lock.  If the process exits,//the lock will be automatically released.////If somebody else already holds the lock, finishes immediately//with a failure.  I.e., this call does not wait for existing locks//to go away.////May create the named file if it does not already exist.virtual Status LockFile(const std::string& fname, FileLock** lock) = 0;//Release the lock acquired by a previous successful call to LockFile.//REQUIRES: lock was returned by a successful LockFile() call//REQUIRES: lock has not already been unlocked.virtual Status UnlockFile(FileLock* lock) = 0;//Arrange to run "(*function)(arg)" once in a background thread.////"function" may run in an unspecified thread.  Multiple functions//added to the same Env may run concurrently in different threads.//I.e., the caller may not assume that background work items are//serialized.virtual void Schedule(void (*function)(void* arg),void* arg) = 0;//Start a new thread, invoking "function(arg)" within the new thread.//When "function(arg)" returns, the thread will be destroyed.virtual void StartThread(void (*function)(void* arg), void* arg) = 0;//*path is set to a temporary directory that can be used for testing. It may//or many not have just been created. The directory may or may not differ//between runs of the same process, but subsequent calls will return the//same directory.virtual Status GetTestDirectory(std::string* path) = 0;//Create and return a log file for storing informational messages.virtual Status NewLogger(const std::string& fname, Logger** result) = 0;//Returns the number of micro-seconds since some fixed point in time. Only//useful for computing deltas of time.virtual uint64_t NowMicros() = 0;//Sleep/delay the thread for the perscribed number of micro-seconds.virtual void SleepForMicroseconds(int micros) = 0;
   };
   

   稍微总结一下这些接口.对于这些接口创建的抽象对象，在后面也会详细解释。实现是PosixEnv.

   • Default // 获得默认的环境
   • NewSequentialFile // 创建顺序文件
   • NewRandomAccessFile // 创建随机文件
   • NewWritableFile // 创建可写文件
   • FileExists // 文件是否存在
   • GetChildren // 目录下面的文件
   • DeleteFile // 删除文件
   • CreateDir // 创建目录
   • DeleteDir // 删除目录
   • GetFileSize // filesize.
   • RenameFile // rename.
   • LockFile // 锁住文件
   • UnlockFile // 解锁文件
   • StartThread // 创建线程
   • GetTestDirectory // 测试目录
   • NewLogger // 创建logger打印到对应文件
   • NowMicros // 当前us.
   • SleepForMicroseconds // sleep us

   环境还提供了一个EnvWrapper.这个类就是得到一个Env*对象然后重新转发出去。

2. FileLock

   FileLock接口非常简单，可以说就没有接口.唯一要做的事情就是和Env里面的LockFile与UnlockFile配合。实现是PosixFileLock.

   //Identifies a locked file.
   class FileLock {public:FileLock() { }virtual ~FileLock();private://No copying allowedFileLock(const FileLock&);void operator=(const FileLock&);
   };
   

3. Logger

   Logger接口也非常简单，就是Logv.允许打印变长参数.实现是PosixLogger.

   //An interface for writing log messages.
   class Logger {public:Logger() { }virtual ~Logger();//Write an entry to the log file with the specified format.virtual void Logv(const char* format, va_list ap) = 0;private://No copying allowedLogger(const Logger&);void operator=(const Logger&);
   };
   

   为了编写方便提供了这么一个宏

   //Log the specified data to *info_log if info_log is non-NULL.
   extern void Log(Logger* info_log, const char* format, ...)
   #   if defined(__GNUC__) || defined(__clang__)__attribute__((__format__ (__printf__, 2, 3)))
   #   endif;
   

4. RandomAccessFile

   RandomAccessFile所提供的语义就是能够随机从offset读取n个字节，存放在scratch里面。然后将result里面的内容指向scratch.随机读取并且要求能够多线程安全。实现是PosixRandomAccessFile.

   //A file abstraction for randomly reading the contents of a file.
   class RandomAccessFile {public:RandomAccessFile() { }virtual ~RandomAccessFile();//Read up to "n" bytes from the file starting at "offset".//"scratch[0..n-1]" may be written by this routine.  Sets "*result"//to the data that was read (including if fewer than "n" bytes were//successfully read).  May set "*result" to point at data in//"scratch[0..n-1]", so "scratch[0..n-1]" must be live when//"*result" is used.  If an error was encountered, returns a non-OK//status.////Safe for concurrent use by multiple threads.virtual Status Read(uint64_t offset, size_t n, Slice* result,char* scratch) const = 0;
   };
   

5. SequentialFile

   SequentialFile提供的语义包括顺序读取以及Skip字节。这些都是外部来保证同步的。实现是PosixSequentialFile.

   //A file abstraction for reading sequentially through a file
   class SequentialFile {public:SequentialFile() { }virtual ~SequentialFile();//Read up to "n" bytes from the file.  "scratch[0..n-1]" may be//written by this routine.  Sets "*result" to the data that was//read (including if fewer than "n" bytes were successfully read).//May set "*result" to point at data in "scratch[0..n-1]", so//"scratch[0..n-1]" must be live when "*result" is used.//If an error was encountered, returns a non-OK status.////REQUIRES: External synchronizationvirtual Status Read(size_t n, Slice* result, char* scratch) = 0;//Skip "n" bytes from the file. This is guaranteed to be no//slower that reading the same data, but may be faster.////If end of file is reached, skipping will stop at the end of the//file, and Skip will return OK.////REQUIRES: External synchronizationvirtual Status Skip(uint64_t n) = 0;
   };
   

6. WritableFile

   WritableFile语义就是允许Append,Close,Flush,Sync.这里Flush的语义应该是将内部缓存数据完全写入，而Sync表示让磁盘进行同步。因为可能外部会调用小对象的写入，所以这里需要进行缓存。实现是PosixMmapFile.

   //A file abstraction for sequential writing.  The implementation
   //must provide buffering since callers may append small fragments
   //at a time to the file.
   class WritableFile {public:WritableFile() { }virtual ~WritableFile();virtual Status Append(const Slice& data) = 0;virtual Status Close() = 0;virtual Status Flush() = 0;virtual Status Sync() = 0;private://No copying allowedWritableFile(const WritableFile&);void operator=(const WritableFile&);
   };
   

10.1.5 filter_policy.h

class FilterPolicy {public:virtual ~FilterPolicy();//用来做兼容判断。如果新的filter policy发生改变的话，那么这个名字也必须变化。//Return the name of this policy.  Note that if the filter encoding//changes in an incompatible way, the name returned by this method//must be changed.  Otherwise, old incompatible filters may be//passed to methods of this type.virtual const char* Name() const = 0;//keys都是排好序的，将这些keys加入filter.相当于告知这些keys已经存在。//keys[0,n-1] contains a list of keys (potentially with duplicates)//that are ordered according to the user supplied comparator.//Append a filter that summarizes keys[0,n-1] to *dst.////Warning: do not change the initial contents of *dst.  Instead,//append the newly constructed filter to *dst.virtual void CreateFilter(const Slice* keys, int n, std::string* dst)const = 0;//判断key是否在filter里面。这个filter是在CreateFilter里面的dst包装出来的。//"filter" contains the data appended by a preceding call to//CreateFilter() on this class.  This method must return true if//the key was in the list of keys passed to CreateFilter().//This method may return true or false if the key was not on the//list, but it should aim to return false with a high probability.virtual bool KeyMayMatch(const Slice& key, const Slice& filter) const = 0;
};//Return a new filter policy that uses a bloom filter with approximately
//the specified number of bits per key.  A good value for bits_per_key
//is 10, which yields a filter with ~ 1% false positive rate.
//
//Callers must delete the result after any database that is using the
//result has been closed.
//
//Note: if you are using a custom comparator that ignores some parts
//of the keys being compared, you must not use NewBloomFilterPolicy()
//and must provide your own FilterPolicy that also ignores the
//corresponding parts of the keys.  For example, if the comparator
//ignores trailing spaces, it would be incorrect to use a
//FilterPolicy (like NewBloomFilterPolicy) that does not ignore
//trailing spaces in keys.
extern const FilterPolicy* NewBloomFilterPolicy(int bits_per_key);

10.1.6 iterator.h

class Iterator {public:Iterator();virtual ~Iterator();//An iterator is either positioned at a key/value pair, or//not valid.  This method returns true iff the iterator is valid.virtual bool Valid() const = 0;//Position at the first key in the source.  The iterator is Valid()//after this call iff the source is not empty.virtual void SeekToFirst() = 0;//Position at the last key in the source.  The iterator is//Valid() after this call iff the source is not empty.virtual void SeekToLast() = 0;//Position at the first key in the source that at or past target//The iterator is Valid() after this call iff the source contains//an entry that comes at or past target.virtual void Seek(const Slice& target) = 0;//Moves to the next entry in the source.  After this call, Valid() is//true iff the iterator was not positioned at the last entry in the source.//REQUIRES: Valid()virtual void Next() = 0;//Moves to the previous entry in the source.  After this call, Valid() is//true iff the iterator was not positioned at the first entry in source.//REQUIRES: Valid()virtual void Prev() = 0;//Return the key for the current entry.  The underlying storage for//the returned slice is valid only until the next modification of//the iterator.//REQUIRES: Valid()virtual Slice key() const = 0;//Return the value for the current entry.  The underlying storage for//the returned slice is valid only until the next modification of//the iterator.//REQUIRES: !AtEnd() && !AtStart()virtual Slice value() const = 0;//If an error has occurred, return it.  Else return an ok status.virtual Status status() const = 0;//Clients are allowed to register function/arg1/arg2 triples that//will be invoked when this iterator is destroyed.////Note that unlike all of the preceding methods, this method is//not abstract and therefore clients should not override it.typedef void (*CleanupFunction)(void* arg1, void* arg2);void RegisterCleanup(CleanupFunction function, void* arg1, void* arg2);private:struct Cleanup {CleanupFunction function;void* arg1;void* arg2;Cleanup* next;};Cleanup cleanup_;//No copying allowedIterator(const Iterator&);void operator=(const Iterator&);
};

Iterator::Iterator() {cleanup_.function = NULL;cleanup_.next = NULL;
}Iterator::~Iterator() {if (cleanup_.function != NULL) {(*cleanup_.function)(cleanup_.arg1, cleanup_.arg2);for (Cleanup* c = cleanup_.next; c != NULL; ) {(*c->function)(c->arg1, c->arg2);Cleanup* next = c->next;delete c;c = next;}}
}void Iterator::RegisterCleanup(CleanupFunction func, void* arg1, void* arg2) {assert(func != NULL);Cleanup* c;if (cleanup_.function == NULL) {c = &cleanup_;} else {c = new Cleanup;c->next = cleanup_.next;cleanup_.next = c;}c->function = func;c->arg1 = arg1;c->arg2 = arg2;
}

10.1.7 options.h

1. Options

   Options控制DB行为，在DB::Open时候就传入.我们需要针对这些字段仔细看看.

   //DB contents are stored in a set of blocks, each of which holds a
   //sequence of key,value pairs.  Each block may be compressed before
   //being stored in a file.  The following enum describes which
   //compression method (if any) is used to compress a block.
   enum CompressionType {//NOTE: do not change the values of existing entries, as these are//part of the persistent format on disk.kNoCompression     = 0x0,kSnappyCompression = 0x1
   };//Options to control the behavior of a database (passed to DB::Open)
   struct Options {//-------------------//Parameters that affect behavior//Comparator used to define the order of keys in the table.//Default: a comparator that uses lexicographic byte-wise ordering////REQUIRES: The client must ensure that the comparator supplied//here has the same name and orders keys *exactly* the same as the//comparator provided to previous open calls on the same DB.const Comparator* comparator; //如何进行slice compare//If true, the database will be created if it is missing.//Default: falsebool create_if_missing; //database不存在是否需要创建.//If true, an error is raised if the database already exists.//Default: falsebool error_if_exists; //如果数据库存在是否error//If true, the implementation will do aggressive checking of the//data it is processing and will stop early if it detects any//errors.  This may have unforeseen ramifications: for example, a//corruption of one DB entry may cause a large number of entries to//become unreadable or for the entire DB to become unopenable.//Default: falsebool paranoid_checks; //会更多地检查数据正确性.这个在日志回放的时候有效.//MaybeIgnoreError里面使用.//Use the specified object to interact with the environment,//e.g. to read/write files, schedule background work, etc.//Default: Env::Default()Env* env; //操作环境.//Any internal progress/error information generated by the db will//be written to info_log if it is non-NULL, or to a file stored//in the same directory as the DB contents if info_log is NULL.//Default: NULLLogger* info_log; //Logger.//-------------------//Parameters that affect performance//Amount of data to build up in memory (backed by an unsorted log//on disk) before converting to a sorted on-disk file.////Larger values increase performance, especially during bulk loads.//Up to two write buffers may be held in memory at the same time,//so you may wish to adjust this parameter to control memory usage.//Also, a larger write buffer will result in a longer recovery time//the next time the database is opened.////Default: 4MBsize_t write_buffer_size; //memtable超过这些内存大小的话就会写table.//Number of open files that can be used by the DB.  You may need to//increase this if your database has a large working set (budget//one open file per 2MB of working set).////Default: 1000int max_open_files; //最大文件句柄数.//Control over blocks (user data is stored in a set of blocks, and//a block is the unit of reading from disk).//If non-NULL, use the specified cache for blocks.//If NULL, leveldb will automatically create and use an 8MB internal cache.//Default: NULLCache* block_cache; //内部cahce.//Approximate size of user data packed per block.  Note that the//block size specified here corresponds to uncompressed data.  The//actual size of the unit read from disk may be smaller if//compression is enabled.  This parameter can be changed dynamically.////Default: 4Ksize_t block_size; //SSTable里面的Block大小.具体可以看BlockBuilder.//Number of keys between restart points for delta encoding of keys.//This parameter can be changed dynamically.  Most clients should//leave this parameter alone.////Default: 16int block_restart_interval; //block里面的restart间隔.具体可以看BlockBuilder.//Compress blocks using the specified compression algorithm.  This//parameter can be changed dynamically.////Default: kSnappyCompression, which gives lightweight but fast//compression.////Typical speeds of kSnappyCompression on an Intel(R) Core(TM)2 2.4GHz://~200-500MB/s compression//~400-800MB/s decompression//Note that these speeds are significantly faster than most//persistent storage speeds, and therefore it is typically never//worth switching to kNoCompression.  Even if the input data is//incompressible, the kSnappyCompression implementation will//efficiently detect that and will switch to uncompressed mode.CompressionType compression; //压缩类型.不压缩和Snappy压缩.//Create an Options object with default values for all fields.Options();
   };
   

2. ReadOptions

   和Options一样里面也都是控制字段.

   //Options that control read operations
   struct ReadOptions {//If true, all data read from underlying storage will be//verified against corresponding checksums.//Default: falsebool verify_checksums; //读取时候进行checksum校验//Should the data read for this iteration be cached in memory?//Callers may wish to set this field to false for bulk scans.//Default: truebool fill_cache; //是否需要从cache里面查找.如果是bulk scans的话那么设置false.//If "snapshot" is non-NULL, read as of the supplied snapshot//(which must belong to the DB that is being read and which must//not have been released).  If "snapshot" is NULL, use an impliicit//snapshot of the state at the beginning of this read operation.//Default: NULLconst Snapshot* snapshot; //如果存在Snapshot的话那么就在snapshot上面读取.ReadOptions(): verify_checksums(false),fill_cache(true),snapshot(NULL) {}
   };
   

3. WriteOptions

   和Options一样里面也都是控制字段.

   //Options that control write operations
   struct WriteOptions {//If true, the write will be flushed from the operating system//buffer cache (by calling WritableFile::Sync()) before the write//is considered complete.  If this flag is true, writes will be//slower.////If this flag is false, and the machine crashes, some recent//writes may be lost.  Note that if it is just the process that//crashes (i.e., the machine does not reboot), no writes will be//lost even if sync==false.////In other words, a DB write with sync==false has similar//crash semantics as the "write()" system call.  A DB write//with sync==true has similar crash semantics to a "write()"//system call followed by "fsync()".////Default: falsebool sync; //是否每次写都需要fsync.WriteOptions(): sync(false) {}
   };
   

10.1.8 slice.h

inline int Slice::compare(const Slice& b) const {const int min_len = (size_ < b.size_) ? size_ : b.size_;int r = memcmp(data_, b.data_, min_len);if (r == 0) {if (size_ < b.size_) r = -1;else if (size_ > b.size_) r = +1;}return r;
}

10.1.9 status.h

class Status {public://Create a success status.Status() : state_(NULL) { }~Status() { delete[] state_; }//Copy the specified status.Status(const Status& s);void operator=(const Status& s);//Return a success status.static Status OK() { return Status(); }//Return error status of an appropriate type.static Status NotFound(const Slice& msg, const Slice& msg2 = Slice()) {return Status(kNotFound, msg, msg2);}static Status Corruption(const Slice& msg, const Slice& msg2 = Slice()) {return Status(kCorruption, msg, msg2);}static Status NotSupported(const Slice& msg, const Slice& msg2 = Slice()) {return Status(kNotSupported, msg, msg2);}static Status InvalidArgument(const Slice& msg, const Slice& msg2 = Slice()) {return Status(kInvalidArgument, msg, msg2);}static Status IOError(const Slice& msg, const Slice& msg2 = Slice()) {return Status(kIOError, msg, msg2);}//Returns true iff the status indicates success.bool ok() const { return (state_ == NULL); }//Returns true iff the status indicates a NotFound error.bool IsNotFound() const { return code() == kNotFound; }//Return a string representation of this status suitable for printing.//Returns the string "OK" for success.std::string ToString() const;private://OK status has a NULL state_.  Otherwise, state_ is a new[] array//of the following form://state_[0..3] == length of message//state_[4]    == code//state_[5..]  == messageconst char* state_;enum Code {kOk = 0,kNotFound = 1,kCorruption = 2,kNotSupported = 3,kInvalidArgument = 4,kIOError = 5};Code code() const {return (state_ == NULL) ? kOk : static_cast<Code>(state_[4]);}Status(Code code, const Slice& msg, const Slice& msg2);static const char* CopyState(const char* s);
};inline Status::Status(const Status& s) {state_ = (s.state_ == NULL) ? NULL : CopyState(s.state_);
}
inline void Status::operator=(const Status& s) {//The following condition catches both aliasing (when this == &s),//and the common case where both s and *this are ok.if (state_ != s.state_) {delete[] state_;state_ = (s.state_ == NULL) ? NULL : CopyState(s.state_);}
}

10.1.10 table_builder.h

class TableBuilder {public://Create a builder that will store the contents of the table it is//building in *file.  Does not close the file.  It is up to the//caller to close the file after calling Finish().TableBuilder(const Options& options, WritableFile* file); //传入options和可写文件(追加写模式)//REQUIRES: Either Finish() or Abandon() has been called.~TableBuilder();//Change the options used by this builder.  Note: only some of the//option fields can be changed after construction.  If a field is//not allowed to change dynamically and its value in the structure//passed to the constructor is different from its value in the//structure passed to this method, this method will return an error//without changing any fields.Status ChangeOptions(const Options& options); //一旦构建好之后只允许修改部分字段.//Add key,value to the table being constructed.//REQUIRES: key is after any previously added key according to comparator.//REQUIRES: Finish(), Abandon() have not been calledvoid Add(const Slice& key, const Slice& value); //添加kv//Advanced operation: flush any buffered key/value pairs to file.//Can be used to ensure that two adjacent entries never live in//the same data block.  Most clients should not need to use this method.//REQUIRES: Finish(), Abandon() have not been calledvoid Flush(); //将buffered的kv刷新到文件//Return non-ok iff some error has been detected.Status status() const;//Finish building the table.  Stops using the file passed to the//constructor after this function returns.//REQUIRES: Finish(), Abandon() have not been calledStatus Finish(); //构建完成.//Indicate that the contents of this builder should be abandoned.  Stops//using the file passed to the constructor after this function returns.//If the caller is not going to call Finish(), it must call Abandon()//before destroying this builder.//REQUIRES: Finish(), Abandon() have not been calledvoid Abandon(); //如果放弃构建的话//Number of calls to Add() so far.uint64_t NumEntries() const; //添加了多少次//Size of the file generated so far.  If invoked after a successful//Finish() call, returns the size of the final generated file.uint64_t FileSize() const; //当前已经写入多少文件了.private:bool ok() const { return status().ok(); }void WriteBlock(BlockBuilder* block, BlockHandle* handle);struct Rep;Rep* rep_;//No copying allowedTableBuilder(const TableBuilder&);void operator=(const TableBuilder&);
};

10.1.11 table.h

//A Table is a sorted map from strings to strings.  Tables are
//immutable and persistent.  A Table may be safely accessed from
//multiple threads without external synchronization.
class Table {public://Attempt to open the table that is stored in bytes [0..file_size)//of "file", and read the metadata entries necessary to allow//retrieving data from the table.////If successful, returns ok and sets "*table" to the newly opened//table.  The client should delete "*table" when no longer needed.//If there was an error while initializing the table, sets "*table"//to NULL and returns a non-ok status.  Does not take ownership of//"*source", but the client must ensure that "source" remains live//for the duration of the returned table's lifetime.////*file must remain live while this Table is in use.static Status Open(const Options& options,RandomAccessFile* file,uint64_t file_size,Table** table);~Table();//Returns a new iterator over the table contents.//The result of NewIterator() is initially invalid (caller must//call one of the Seek methods on the iterator before using it).Iterator* NewIterator(const ReadOptions&) const;//Given a key, return an approximate byte offset in the file where//the data for that key begins (or would begin if the key were//present in the file).  The returned value is in terms of file//bytes, and so includes effects like compression of the underlying data.//E.g., the approximate offset of the last key in the table will//be close to the file length.uint64_t ApproximateOffsetOf(const Slice& key) const;private:struct Rep;Rep* rep_;explicit Table(Rep* rep) { rep_ = rep; }static Iterator* BlockReader(void*, const ReadOptions&, const Slice&);//No copying allowedTable(const Table&);void operator=(const Table&);
};

10.1.12 write_batch.h

class WriteBatch {public:WriteBatch();~WriteBatch();//Store the mapping "key->value" in the database.void Put(const Slice& key, const Slice& value);//If the database contains a mapping for "key", erase it.  Else do nothing.void Delete(const Slice& key);//Clear all updates buffered in this batch.void Clear();//Support for iterating over the contents of a batch.class Handler {public:virtual ~Handler();virtual void Put(const Slice& key, const Slice& value) = 0;virtual void Delete(const Slice& key) = 0;};Status Iterate(Handler* handler) const;//阅读后面会发现，这个意思是遍历所有的WriteBatch里面的item//然后操作handler.handler可能是一个memtable.这样可以将里面所有的内容//全部存放到memtable里面去.private:friend class WriteBatchInternal; //底层实现是这个.std::string rep_;  //See comment in write_batch.cc for the format of rep_ // 将所有的操作请求做成二进制存放在rep_//Intentionally copyable
};

10.3.1 FileName

Status SetCurrentFile(Env* env, const std::string& dbname,uint64_t descriptor_number) {//Remove leading "dbname/" and add newline to manifest file namestd::string manifest = DescriptorFileName(dbname, descriptor_number);Slice contents = manifest;assert(contents.starts_with(dbname + "/"));contents.remove_prefix(dbname.size() + 1);std::string tmp = TempFileName(dbname, descriptor_number);Status s = WriteStringToFile(env, contents.ToString() + "\n", tmp);if (s.ok()) {s = env->RenameFile(tmp, CurrentFileName(dbname));}if (!s.ok()) {env->DeleteFile(tmp);}return s;
}

10.3.2 Config

//Grouping of constants.  We may want to make some of these
//parameters set via options.
namespace config {
//leveldb最大level多少.
static const int kNumLevels = 7;//Level-0 compaction is started when we hit this many files.
//level0文件超过多少个触发compaction.
static const int kL0_CompactionTrigger = 4;//下面两个可以在MakeRoomForWrite里面看到.
//Soft limit on number of level-0 files.  We slow down writes at this point.
//如果level0文件超过这么多的话那么可能会放缓memtable写为level0的速度.比如delay 1s啥的.
static const int kL0_SlowdownWritesTrigger = 8;//Maximum number of level-0 files.  We stop writes at this point.
//如果level0文件超过这么多的话那不会memtable写为level0.
static const int kL0_StopWritesTrigger = 12;//对于memtable进行compaction的话选择的最高level.
//这个可以在PickLevelForMemtableOutput里面可以看到.
//Maximum level to which a new compacted memtable is pushed if it
//does not create overlap.  We try to push to level 2 to avoid the
//relatively expensive level 0=>1 compactions and to avoid some
//expensive manifest file operations.  We do not push all the way to
//the largest level since that can generate a lot of wasted disk
//space if the same key space is being repeatedly overwritten.
static const int kMaxMemCompactLevel = 2;}  //namespace config

10.3.3 DBImpl

class DBImpl : public DB {private://Constant after constructionEnv* const env_; //环境.const InternalKeyComparator internal_comparator_; //内部比较器.const Options options_;  //options_.comparator == &internal_comparator_ // 选项.bool owns_info_log_; //是否自己分配的log.bool owns_cache_; //是否自己分配的cache.const std::string dbname_; //db名称.//table_cache_ provides its own synchronizationTableCache* table_cache_; //TableCache.//Lock over the persistent DB state.  Non-NULL iff successfully acquired.FileLock* db_lock_; //db FileLock. // 对于外部进程标记DB对象互斥锁.//State below is protected by mutex_port::Mutex mutex_; //整个DB对象互斥锁.port::AtomicPointer shutting_down_; //标记这个DB对象正在退出。后台线程或看到之后不应该进行任何操作。port::CondVar bg_cv_;          //Signalled when background work finishesMemTable* mem_; //正在操作的memtable.MemTable* imm_;                //Memtable being compacted.正在被compacted的memtableport::AtomicPointer has_imm_;  //So bg thread can detect non-NULL imm_WritableFile* logfile_;uint64_t logfile_number_;log::Writer* log_;LoggerId* logger_;            //NULL, or the id of the current logging threadport::CondVar logger_cv_;     //For threads waiting to log.和mutex_关联.SnapshotList snapshots_; //当前所有的snapshot//Set of table files to protect from deletion because they are//part of ongoing compactions.std::set<uint64_t> pending_outputs_; //正在输出或者是进行compaction的file.//保存这个信息的话这样在delete文件的放置不被删除.//Has a background compaction been scheduled or is running?bool bg_compaction_scheduled_; //后台线程是否在运行.还是可以退出.//Information for a manual compactionstruct ManualCompaction {int level;bool done;const InternalKey* begin;   //NULL means beginning of key rangeconst InternalKey* end;     //NULL means end of key rangeInternalKey tmp_storage;    //Used to keep track of compaction progress};ManualCompaction* manual_compaction_;VersionSet* versions_; //版本集合.//Have we encountered a background error in paranoid mode?Status bg_error_; //后台线程运行状态.//Per level compaction stats.  stats_[level] stores the stats for//compactions that produced data for the specified "level".struct CompactionStats { //压缩状态信息.int64_t micros;int64_t bytes_read;int64_t bytes_written;CompactionStats() : micros(0), bytes_read(0), bytes_written(0) { }void Add(const CompactionStats& c) {this->micros += c.micros;this->bytes_read += c.bytes_read;this->bytes_written += c.bytes_written;}};CompactionStats stats_[config::kNumLevels]; //各个级别的压缩状态信息.
};

1. Open

   在Interface部分的db.h里面可以知道构造这个DB对象是通过DB::Open来构造的。

   Status DB::Open(const Options& options, const std::string& dbname,DB** dbptr) {*dbptr = NULL;DBImpl* impl = new DBImpl(options, dbname); //创建DBImpl实例.impl->mutex_.Lock(); //似乎没有太大必要.这里应该没有竞争.VersionEdit edit;//回复自上次依赖的edit所有内容.然后在后面直接log and apply这个edit对象.Status s = impl->Recover(&edit); //Handles create_if_missing, error_if_exists//下面每个步骤大致上都很清楚//1.创建新的log文件//2.回放原来log信息//3.删除不必要的文件//4.进行compaction.if (s.ok()) {uint64_t new_log_number = impl->versions_->NewFileNumber();WritableFile* lfile;s = options.env->NewWritableFile(LogFileName(dbname, new_log_number),&lfile);if (s.ok()) {edit.SetLogNumber(new_log_number);impl->logfile_ = lfile;impl->logfile_number_ = new_log_number;impl->log_ = new log::Writer(lfile);s = impl->versions_->LogAndApply(&edit, &impl->mutex_);}if (s.ok()) {impl->DeleteObsoleteFiles();impl->MaybeScheduleCompaction();}}//将产生的DB对象返回.impl->mutex_.Unlock();if (s.ok()) {*dbptr = impl;} else {delete impl;}return s;
   }
   

2. DBImpl

   先看看构造函数

   DBImpl::DBImpl(const Options& options, const std::string& dbname): env_(options.env),internal_comparator_(options.comparator),//注意这里的顺序.必须先调用SanitizeOptions.options_(SanitizeOptions(dbname, &internal_comparator_, options)),//判断是否自己持有info log和block cache.owns_info_log_(options_.info_log != options.info_log),owns_cache_(options_.block_cache != options.block_cache),dbname_(dbname),db_lock_(NULL),shutting_down_(NULL),bg_cv_(&mutex_),//创建新的MemTable对象.mem_(new MemTable(internal_comparator_)),imm_(NULL),//这里log都是指db的log而不是程序log.logfile_(NULL),logfile_number_(0),log_(NULL),logger_(NULL),logger_cv_(&mutex_),bg_compaction_scheduled_(false),manual_compaction_(NULL) {mem_->Ref();has_imm_.Release_Store(NULL);//Reserve ten files or so for other uses and give the rest to TableCache.const int table_cache_size = options.max_open_files - 10;//对于文件描述符限制可以通过TableCache来完成.不错:).table_cache_ = new TableCache(dbname_, &options_, table_cache_size);//创建VersionSet.versions_ = new VersionSet(dbname_, &options_, table_cache_,&internal_comparator_);
   }
   

   然后看看析构函数

   DBImpl::~DBImpl() {//Wait for background work to finishmutex_.Lock();//设置shuting down标记.后台线程等待标记退出.//等待bg_compaction_scheduled_置位，这样bg线程就退出了。shutting_down_.Release_Store(this);  //Any non-NULL value is okwhile (bg_compaction_scheduled_) {bg_cv_.Wait();}mutex_.Unlock();//释放db锁文件.if (db_lock_ != NULL) {env_->UnlockFile(db_lock_);}//删除VersionSet.delete versions_;//释放引用计数if (mem_ != NULL) mem_->Unref();if (imm_ != NULL) imm_->Unref();//删除可能产生的log,logfile以及table_cache.delete log_;delete logfile_;delete table_cache_;//释放程序log以及cache.if (owns_info_log_) {delete options_.info_log;}if (owns_cache_) {delete options_.block_cache;}
   }
   

3. NewDB

   但从名字上我们不知道里面做了哪些事情，所以还是来看看代码.通过代码阅读会发现，完成的事情大致就是建立一个Manifest文件，然后将这个版本的Manifest文件的文件名作为内容写入CURRENT文件。Manifest文件里面包含的就是VersionEdit信息。VersionEdit可以认为就是这个数据库的元信息。

   Status DBImpl::NewDB() {VersionEdit new_db;//创建新的VersionEdit.设置好comparator的名字以及必要字段//关于VersionEdit的信息会在后面仔细分析.new_db.SetComparatorName(user_comparator()->Name());new_db.SetLogNumber(0); //从log number从0开始分配.new_db.SetNextFile(2);  //下一个number从2开始分配new_db.SetLastSequence(0);//创建新的Manifest文件.然后作为writable文件打开.//1分配给manifest文件.const std::string manifest = DescriptorFileName(dbname_, 1);WritableFile* file;Status s = env_->NewWritableFile(manifest, &file);if (!s.ok()) {return s;}//文件作为log格式打开.将VersionEdit序列化之后作为log写入.{log::Writer log(file);std::string record;new_db.EncodeTo(&record);s = log.AddRecord(record);if (s.ok()) {s = file->Close();}}delete file;//然后将CURRENT里面的文件设置为版本1的manifest文件.if (s.ok()) {//Make "CURRENT" file that points to the new manifest file.s = SetCurrentFile(env_, dbname_, 1);} else {env_->DeleteFile(manifest);}return s;
   }
   

4. Recover

   恢复状态并且将恢复对于version日志操作.

   Status DBImpl::Recover(VersionEdit* edit) {mutex_.AssertHeld();//Ignore error from CreateDir since the creation of the DB is//committed only when the descriptor is created, and this directory//may already exist from a previous failed creation attempt.env_->CreateDir(dbname_);assert(db_lock_ == NULL);Status s = env_->LockFile(LockFileName(dbname_), &db_lock_);if (!s.ok()) {return s;}if (!env_->FileExists(CurrentFileName(dbname_))) {if (options_.create_if_missing) {s = NewDB(); //创建DB.if (!s.ok()) {return s;}} else {return Status::InvalidArgument(dbname_, "does not exist (create_if_missing is false)");}} else {if (options_.error_if_exists) {return Status::InvalidArgument(dbname_, "exists (error_if_exists is true)");}}s = versions_->Recover(); //对于VersionSet首先进行恢复.//恢复之后,根据里面的内容然后读取里面对应的version edit内容.//不过从versionset的Recover方法来看的话里面所有的内容已经全部恢复了.//后面edit的内容是因为在恢复log里面的内容造成的.然后将这个部分造成的edit//之后调用LogAndApply.if (s.ok()) {SequenceNumber max_sequence(0);//Recover from all newer log files than the ones named in the//descriptor (new log files may have been added by the previous//incarnation without registering them in the descriptor).////Note that PrevLogNumber() is no longer used, but we pay//attention to it in case we are recovering a database//produced by an older version of leveldb.//从注释这里看以看出prev_log已经不适用了.const uint64_t min_log = versions_->LogNumber();const uint64_t prev_log = versions_->PrevLogNumber();std::vector<std::string> filenames;s = env_->GetChildren(dbname_, &filenames);if (!s.ok()) {return s;}uint64_t number;FileType type;std::vector<uint64_t> logs;//分析logs文件然后判断哪些log文件是后来追加我们需要重放的.for (size_t i = 0; i < filenames.size(); i++) {if (ParseFileName(filenames[i], &number, &type)&& type == kLogFile&& ((number >= min_log) || (number == prev_log))) {logs.push_back(number);}}//Recover in the order in which the logs were generated//按照顺序进行重放.std::sort(logs.begin(), logs.end());for (size_t i = 0; i < logs.size(); i++) {//恢复某个log文件.并且将里面的操作修改填写到edit//同时知道当前最大的sequence是多少.s = RecoverLogFile(logs[i], edit, &max_sequence);//The previous incarnation may not have written any MANIFEST//records after allocating this log number.  So we manually//update the file number allocation counter in VersionSet.//标记file number已经被使用了.versions_->MarkFileNumberUsed(logs[i]);}if (s.ok()) {//更新一下sequence number.if (versions_->LastSequence() < max_sequence) {versions_->SetLastSequence(max_sequence);}}}return s;
   }
   

5. RecoverLogFile

   从单个log恢复写入的内容.并且根据log里面内容了解到对于version修改以及max_sequence.

   Status DBImpl::RecoverLogFile(uint64_t log_number,VersionEdit* edit,SequenceNumber* max_sequence) {struct LogReporter : public log::Reader::Reporter {Env* env;Logger* info_log;const char* fname;Status* status;  //NULL if options_.paranoid_checks==falsevirtual void Corruption(size_t bytes, const Status& s) {Log(info_log, "%s%s: dropping %d bytes; %s",(this->status == NULL ? "(ignoring error) " : ""),fname, static_cast<int>(bytes), s.ToString().c_str());if (this->status != NULL && this->status->ok()) *this->status = s;}};mutex_.AssertHeld();//打开日志文件//Open the log filestd::string fname = LogFileName(dbname_, log_number);SequentialFile* file;Status status = env_->NewSequentialFile(fname, &file);if (!status.ok()) {MaybeIgnoreError(&status);return status;}//构造reporter对象.//Create the log reader.LogReporter reporter;reporter.env = env_;reporter.info_log = options_.info_log;reporter.fname = fname.c_str();reporter.status = (options_.paranoid_checks ? &status : NULL);//We intentially make log::Reader do checksumming even if//paranoid_checks==false so that corruptions cause entire commits//to be skipped instead of propagating bad information (like overly//large sequence numbers).//读取日志文件.做校验并且从0偏移开始读取.log::Reader reader(file, &reporter, true/*checksum*/,0/*initial_offset*/);Log(options_.info_log, "Recovering log #%llu",(unsigned long long) log_number);//Read all the records and add to a memtablestd::string scratch;Slice record;WriteBatch batch;MemTable* mem = NULL;//不断地读取while (reader.ReadRecord(&record, &scratch) &&status.ok()) {if (record.size() < 12) {reporter.Corruption(record.size(), Status::Corruption("log record too small"));continue;}//log里面都是write batch的内容//所以可以直接设置.WriteBatchInternal::SetContents(&batch, record);if (mem == NULL) {//创建mem table.mem = new MemTable(internal_comparator_);mem->Ref();}status = WriteBatchInternal::InsertInto(&batch, mem);//对于日志恢复的话我们也可以选择性地进行status判断检查.MaybeIgnoreError(&status);if (!status.ok()) {break;}//更新sequence.const SequenceNumber last_seq =WriteBatchInternal::Sequence(&batch) +WriteBatchInternal::Count(&batch) - 1;if (last_seq > *max_sequence) {*max_sequence = last_seq;}//如果占用内存大于这个大小的话那么就会写level0.if (mem->ApproximateMemoryUsage() > options_.write_buffer_size) {status = WriteLevel0Table(mem, edit, NULL);if (!status.ok()) {//Reflect errors immediately so that conditions like full//file-systems cause the DB::Open() to fail.break;}mem->Unref();mem = NULL;}}//剩下的内存文件也会做table.if (status.ok() && mem != NULL) {status = WriteLevel0Table(mem, edit, NULL);//Reflect errors immediately so that conditions like full//file-systems cause the DB::Open() to fail.}if (mem != NULL) mem->Unref();delete file;return status;
   }
   

6. MaybeIgnoreError

   这个是在日志恢复部分是否进行错误恢复.

   void DBImpl::MaybeIgnoreError(Status* s) const {if (s->ok() || options_.paranoid_checks) {//No change needed} else {Log(options_.info_log, "Ignoring error %s", s->ToString().c_str());*s = Status::OK();}
   }
   

7. WriteLevel0Table

   将memtable写到level0.不过现在就实现来看的话不一定是写到level0.对于产生或者是删除的文件等对于version的操作都会反映到edit里面

   Status DBImpl::WriteLevel0Table(MemTable* mem, VersionEdit* edit,Version* base) {mutex_.AssertHeld();//会针对这个操作进行计时.const uint64_t start_micros = env_->NowMicros();FileMetaData meta;//产生新的file number.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);//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 != NULL) {//如果存在base version的话//那么会根据base version以及range来选择新的level进行序列化.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;//修改这个level的compaction数据.stats_[level].Add(stats);return s;
   }
   

8. Put

   Status DBImpl::Put(const WriteOptions& o, const Slice& key, const Slice& val) {return DB::Put(o, key, val);
   }
   Status DB::Put(const WriteOptions& opt, const Slice& key, const Slice& value) {WriteBatch batch;batch.Put(key, value); //将内容写到WriteBatch里面.然后通过return Write(opt, &batch); //Write写入到db内部.
   }
   

9. Delete

   Status DBImpl::Delete(const WriteOptions& options, const Slice& key) {return DB::Delete(options, key);
   }
   Status DB::Delete(const WriteOptions& opt, const Slice& key) {WriteBatch batch;batch.Delete(key); //将删除内容写到WriteBatch里面，然后通过return Write(opt, &batch); //Write写入到db内部.
   }
   

10. LoggingResponsibility

    对于Log信息的打印的话确保每次只有一个实例在操作.这里logger_cv和mutex_关联起来的.可以看到在调用AcquireLoggingResponsibility的地方之前都会加锁

    //There is at most one thread that is the current logger.  This call
    //waits until preceding logger(s) have finished and becomes the
    //current logger.
    void DBImpl::AcquireLoggingResponsibility(LoggerId* self) {while (logger_ != NULL) {logger_cv_.Wait();}logger_ = self;
    }
    

    而ReleaseLoggingResponsibility是释放logger的所有权.注意这里我们不会去主动操作解锁而是外部解锁。

    void DBImpl::ReleaseLoggingResponsibility(LoggerId* self) {assert(logger_ == self);logger_ = NULL;logger_cv_.SignalAll();
    }
    

11. MakeRoomForWrite

    仅仅从函数名称上面开不出到底是开辟什么空间.看看实现吧.代码里面可以分析出这个部分是存在compaction的.我们这里稍微总结一下逻辑

    • 如果允许delay的话并且level0文件数目超过slowdown的阈值的话，那么就会先尝试delay 1s.下次不会进行delay
    • 如果不是force的话并且memtable空间允许的话那么直接返回
    • 剩下的逻辑就是force出一个memtable了.那么这个时候必须进行compaction to level0.
    • 先检查是否正在被memtable compaction.如果正在的话那么等待
    • 然后查看level0文件数目是否过多.如果过多的话那么也等待
    • 最后创建新的memtable以及logfile.将原来的memtable保存起来准备后台compaction
    • 发起compaction.并且force=false.

    按照性能角度出发的话，这种逻辑应该非常make sense.

    //REQUIRES: mutex_ is held
    //REQUIRES: this thread is the current logger
    Status DBImpl::MakeRoomForWrite(bool force) {mutex_.AssertHeld();assert(logger_ != NULL);bool allow_delay = !force; //如果设置force的话//如果空间不够的话那么就会发起compaction操作//然后等待compaction操作完成看空间是否足够.Status s;while (true) {//如果后台存在错误的话那么直接返回.if (!bg_error_.ok()) {//yield previous errors = bg_error_;break;} else if (allow_delay &&versions_->NumLevelFiles(0) >= config::kL0_SlowdownWritesTrigger) {//We are getting close to hitting a hard limit on the number of//L0 files.  Rather than delaying a single write by several//seconds when we hit the hard limit, start delaying each//individual write by 1ms to reduce latency variance.  Also,//this delay hands over some CPU to the compaction thread in//case it is sharing the same core as the writer.mutex_.Unlock();env_->SleepForMicroseconds(1000); //延迟1s来看level0文件个数.allow_delay = false;  //Do not delay a single write more than oncemutex_.Lock();} else if (!force &&(mem_->ApproximateMemoryUsage() <= options_.write_buffer_size)) {//如果memtable允许写入的话那么没有任何问题.//There is room in current memtablebreak;} else if (imm_ != NULL) {//We have filled up the current memtable, but the previous//one is still being compacted, so we wait.//imm_应该是表示正在进行compact的memtable.//这里我担心condition是边缘触发而不是水平触发的话那么signal就会丢失.bg_cv_.Wait();} else if (versions_->NumLevelFiles(0) >= config::kL0_StopWritesTrigger) {//There are too many level-0 files.//level0文件个数过多的话那么等待compaction的完成.//note:这个部分非常重要，这样可以限制写入速度保证系统balance.Log(options_.info_log, "waiting...\n");bg_cv_.Wait();} else {//Attempt to switch to a new memtable and trigger compaction of old//试图创建一个新的memtable用来写.创建memtable的话同时也创建一个log文件.assert(versions_->PrevLogNumber() == 0);uint64_t new_log_number = versions_->NewFileNumber();WritableFile* lfile = NULL;s = env_->NewWritableFile(LogFileName(dbname_, new_log_number), &lfile);if (!s.ok()) {break;}//销毁原来的log文件b并且创建新的memtable.//新memtable存放在imm里面.后台应该是将imm进行compaction或者是level disk files之间进行compaction.delete log_;delete logfile_;logfile_ = lfile;logfile_number_ = new_log_number;log_ = new log::Writer(lfile);imm_ = mem_; //将原来的mem_存放起来.has_imm_.Release_Store(imm_); //imm存在标记.mem_ = new MemTable(internal_comparator_); //新创建memtable.mem_->Ref();force = false;   //Do not force another compaction if have roomMaybeScheduleCompaction(); //尝试进行compaction.这个函数我们后面分析.}}return s;
    }
    

12. MaybeScheduleCompaction

    发起compaction调度.

    void DBImpl::MaybeScheduleCompaction() {mutex_.AssertHeld();if (bg_compaction_scheduled_) { //如果已经被调度的话.那么没有必要产能重新发起.//Already scheduled} else if (shutting_down_.Acquire_Load()) { //如果已经关闭的话那么忽略.//DB is being deleted; no more background compactions} else if (imm_ == NULL && //如果memtable没有manual_compaction_ == NULL && //如果没有人工compaction!versions_->NeedsCompaction()) { //如果leveldb本身也没有必要认为compaction.//No work to be done} else {//compaction之前标记bg_compaction_scheduled_ = true;//并且这个时候发起compaction操作.这个后面会仔细分析BGWork这个过程.//BGWork是pthread_create封装的接口，里面仅仅是调用了BackgroundCall这个函数.env_->Schedule(&DBImpl::BGWork, this);}
    }
    

13. BackgroundCall

    void DBImpl::BackgroundCall() {MutexLock l(&mutex_);assert(bg_compaction_scheduled_);if (!shutting_down_.Acquire_Load()) { //如果没有关闭的话.BackgroundCompaction(); //那么会调用这个函数发起compaction.}bg_compaction_scheduled_ = false;//Previous compaction may have produced too many files in a level,//so reschedule another compaction if needed.MaybeScheduleCompaction(); //这里可能会重新发起compaction.bg_cv_.SignalAll();
    }
    

14. BackgroundCompaction

    这个部分是真正进行compaction的部分.我们仔细分析其中的行为.

    void DBImpl::BackgroundCompaction() {mutex_.AssertHeld();//从代码实现上来看的话，如果真实地进行compaction的话//对于version造成的修改都会记录为edit,然后调用VersionSet::LogAndApply保存起来.if (imm_ != NULL) {CompactMemTable(); //如果imm!=NULL的话那么首先对imm进行compaction.return;}//注意这里imm_==NULL.Compaction* c;bool is_manual = (manual_compaction_ != NULL);InternalKey manual_end;if (is_manual) {//如果是manual compaction的话//那么compaction里面需要提供level以及range.//但是我猜想仅仅是将这个compaction提供一个包装信息出来//具体操作延迟到后面进行.ManualCompaction* m = manual_compaction_;c = versions_->CompactRange(m->level, m->begin, m->end);m->done = (c == NULL);if (c != NULL) {manual_end = c->input(0, c->num_input_files(0) - 1)->largest;}Log(options_.info_log,"Manual compaction at level-%d from %s .. %s; will stop at %s\n",m->level,(m->begin ? m->begin->DebugString().c_str() : "(begin)"),(m->end ? m->end->DebugString().c_str() : "(end)"),(m->done ? "(end)" : manual_end.DebugString().c_str()));} else { //如果不是manual compaction那么挑选一个出来.c = versions_->PickCompaction();}Status status;if (c == NULL) {//Nothing to do} else if (!is_manual && c->IsTrivialMove()) {//Move file to next levelassert(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_);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));} else { //将compaction信息包装称为state进行操作.CompactionState* compact = new CompactionState(c);status = DoCompactionWork(compact);CleanupCompaction(compact);}delete c;if (status.ok()) {//Done} else if (shutting_down_.Acquire_Load()) {//Ignore compaction errors found during shutting down} else {Log(options_.info_log,"Compaction error: %s", status.ToString().c_str());if (options_.paranoid_checks && bg_error_.ok()) {bg_error_ = status;}}if (is_manual) { //如果是manual_compaction的话那么我们//有方法可以知道这次manual compaction实际操作范围有多少.//实际范围就是range所对应文件的largest key.:).这个之前可以看到.ManualCompaction* m = manual_compaction_;if (!m->done) {//We only compacted part of the requested range.  Update *m//to the range that is left to be compacted.m->tmp_storage = manual_end;m->begin = &m->tmp_storage;}manual_compaction_ = NULL;}
    }
    

15. CompactMemTable

    对memtable进行compaction.注意这里针对的对象是imm_.mem对象是write操作的对象.

    Status DBImpl::CompactMemTable() {mutex_.AssertHeld();assert(imm_ != NULL);//Save the contents of the memtable as a new TableVersionEdit 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");}//Replace immutable memtable with the generated Tableif (s.ok()) {//一旦这个memtable进行compaction完成之后//那么就可以认为这个log file number不需要了.//将这个信息记录在version edit里面.edit.SetPrevLogNumber(0);edit.SetLogNumber(logfile_number_);  //Earlier logs no longer neededs = versions_->LogAndApply(&edit, &mutex_);}//然后释放imm这个内存并且删除多余文件.if (s.ok()) {//Commit to the new stateimm_->Unref();imm_ = NULL;has_imm_.Release_Store(NULL);DeleteObsoleteFiles();}return s;
    }
    

16. CompactRange

    这个语义应该是针对某个range来进行compaction的.

    • 首先查看和这些range存在overlap的最大level是多少
    • 首先对memtable进行compaction(不管是否存在overlap)(TEST_CompactMemTable)
    • 然后遍历这些level,分别对每层进行compact range.(TEST_CompactRange)

    void DBImpl::CompactRange(const Slice* begin, const Slice* end) {int max_level_with_files = 1;{MutexLock l(&mutex_);Version* base = versions_->current();for (int level = 1; level < config::kNumLevels; level++) {if (base->OverlapInLevel(level, begin, end)) {max_level_with_files = level;}}}TEST_CompactMemTable(); //TODO(sanjay): Skip if memtable does not overlapfor (int level = 0; level < max_level_with_files; level++) {TEST_CompactRange(level, begin, end);}
    }
    

17. TEST_CompactMemTable

    Status DBImpl::TEST_CompactMemTable() {MutexLock l(&mutex_);LoggerId self;AcquireLoggingResponsibility(&self);//这里应该是要求imm==NULL也就是说没有memtable在进行compaction的.//那么这样的行为非常明显就是希望mem重新开辟//将原来的mem进行compactionStatus s = MakeRoomForWrite(true /*force compaction */);ReleaseLoggingResponsibility(&self);if (s.ok()) {//Wait until the compaction completes//并且等待memtable compaction结束.while (imm_ != NULL && bg_error_.ok()) {bg_cv_.Wait();}if (imm_ != NULL) {s = bg_error_;}}return s;
    }
    

18. TEST_CompactRange

    针对某个Level来进行range compaction.

    void DBImpl::TEST_CompactRange(int level, const Slice* begin,const Slice* end) {assert(level >= 0);assert(level + 1 < config::kNumLevels);InternalKey begin_storage, end_storage;ManualCompaction manual;manual.level = level; //levelmanual.done = false; //是否已经完成.if (begin == NULL) { //选择manual compaction range.manual.begin = NULL;} else {begin_storage = InternalKey(*begin, kMaxSequenceNumber, kValueTypeForSeek);manual.begin = &begin_storage;}if (end == NULL) {manual.end = NULL;} else {end_storage = InternalKey(*end, 0, static_cast<ValueType>(0));manual.end = &end_storage;}MutexLock l(&mutex_);while (!manual.done) { //如果没有完成的话//等待上一次完成while (manual_compaction_ != NULL) {bg_cv_.Wait();}//等待本次完成//还是调用maybe schedule compaction//按照代码来看的话走的分支主要是关注VersionSet::CompactRange这个部分.manual_compaction_ = &manual;MaybeScheduleCompaction();while (manual_compaction_ == &manual) {bg_cv_.Wait();}}
    }
    

19. DoCompactionWork

    完成的工作是实质性地Compaction.通过读取提交的CompactionState来完成.过程比较长可以逐段逐段地阅读代码.

    Status DBImpl::DoCompactionWork(CompactionState* compact) {const uint64_t start_micros = env_->NowMicros();int64_t imm_micros = 0;  //Micros spent doing imm_ compactionsLog(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 == NULL);assert(compact->outfile == NULL);//首先根据当前所有提交的snapshot知道当前最小的sequence number是多少.if (snapshots_.empty()) {compact->smallest_snapshot = versions_->LastSequence();} else {compact->smallest_snapshot = snapshots_.oldest()->number_;}//Release mutex while we're actually doing the compaction workmutex_.Unlock();//针对这个compaction创建出iterator.我们在后面看看这个iterator是如何建立的.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;for (; input->Valid() && !shutting_down_.Acquire_Load(); ) {//Prioritize immutable compaction work//做compaction之前先将immutable table compaction.if (has_imm_.NoBarrier_Load() != NULL) {const uint64_t imm_start = env_->NowMicros();mutex_.Lock();if (imm_ != NULL) {CompactMemTable();bg_cv_.SignalAll();  //Wakeup MakeRoomForWrite() if necessary}mutex_.Unlock();imm_micros += (env_->NowMicros() - imm_start);}//判断在这个key是否应该独立产生一个文件.Slice key = input->key();if (compact->compaction->ShouldStopBefore(key) &&compact->builder != NULL) {status = FinishCompactionOutputFile(compact, input);if (!status.ok()) {break;}}//Handle key/value, add to state, etc.//drop表示这个key是否应该直接丢弃.//这个地方始终感觉有点问题.为什么计算last seuqnce number呢?.//不过大致思想是了解的.bool drop = false;if (!ParseInternalKey(key, &ikey)) {//Do not hide error keyscurrent_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 keycurrent_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 keydrop = true;    //(A)} else if (ikey.type == kTypeDeletion && //如果是删除并且这个key < snapshot.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;}
    #if 0Log(options_.info_log,"  Compact: %s, seq %d, type: %d %d, drop: %d, is_base: %d, ""%d smallest_snapshot: %d",ikey.user_key.ToString().c_str(),(int)ikey.sequence, ikey.type, kTypeValue, drop,compact->compaction->IsBaseLevelForKey(ikey.user_key),(int)last_sequence_for_key, (int)compact->smallest_snapshot);
    #endif//不管前面细节如何到这个步骤表明这个kv没有被drop掉.if (!drop) {//Open output file if necessaryif (compact->builder == NULL) { //根据compaction状态产生新输出文件.status = OpenCompactionOutputFile(compact);if (!status.ok()) {break;}}//==0的时候记录最小key.if (compact->builder->NumEntries() == 0) {compact->current_output()->smallest.DecodeFrom(key);}//之后每次更新最大keycompact->current_output()->largest.DecodeFrom(key);//记录这个kv.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");}//最后可能需要关闭compaction的文件.if (status.ok() && compact->builder != NULL) {status = FinishCompactionOutputFile(compact, input);}if (status.ok()) {status = input->status();}delete input; //删除这个iterator.input = NULL;CompactionStats stats;//compaction部分的时间还需要出去内存memtable compaction是时间.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);if (status.ok()) {status = InstallCompactionResults(compact);}VersionSet::LevelSummaryStorage tmp;Log(options_.info_log,"compacted to: %s", versions_->LevelSummary(&tmp));return status;
    }
    

20. CleanupCompaction

    完成compaction之后的工作.

    void DBImpl::CleanupCompaction(CompactionState* compact) {mutex_.AssertHeld();if (compact->builder != NULL) { //如果这个builder没有销毁的话那么认为//中途是shutdown了.那么我们这里直接abandon掉.//May happen if we get a shutdown call in the middle of compactioncompact->builder->Abandon();delete compact->builder;} else {assert(compact->outfile == NULL);}delete compact->outfile;//compaction过程中的话新输出的output文件应该都加入了pengding_outputs//防止这个文件部分被删除.完成compaction之后的话可以移除了//考虑这里外部应该有标记通知这些文件还是需要的.//所以针对这些文件可能已经创建了另外一个version对象存放在version_set内部.//但是啥时候释放version这个对象呢???.for (size_t i = 0; i < compact->outputs.size(); i++) {const CompactionState::Output& out = compact->outputs[i];pending_outputs_.erase(out.number);}delete compact;
    }
    

21. InstallCompactionResults

    这个函数调用过程是在DoCompactionWork里面.大致工作就是将这次compaction工作内容作为日志保存起来.

    Status DBImpl::InstallCompactionResults(CompactionState* compact) {mutex_.AssertHeld();Log(options_.info_log,  "Compacted %d@%d + %d@%d files => %lld bytes",compact->compaction->num_input_files(0),compact->compaction->level(),compact->compaction->num_input_files(1),compact->compaction->level() + 1,static_cast<long long>(compact->total_bytes));//Add compaction outputscompact->compaction->AddInputDeletions(compact->compaction->edit()); //将compaction这次操作删除和增加文件加入version_edit.const int level = compact->compaction->level();for (size_t i = 0; i < compact->outputs.size(); i++) {const CompactionState::Output& out = compact->outputs[i];//将本地操作放在version_edit里面.compact->compaction->edit()->AddFile(level + 1,out.number, out.file_size, out.smallest, out.largest);pending_outputs_.erase(out.number); //从pending output里面删除.}compact->outputs.clear();Status s = versions_->LogAndApply(compact->compaction->edit(), &mutex_); //通知version set本次修改内容.//并且将这次compaction的内容作为version写入log里面去.if (s.ok()) {compact->compaction->ReleaseInputs();DeleteObsoleteFiles();} else {//Discard any files we may have created during this failed compactionfor (size_t i = 0; i < compact->outputs.size(); i++) {env_->DeleteFile(TableFileName(dbname_, compact->outputs[i].number));}}return s;
    }
    

22. OpenCompactionOutputFile

    打开用于compaction输出的文件.倒不是非常麻烦.

    Status DBImpl::OpenCompactionOutputFile(CompactionState* compact) {assert(compact != NULL);assert(compact->builder == NULL);uint64_t file_number;{mutex_.Lock();file_number = versions_->NewFileNumber();pending_outputs_.insert(file_number); //放在pending output里面.CompactionState::Output out;out.number = file_number;out.smallest.Clear();out.largest.Clear();compact->outputs.push_back(out);mutex_.Unlock();}//Make the output file//创建输出文件以及对应的table builder.std::string fname = TableFileName(dbname_, file_number);Status s = env_->NewWritableFile(fname, &compact->outfile);if (s.ok()) {compact->builder = new TableBuilder(options_, compact->outfile);}return s;
    }
    

23. FinishCompactionOutputFile

    对于输出了文件之后我们需要finish的操作.倒不是非常麻烦.

    Status DBImpl::FinishCompactionOutputFile(CompactionState* compact,Iterator* input) {assert(compact != NULL);assert(compact->outfile != NULL);assert(compact->builder != NULL);const uint64_t output_number = compact->current_output()->number;assert(output_number != 0);//Check for iterator errorsStatus s = input->status();const uint64_t current_entries = compact->builder->NumEntries();if (s.ok()) {s = compact->builder->Finish(); //通知对应的builder对象finish.} else {compact->builder->Abandon();}const uint64_t current_bytes = compact->builder->FileSize();compact->current_output()->file_size = current_bytes;compact->total_bytes += current_bytes;delete compact->builder; //释放原来builder.compact->builder = NULL;//Finish and check for file errorsif (s.ok()) {s = compact->outfile->Sync();}if (s.ok()) {s = compact->outfile->Close();}delete compact->outfile; //释放原来的file对象.compact->outfile = NULL;//验证这个文件是否OK.if (s.ok() && current_entries > 0) {//Verify that the table is usableIterator* iter = table_cache_->NewIterator(ReadOptions(),output_number,current_bytes);s = iter->status();delete iter;if (s.ok()) {Log(options_.info_log,"Generated table #%llu: %lld keys, %lld bytes",(unsigned long long) output_number,(unsigned long long) current_entries,(unsigned long long) current_bytes);}}return s;
    }
    

24. Write

    这个是Put/Delete底层的最终实现.仔细看看吧

    Status DBImpl::Write(const WriteOptions& options, WriteBatch* updates) {Status status;MutexLock l(&mutex_);LoggerId self;AcquireLoggingResponsibility(&self);status = MakeRoomForWrite(false);  //May temporarily release lock and waituint64_t last_sequence = versions_->LastSequence(); //分配sequence.if (status.ok()) {WriteBatchInternal::SetSequence(updates, last_sequence + 1); //WriteBatch的sequence从+1开始.last_sequence += WriteBatchInternal::Count(updates); //更新上次sequence.//Add to log and apply to memtable.  We can release the lock during//this phase since the "logger_" flag protects against concurrent//loggers and concurrent writes into mem_.{assert(logger_ == &self);mutex_.Unlock();//这个部分是肯定需要写入log的.//注意这里释放锁是没有问题，因为这个地方logger可以作为锁存在.//这里解开mutex似乎可以使得Get进行.//我觉得第一遍看leveldb代码的时候可以撇开并发这个部分.status = log_->AddRecord(WriteBatchInternal::Contents(updates)); //将updates的内容作为contents得到写入log.if (status.ok() && options.sync) {status = logfile_->Sync();}if (status.ok()) {//然后将updates写入memtable.status = WriteBatchInternal::InsertInto(updates, mem_);}mutex_.Lock();assert(logger_ == &self);}//写入成功之后的话那么将这个sequence重新写回.versions_->SetLastSequence(last_sequence); //设置新的sequence.}ReleaseLoggingResponsibility(&self);return status;
    }
    

25. SanitizeOptions

    这个函数应该就是规范一下options这个结构.

    //Fix user-supplied options to be reasonable
    template <class T,class V>
    static void ClipToRange(T* ptr, V minvalue, V maxvalue) {if (static_cast<V>(*ptr) > maxvalue) *ptr = maxvalue;if (static_cast<V>(*ptr) < minvalue) *ptr = minvalue;
    }
    Options SanitizeOptions(const std::string& dbname,const InternalKeyComparator* icmp,const Options& src) {Options result = src;result.comparator = icmp;//规范取值范围.ClipToRange(&result.max_open_files,           20,     50000); //20 , 50000ClipToRange(&result.write_buffer_size,        64<<10, 1<<30); //64K , 1GClipToRange(&result.block_size,               1<<10,  4<<20); //1K, 4M.//如果用户没有填写logger对象的话//那么产生一个logger对象.if (result.info_log == NULL) {//Open a log file in the same directory as the dbsrc.env->CreateDir(dbname);  //In case it does not existsrc.env->RenameFile(InfoLogFileName(dbname), OldInfoLogFileName(dbname));Status s = src.env->NewLogger(InfoLogFileName(dbname), &result.info_log);if (!s.ok()) {//No place suitable for loggingresult.info_log = NULL;}}//对于block cache也是.开辟的block cache大小8MB.if (result.block_cache == NULL) {result.block_cache = NewLRUCache(8 << 20);}return result;
    }
    

26. GetProperty

    关于db的属性信息.

    bool DBImpl::GetProperty(const Slice& property, std::string* value) {value->clear();MutexLock l(&mutex_);Slice in = property;//属性的key必须是以leveldb.开头的.Slice prefix("leveldb.");if (!in.starts_with(prefix)) return false;in.remove_prefix(prefix.size());if (in.starts_with("num-files-at-level")) { //每个level的文件数目.in.remove_prefix(strlen("num-files-at-level"));uint64_t level;bool ok = ConsumeDecimalNumber(&in, &level) && in.empty();if (!ok || level >= config::kNumLevels) {return false;} else {char buf[100];snprintf(buf, sizeof(buf), "%d",versions_->NumLevelFiles(static_cast<int>(level)));*value = buf;return true;}} else if (in == "stats") { //整个db的状态.char buf[200];snprintf(buf, sizeof(buf),"                               Compactions\n""Level  Files Size(MB) Time(sec) Read(MB) Write(MB)\n""--------------------------------------------------\n");value->append(buf);for (int level = 0; level < config::kNumLevels; level++) {int files = versions_->NumLevelFiles(level);if (stats_[level].micros > 0 || files > 0) {snprintf(buf, sizeof(buf),"%3d %8d %8.0f %9.0f %8.0f %9.0f\n",level,files,versions_->NumLevelBytes(level) / 1048576.0,stats_[level].micros / 1e6, //在这个level上面操作的时间.stats_[level].bytes_read / 1048576.0, //在这个level上面读取的字节数目.stats_[level].bytes_written / 1048576.0); //在这个level上面写入的字节数目.value->append(buf);}}return true;} else if (in == "sstables") { //当前sstables的状态.//这个是Version打印出的字符串.*value = versions_->current()->DebugString();return true;}return false;
    }
    

27. GetApproximateSizes

    得到某个Range占用的大小.底层依赖于VersionSet

    void DBImpl::GetApproximateSizes(const Range* range, int n,uint64_t* sizes) {//TODO(opt): better implementationVersion* v;{MutexLock l(&mutex_); //注意这里是全局加锁的.versions_->current()->Ref(); //对当前version加上引用计数.v = versions_->current();}for (int i = 0; i < n; i++) {//Convert user_key into a corresponding internal key.//这里sequence number是否正确呢?//不过大致上占用的空间差不多就是这么大.InternalKey k1(range[i].start, kMaxSequenceNumber, kValueTypeForSeek);InternalKey k2(range[i].limit, kMaxSequenceNumber, kValueTypeForSeek);//具体实现可以查看VersionSet方法.根据某个key判断这个key在文件中的偏移.//可能会有一部分偏差吧但是大致正确.uint64_t start = versions_->ApproximateOffsetOf(v, k1);uint64_t limit = versions_->ApproximateOffsetOf(v, k2);sizes[i] = (limit >= start ? limit - start : 0);}{MutexLock l(&mutex_);v->Unref();}
    }
    

28. GetSnapshot

    const Snapshot* DBImpl::GetSnapshot() {MutexLock l(&mutex_);//返回最新的sequence number创建一个Snapshot实例.//并且添加到snapshot list里面.return snapshots_.New(versions_->LastSequence());
    }
    

29. ReleaseSnapshot

    void DBImpl::ReleaseSnapshot(const Snapshot* s) {MutexLock l(&mutex_);//从snapshot list里面删除.snapshots_.Delete(reinterpret_cast<const SnapshotImpl*>(s));
    }
    

30. DeleteObsoleteFiles

    根据当前所有version所持有的file来决定哪些文件是不再需要的.version里面会持有file meta信息.

    void DBImpl::DeleteObsoleteFiles() {//Make a set of all of the live filesstd::set<uint64_t> live = pending_outputs_;versions_->AddLiveFiles(&live); //持有pending outputs里面正在输出的文件//并且将verisons里面所有version持有的文件得到.std::vector<std::string> filenames;//遍历当前目录下面的文件.//注意这里文件数目不会过多.//因为通常来说每个level文件都会有一个下限大小数目//而每个level的bytes有一个限制.对于最高层的level的话那么文件已经非常大了:)env_->GetChildren(dbname_, &filenames); //Ignoring errors on purposeuint64_t number;FileType type;for (size_t i = 0; i < filenames.size(); i++) {if (ParseFileName(filenames[i], &number, &type)) {bool keep = true;switch (type) {case kLogFile://对于log文件的number是按照顺序分配的.keep = ((number >= versions_->LogNumber()) || //(number == versions_->PrevLogNumber())); //prev log number已经没有必要了.break;case kDescriptorFile://Keep my manifest file, and any newer incarnations'//(in case there is a race that allows other incarnations)keep = (number >= versions_->ManifestFileNumber());break;case kTableFile:keep = (live.find(number) != live.end()); //判断这个文件是否需要删除.break;case kTempFile://Any temp files that are currently being written to must//be recorded in pending_outputs_, which is inserted into "live"keep = (live.find(number) != live.end()); //判断文件是否需要删除.break;case kCurrentFile:case kDBLockFile:case kInfoLogFile:keep = true; //对于其他文件的话直接keep住就好了.break;}if (!keep) {if (type == kTableFile) { //如果是table文件的话还需要从cache里面去掉.table_cache_->Evict(number);}Log(options_.info_log, "Delete type=%d #%lld\n",int(type),static_cast<unsigned long long>(number));env_->DeleteFile(dbname_ + "/" + filenames[i]); //然后删除文件.}}}
    }
    

31. Get

    Get的过程非常简单.因为实际Get的过程已经托管为Version::Get这个方法了.这里面我们只需要构造出正确的seuqnce number,和user key拼接成为internal key即可.

    Status DBImpl::Get(const ReadOptions& options,const Slice& key,std::string* value) {Status s;MutexLock l(&mutex_);SequenceNumber snapshot;if (options.snapshot != NULL) { //判断当前读取是否需要snapshot.snapshot = reinterpret_cast<const SnapshotImpl*>(options.snapshot)->number_;} else {snapshot = versions_->LastSequence();}MemTable* mem = mem_;MemTable* imm = imm_;Version* current = versions_->current();mem->Ref();if (imm != NULL) imm->Ref();current->Ref();//这个查询是否会造成compaction触发.bool have_stat_update = false;Version::GetStats stats;//Unlock while reading from files and memtables{mutex_.Unlock();//以这个key去进行查询.//1.首先查询memtable 2.然后查询正在dump的memtable 3.查询磁盘.//First look in the memtable, then in the immutable memtable (if any).LookupKey lkey(key, snapshot);if (mem->Get(lkey, value, &s)) {//Done} else if (imm != NULL && imm->Get(lkey, value, &s)) {//Done} else {s = current->Get(options, lkey, value, &stats);have_stat_update = true; //查询磁盘的话会进行标记.}mutex_.Lock();}//得到的内容会反馈到当前的version里面然后尝试触发compaction.if (have_stat_update && current->UpdateStats(stats)) {MaybeScheduleCompaction();}mem->Unref();if (imm != NULL) imm->Unref();current->Unref();return s;
    }
    

32. NewInternalIterator

    开辟内部的迭代器.但是功能上来说基本上以及完成了db::iterator所需要完成的工作.但是需要注意这里面merge iterator接口是针对internal key的，所以外部的话还是需要保证user key并且配合sequence number的行为.另外还需要考虑存在deletion情况.

    struct IterState {port::Mutex* mu;Version* version;MemTable* mem;MemTable* imm;
    };static void CleanupIteratorState(void* arg1, void* arg2) {IterState* state = reinterpret_cast<IterState*>(arg1);state->mu->Lock();state->mem->Unref();if (state->imm != NULL) state->imm->Unref();state->version->Unref();state->mu->Unlock();delete state;
    }Iterator* DBImpl::NewInternalIterator(const ReadOptions& options,SequenceNumber* latest_snapshot) {IterState* cleanup = new IterState;mutex_.Lock();*latest_snapshot = versions_->LastSequence();//将可能存在的iterator放在一个list里面做成一个merge list内容.//Collect together all needed child iteratorsstd::vector<Iterator*> list;//首先memtable下面需要遍历.list.push_back(mem_->NewIterator());mem_->Ref();//其次imm memtable需要遍历if (imm_ != NULL) {list.push_back(imm_->NewIterator());imm_->Ref();}//对于version来说可能存在很多文件需要遍历.versions_->current()->AddIterators(options, &list);//将这些内容构造称为一个merge iterator.//注意这里的内容都加了引用计数.Iterator* internal_iter =NewMergingIterator(&internal_comparator_, &list[0], list.size());versions_->current()->Ref();//然后将这些释放内容存放在internal iter销毁回调里面.cleanup->mu = &mutex_;cleanup->mem = mem_;cleanup->imm = imm_;cleanup->version = versions_->current();internal_iter->RegisterCleanup(CleanupIteratorState, cleanup, NULL);mutex_.Unlock();return internal_iter;
    }
    

33. NewInterator

    实现在DBIter里面.工厂方式进行创建.

    Iterator* DBImpl::NewIterator(const ReadOptions& options) {//了解当前最近的snapshot sequence number.SequenceNumber latest_snapshot;Iterator* internal_iter = NewInternalIterator(options, &latest_snapshot);//调用NewDBIterator来进行创建.具体可以查看DBIter实现.return NewDBIterator(&dbname_, env_, user_comparator(), internal_iter,(options.snapshot != NULL? reinterpret_cast<const SnapshotImpl*>(options.snapshot)->number_: latest_snapshot));
    }
    

10.3.4 DBIter

1. NewDBIterator

   Iterator* NewDBIterator(const std::string* dbname,Env* env,const Comparator* user_key_comparator,Iterator* internal_iter,const SequenceNumber& sequence) {return new DBIter(dbname, env, user_key_comparator, internal_iter, sequence);
   }
   

2. DBIter

   //Memtables and sstables that make the DB representation contain
   //(userkey,seq,type) => uservalue entries.  DBIter
   //combines multiple entries for the same userkey found in the DB
   //representation into a single entry while accounting for sequence
   //numbers, deletion markers, overwrites, etc.
   class DBIter: public Iterator {public://Which direction is the iterator currently moving?//(1) When moving forward, the internal iterator is positioned at//the exact entry that yields this->key(), this->value()//(2) When moving backwards, the internal iterator is positioned//just before all entries whose user key == this->key().enum Direction {kForward,kReverse};DBIter(const std::string* dbname, Env* env,const Comparator* cmp, Iterator* iter, SequenceNumber s): dbname_(dbname),env_(env),user_comparator_(cmp),iter_(iter),sequence_(s),direction_(kForward), //默认是向前查找.valid_(false) { //当前没有任何kv.}virtual ~DBIter() {delete iter_;}private:const std::string* const dbname_;Env* const env_;const Comparator* const user_comparator_;Iterator* const iter_;SequenceNumber const sequence_;Status status_;//如果是kReverse方向的话,那么从这里返回//否则直接调用iter返回.std::string saved_key_;     //== current key when direction_==kReversestd::string saved_value_;   //== current raw value when direction_==kReverseDirection direction_;bool valid_;
   };
   

3. ClearSavedValue

   清除saved_value这个字段.好像有点技巧.可以看看代码啥的测试一下.

   inline void ClearSavedValue() {if (saved_value_.capacity() > 1048576) { //>1M的话那么直接swap.std::string empty;swap(empty, saved_value_);} else { //否则使用clear.saved_value_.clear();}
   }
   

4. ParseKey

   工作非常简单就是从iter得到对应的key.

   inline bool DBIter::ParseKey(ParsedInternalKey* ikey) {if (!ParseInternalKey(iter_->key(), ikey)) {status_ = Status::Corruption("corrupted internal key in DBIter");return false;} else {return true;}
   }
   

5. SaveKey

   inline void SaveKey(const Slice& k, std::string* dst) {dst->assign(k.data(), k.size());
   }
   

6. Seek

   void DBIter::Seek(const Slice& target) {direction_ = kForward;ClearSavedValue(); //将saved_value_清空.saved_key_.clear(); //将saved_key_清空.AppendInternalKey( //重新设置查询key.保存到saved_key_.&saved_key_, ParsedInternalKey(target, sequence_, kValueTypeForSeek));iter_->Seek(saved_key_);if (iter_->Valid()) {FindNextUserEntry(false, &saved_key_ /*temporary storage */);} else {valid_ = false;}
   }
   

7. Next

   void DBIter::Next() {assert(valid_);if (direction_ == kReverse) {  //Switch directions?direction_ = kForward;//iter_ is pointing just before the entries for this->key(),//so advance into the range of entries for this->key() and then//use the normal skipping code below.if (!iter_->Valid()) {iter_->SeekToFirst();} else {iter_->Next();}if (!iter_->Valid()) {valid_ = false;saved_key_.clear();return;}}//Temporarily use saved_key_ as storage for key to skip.std::string* skip = &saved_key_;SaveKey(ExtractUserKey(iter_->key()), skip);FindNextUserEntry(true, skip);
   }
   

8. Prev

   void DBIter::Prev() {assert(valid_);if (direction_ == kForward) {  //Switch directions?//iter_ is pointing at the current entry.  Scan backwards until//the key changes so we can use the normal reverse scanning code.//首先向前一直找到略过当前saved_key的内容.assert(iter_->Valid());  //Otherwise valid_ would have been falseSaveKey(ExtractUserKey(iter_->key()), &saved_key_);while (true) {iter_->Prev();if (!iter_->Valid()) {valid_ = false;saved_key_.clear();ClearSavedValue();return;}if (user_comparator_->Compare(ExtractUserKey(iter_->key()),saved_key_) < 0) {break;}}direction_ = kReverse;}FindPrevUserEntry();
   }
   

9. FindNextUserEntry

   我们需要考虑到在DBIter里面处理的是mergeiterator.多路的话可能会存在很多.

   :现在才发现原来在find的时候并没有将sequence number放在里面而仅仅是比较user key的内容，然后将所有的user key内容放在一起然后来处理sequence.不过对于memtable等在插入的时候还是按照user key+sequence来进行存放.按照这个理解的话很多事情就比较好懂了.

   总感觉这个地方可能存在问题，没有考虑到delete k然后add k的情况.不过撇开这个问题就好理解多了，我们这里得到skip之后的话就可以在next的时候需要越过skip这个key.因为上次已经得到这个key的内容了.

   void DBIter::FindNextUserEntry(bool skipping, std::string* skip) {//Loop until we hit an acceptable entry to yieldassert(iter_->Valid());assert(direction_ == kForward);do {ParsedInternalKey ikey;if (ParseKey(&ikey) && ikey.sequence <= sequence_) { //首先需要满足sequence.switch (ikey.type) {case kTypeDeletion: //如果是deletion的话,那么保存这个skip的内容.//Arrange to skip all upcoming entries for this key since//they are hidden by this deletion.SaveKey(ikey.user_key, skip);skipping = true;break;case kTypeValue:if (skipping &&user_comparator_->Compare(ikey.user_key, *skip) <= 0) {//Entry hidden} else {valid_ = true;saved_key_.clear();return;}break;}}iter_->Next();} while (iter_->Valid());saved_key_.clear();valid_ = false;
   }
   

10. FindPrevUserEntry

    void DBIter::FindPrevUserEntry() {assert(direction_ == kReverse);ValueType value_type = kTypeDeletion;if (iter_->Valid()) {do {ParsedInternalKey ikey;if (ParseKey(&ikey) && ikey.sequence <= sequence_) {if ((value_type != kTypeDeletion) &&user_comparator_->Compare(ikey.user_key, saved_key_) < 0) {//We encountered a non-deleted value in entries for previous keys,break;}value_type = ikey.type;if (value_type == kTypeDeletion) {saved_key_.clear();ClearSavedValue();} else {Slice raw_value = iter_->value();if (saved_value_.capacity() > raw_value.size() + 1048576) {std::string empty;swap(empty, saved_value_);}SaveKey(ExtractUserKey(iter_->key()), &saved_key_);saved_value_.assign(raw_value.data(), raw_value.size());}}iter_->Prev();} while (iter_->Valid());}if (value_type == kTypeDeletion) {//Endvalid_ = false;saved_key_.clear();ClearSavedValue();direction_ = kForward;} else {valid_ = true;}
    }
    

11. SeekToFirst

    void DBIter::SeekToFirst() {direction_ = kForward;ClearSavedValue();iter_->SeekToFirst();if (iter_->Valid()) {FindNextUserEntry(false, &saved_key_ /*temporary storage */);} else {valid_ = false;}
    }
    

12. SeekToLast

    void DBIter::SeekToLast() {direction_ = kReverse;ClearSavedValue();iter_->SeekToLast();FindPrevUserEntry();
    }
    

10.3.5 LookupKey

//A helper class useful for DBImpl::Get()
class LookupKey {public://Initialize *this for looking up user_key at a snapshot with//the specified sequence number.LookupKey(const Slice& user_key, SequenceNumber sequence);~LookupKey();//Return a key suitable for lookup in a MemTable.//返回能够为memtable查询的key.Slice memtable_key() const { return Slice(start_, end_ - start_); }//Return an internal key (suitable for passing to an internal iterator)//返回internal key.这个应该是作为sstable查询的key.//后面我们看看InternalKey的结构.Slice internal_key() const { return Slice(kstart_, end_ - kstart_); }//返回user_key本身.//Return the user keySlice user_key() const { return Slice(kstart_, end_ - kstart_ - 8); }private://这里面给出了传入user_key之后构造出key的格式.//We construct a char array of the form://klength  varint32               <-- start_//userkey  char[klength]          <-- kstart_//tag      uint64//<-- end_//The array is a suitable MemTable key.//The suffix starting with "userkey" can be used as an InternalKey.const char* start_;const char* kstart_;const char* end_;char space_[200];      //Avoid allocation for short keys
};

inline LookupKey::~LookupKey() {//这里space_是为了对于short keys不进行分配.if (start_ != space_) delete[] start_;
}LookupKey::LookupKey(const Slice& user_key, SequenceNumber s) {size_t usize = user_key.size();//klength占用5个字节//tag占用8个字节.size_t needed = usize + 13;  //A conservative estimatechar* dst;if (needed <= sizeof(space_)) {dst = space_;} else {dst = new char[needed];}start_ = dst;dst = EncodeVarint32(dst, usize + 8);kstart_ = dst;memcpy(dst, user_key.data(), usize);dst += usize;EncodeFixed64(dst, PackSequenceAndType(s, kValueTypeForSeek));dst += 8;end_ = dst;
}

10.3.6 ValueType

//Value types encoded as the last component of internal keys.
//DO NOT CHANGE THESE ENUM VALUES: they are embedded in the on-disk
//data structures.
enum ValueType {kTypeDeletion = 0x0,kTypeValue = 0x1
};
//kValueTypeForSeek defines the ValueType that should be passed when
//constructing a ParsedInternalKey object for seeking to a particular
//sequence number (since we sort sequence numbers in decreasing order
//and the value type is embedded as the low 8 bits in the sequence
//number in internal keys, we need to use the highest-numbered
//ValueType, not the lowest).
static const ValueType kValueTypeForSeek = kTypeValue;

10.3.7 SequenceNumber

typedef uint64_t SequenceNumber;//We leave eight bits empty at the bottom so a type and sequence#
//can be packed together into 64-bits.
static const SequenceNumber kMaxSequenceNumber =((0x1ull << 56) - 1);

10.3.8 InternalKey

//Modules in this directory should keep internal keys wrapped inside
//the following class instead of plain strings so that we do not
//incorrectly use string comparisons instead of an InternalKeyComparator.
class InternalKey {private:std::string rep_;public:InternalKey() { }   //Leave rep_ as empty to indicate it is invalidInternalKey(const Slice& user_key, SequenceNumber s, ValueType t) {AppendInternalKey(&rep_, ParsedInternalKey(user_key, s, t));}void DecodeFrom(const Slice& s) { rep_.assign(s.data(), s.size()); }Slice Encode() const {assert(!rep_.empty());return rep_;}Slice user_key() const { return ExtractUserKey(rep_); }void SetFrom(const ParsedInternalKey& p) {rep_.clear();AppendInternalKey(&rep_, p);}void Clear() { rep_.clear(); }std::string DebugString() const;
};

1. ExtracrUserKey

   之前我们知道user key是如何分布的了.后面8个字节有附加信息.

   inline Slice ExtractUserKey(const Slice& internal_key) {assert(internal_key.size() >= 8);return Slice(internal_key.data(), internal_key.size() - 8);
   }
   

2. ExtractValueType

   inline ValueType ExtractValueType(const Slice& internal_key) {assert(internal_key.size() >= 8);const size_t n = internal_key.size();uint64_t num = DecodeFixed64(internal_key.data() + n - 8);unsigned char c = num & 0xff;return static_cast<ValueType>(c);
   }
   

3. ParsedInternalKey

   ParsedInternalKey是从InternalKey解析之后的表示.非常简单.

   struct ParsedInternalKey {Slice user_key;SequenceNumber sequence;ValueType type;ParsedInternalKey() { }  //Intentionally left uninitialized (for speed)ParsedInternalKey(const Slice& u, const SequenceNumber& seq, ValueType t): user_key(u), sequence(seq), type(t) { }std::string DebugString() const;
   };
   

4. InternalKeyEncodingLength

   如果ParsedInternalKey打包称为InternalKey的长度.

   //Return the length of the encoding of "key".
   inline size_t InternalKeyEncodingLength(const ParsedInternalKey& key) {return key.user_key.size() + 8;
   }
   

5. AppendInternalKey

   将ParsedInternalKey直接序列化到二进制格式.

   void AppendInternalKey(std::string* result, const ParsedInternalKey& key) {result->append(key.user_key.data(), key.user_key.size());PutFixed64(result, PackSequenceAndType(key.sequence, key.type));
   }
   

6. PackSequenceAndType

   sequence number占据高56bits,type占据低8bits

   static uint64_t PackSequenceAndType(uint64_t seq, ValueType t) {assert(seq <= kMaxSequenceNumber);assert(t <= kValueTypeForSeek);return (seq << 8) | t;
   }
   

7. ParseInternalKey

   根据InternalKey解析出ParsedInternalKey

   inline bool ParseInternalKey(const Slice& internal_key,ParsedInternalKey* result) {const size_t n = internal_key.size();if (n < 8) return false;uint64_t num = DecodeFixed64(internal_key.data() + n - 8);unsigned char c = num & 0xff;result->sequence = num >> 8;result->type = static_cast<ValueType>(c);result->user_key = Slice(internal_key.data(), n - 8);return (c <= static_cast<unsigned char>(kTypeValue));
   }
   

10.3.9 InternalKeyComparator

//A comparator for internal keys that uses a specified comparator for
//the user key portion and breaks ties by decreasing sequence number.
class InternalKeyComparator : public Comparator {private:const Comparator* user_comparator_;public:explicit InternalKeyComparator(const Comparator* c) : user_comparator_(c) { }virtual const char* Name() const; //"leveldb.InternalKeyComparator"virtual int Compare(const Slice& a, const Slice& b) const;virtual void FindShortestSeparator(std::string* start,const Slice& limit) const;virtual void FindShortSuccessor(std::string* key) const;const Comparator* user_comparator() const { return user_comparator_; }int Compare(const InternalKey& a, const InternalKey& b) const;
};

1. Compare

   int InternalKeyComparator::Compare(const Slice& akey, const Slice& bkey) const {//Order by://increasing user key (according to user-supplied comparator)//decreasing sequence number//decreasing type (though sequence# should be enough to disambiguate)int r = user_comparator_->Compare(ExtractUserKey(akey), ExtractUserKey(bkey));if (r == 0) {const uint64_t anum = DecodeFixed64(akey.data() + akey.size() - 8);const uint64_t bnum = DecodeFixed64(bkey.data() + bkey.size() - 8);if (anum > bnum) { //按照sequence number比较.//之前我们在MemTableInserter里面可以看到sequence number是不断增加的.r = -1;} else if (anum < bnum) {r = +1;}}return r;
   }
   

2. FindShortestSeparator

   void InternalKeyComparator::FindShortestSeparator(std::string* start,const Slice& limit) const {//Attempt to shorten the user portion of the keySlice user_start = ExtractUserKey(*start);Slice user_limit = ExtractUserKey(limit);std::string tmp(user_start.data(), user_start.size());//首先使用user_comparator找到更短的user_comparator_->FindShortestSeparator(&tmp, user_limit);if (tmp.size() < user_start.size() && //然后比较是否真的更短.user_comparator_->Compare(user_start, tmp) < 0) {//如果更短的话，那么会将后面的8个字节补齐//这里的8个字节使用kMaxSequenceNumber以及kValueTypeForSeek//其中kMaxSequenceNumber == (1 << 56)-1//而kValueTypeForSeek == KValueType.(就是普通的kv类型)//User key has become shorter physically, but larger logically.//Tack on the earliest possible number to the shortened user key.PutFixed64(&tmp, PackSequenceAndType(kMaxSequenceNumber,kValueTypeForSeek));assert(this->Compare(*start, tmp) < 0);assert(this->Compare(tmp, limit) < 0);start->swap(tmp);}
   }
   

3. FindShortSuccessor

   void InternalKeyComparator::FindShortSuccessor(std::string* key) const {Slice user_key = ExtractUserKey(*key);std::string tmp(user_key.data(), user_key.size());user_comparator_->FindShortSuccessor(&tmp); //首先通过user_comparator_找到下一个if (tmp.size() < user_key.size() &&user_comparator_->Compare(user_key, tmp) < 0) {//如果确实更短的话那么会加上特定的8字节附加信息.//关于附加信息在上面那个函数已经解释过了.//User key has become shorter physically, but larger logically.//Tack on the earliest possible number to the shortened user key.PutFixed64(&tmp, PackSequenceAndType(kMaxSequenceNumber,kValueTypeForSeek));assert(this->Compare(*key, tmp) < 0);key->swap(tmp);}
   }
   

10.3.10 FileMetaData

struct FileMetaData {int refs;//首先会设置一个初值int allowed_seeks;          //Seeks allowed until compactionuint64_t number; //file_number.比如1.sst.这个结合BuildTable和TableCache可以理解意思.uint64_t file_size;         //File size in bytesInternalKey smallest;       //Smallest internal key served by tableInternalKey largest;        //Largest internal key served by tableFileMetaData() : refs(0), allowed_seeks(1 << 30), file_size(0) { }
};

10.3.11 Version

class Version {public://private:public:friend class Compaction;friend class VersionSet;class LevelFileNumIterator;Iterator* NewConcatenatingIterator(const ReadOptions&, int level) const;VersionSet* vset_;            //VersionSet to which this Version belongsVersion* next_;               //Next version in linked listVersion* prev_;               //Previous version in linked listint refs_;                    //Number of live refs to this version//List of files per level//每个level的files_都是经过排序的。//对于level0可能存在overlap,对于level-x不存在overlap.std::vector<FileMetaData*> files_[config::kNumLevels]; //这个version在各个level保持的文件.//Next file to compact based on seek stats.//根据seek的统计下次需要进行compact的文件.FileMetaData* file_to_compact_;  //下次进行compact文件int file_to_compact_level_; //这个文件所在的level.//Level that should be compacted next and its compaction score.//Score < 1 means compaction is not strictly needed.  These fields//are initialized by Finalize().double compaction_score_;int compaction_level_;//构造函数非常简单.explicit Version(VersionSet* vset): vset_(vset), next_(this), prev_(this), refs_(0),file_to_compact_(NULL),file_to_compact_level_(-1),compaction_score_(-1),compaction_level_(-1) {}~Version();
};

1. AddIterators

   将所持有的所有的level文件打开并且返回iterator.从注释上来看的话得到这些iterators的话是为了进行merge.

   void Version::AddIterators(const ReadOptions& options,std::vector<Iterator*>* iters) {//Merge all level zero files together since they may overlap//打开所有的level0文件.for (size_t i = 0; i < files_[0].size(); i++) {iters->push_back(vset_->table_cache_->NewIterator(options, files_[0][i]->number, files_[0][i]->file_size));}//打开其他level的文件.关于这个concatenating后面会仔细分析.//For levels > 0, we can use a concatenating iterator that sequentially//walks through the non-overlapping files in the level, opening them//lazily.for (int level = 1; level < config::kNumLevels; level++) {if (!files_[level].empty()) {iters->push_back(NewConcatenatingIterator(options, level));}}
   }
   

2. NewConcatenatingIterator

   Iterator* Version::NewConcatenatingIterator(const ReadOptions& options,int level) const {return NewTwoLevelIterator(//注意这里的level>1.每个文件之间是没有overlap的.//并且这里我们也可以知道files_对于meta之间都是进行了排序的。new LevelFileNumIterator(vset_->icmp_, &files_[level]), //一级遍历器采用LevelFileNumberIterator.//映射到二级遍历器的话使用GetFileIterator来获得.&GetFileIterator, vset_->table_cache_, options);
   }
   

3. LevelFileNumIterator

   结构应该非常简单，我们大致看看即可。主要是关心一下二级映射函数GetFileIterator.对于key的话采用这个filemetadata里面的最大的key,而value采用filenumber+filesize表示.

   class Version::LevelFileNumIterator : public Iterator {public:LevelFileNumIterator(const InternalKeyComparator& icmp,const std::vector<FileMetaData*>* flist): icmp_(icmp),flist_(flist),index_(flist->size()) {        //Marks as invalid}virtual bool Valid() const {return index_ < flist_->size();}virtual void Seek(const Slice& target) {index_ = FindFile(icmp_, *flist_, target); //找到这个target所存在的最小的index.//这个后面会具体分析.之前看到了这层level都是disjoint的//所以在FindFile里面可以根据smallest也可以根据largest来进行查找.}virtual void SeekToFirst() { index_ = 0; }virtual void SeekToLast() {index_ = flist_->empty() ? 0 : flist_->size() - 1;}virtual void Next() {assert(Valid());index_++;}virtual void Prev() {assert(Valid());if (index_ == 0) {index_ = flist_->size();  //Marks as invalid} else {index_--;}}Slice key() const {assert(Valid());return (*flist_)[index_]->largest.Encode();}Slice value() const {assert(Valid()); //file_number + file_size.//关于如何使用这个value.应该是根据这个value对应到这个具体文件的iterator.EncodeFixed64(value_buf_, (*flist_)[index_]->number);EncodeFixed64(value_buf_+8, (*flist_)[index_]->file_size);return Slice(value_buf_, sizeof(value_buf_));}virtual Status status() const { return Status::OK(); }private:const InternalKeyComparator icmp_;const std::vector<FileMetaData*>* const flist_;uint32_t index_;//Backing store for value().  Holds the file number and size.mutable char value_buf_[16];
   };
   

4. GetFileIterator

   根据上面的分析file_value就是file_number + file_size.这里我们可以知道file_number应该是全局唯一的，而不是在level上面唯一的。关于TableCache后面会分析。

   static Iterator* GetFileIterator(void* arg,const ReadOptions& options,const Slice& file_value) {TableCache* cache = reinterpret_cast<TableCache*>(arg);if (file_value.size() != 16) {return NewErrorIterator(Status::Corruption("FileReader invoked with unexpected value"));} else {return cache->NewIterator(options,DecodeFixed64(file_value.data()),DecodeFixed64(file_value.data() + 8));}
   }
   

5. FindFile

   db/version_set.h 语义直接阅读注释即可.注意这里files都是排好序并且是不重叠的。按照二分法搜索largest key即可.

   //Return the smallest index i such that files[i]->largest >= key.
   //Return files.size() if there is no such file.
   //REQUIRES: "files" contains a sorted list of non-overlapping files.
   int FindFile(const InternalKeyComparator& icmp,const std::vector<FileMetaData*>& files,const Slice& key) {uint32_t left = 0;uint32_t right = files.size();while (left < right) {uint32_t mid = (left + right) / 2;const FileMetaData* f = files[mid];if (icmp.InternalKeyComparator::Compare(f->largest.Encode(), key) < 0) {//Key at "mid.largest" is < "target".  Therefore all//files at or before "mid" are uninteresting.left = mid + 1;} else {//Key at "mid.largest" is >= "target".  Therefore all files//after "mid" are uninteresting.right = mid;}}return right;
   }
   

6. SomeFileOverlapsRange

   db/version_set.h 判断是否有文件和某个范围[smallest,largest]相交.注意这里这里files也是按照key排序的.

   //Returns true iff some file in "files" overlaps the user key range
   //[*smallest,*largest].
   //smallest==NULL represents a key smaller than all keys in the DB.
   //largest==NULL represents a key largest than all keys in the DB.
   //REQUIRES: If disjoint_sorted_files, files[] contains disjoint ranges
   //in sorted order.
   bool SomeFileOverlapsRange(const InternalKeyComparator& icmp,bool disjoint_sorted_files, //表示filesconst std::vector<FileMetaData*>& files,const Slice* smallest_user_key,const Slice* largest_user_key) {const Comparator* ucmp = icmp.user_comparator();if (!disjoint_sorted_files) {//如果文件之间可能存在overlap的话.那么必须顺序.//那么需要遍历每个文件.判断这个文件是否和[small,large]相交.//如果有相交，那么直接返回.:).//Need to check against all filesfor (int i = 0; i < files.size(); i++) {const FileMetaData* f = files[i];//AfterFile和BeforeFile稍后分析.if (AfterFile(ucmp, smallest_user_key, f) ||BeforeFile(ucmp, largest_user_key, f)) {//No overlap} else {return true;  //Overlap}}return false;}//如果文件本身之间没有overlap的话.//那么我们只需要首先按照二分方法找到相交文件index.//然后针对这个index分析即可.//Binary search over file listuint32_t index = 0;if (smallest_user_key != NULL) {//Find the earliest possible internal key for smallest_user_keyInternalKey small(*smallest_user_key, kMaxSequenceNumber,kValueTypeForSeek);index = FindFile(icmp, files, small.Encode());}if (index >= files.size()) {//beginning of range is after all files, so no overlap.return false;}return !BeforeFile(ucmp, largest_user_key, files[index]);
   }
   

7. AfterFile

   判断user_key是否在这个file之后.

   static bool AfterFile(const Comparator* ucmp,const Slice* user_key, const FileMetaData* f) {//NULL user_key occurs before all keys and is therefore never after *freturn (user_key != NULL &&ucmp->Compare(*user_key, f->largest.user_key()) > 0);
   }
   

8. BeforeFile

   判断user_key是否在这个file之前.

   static bool BeforeFile(const Comparator* ucmp,const Slice* user_key, const FileMetaData* f) {//NULL user_key occurs after all keys and is therefore never before *freturn (user_key != NULL &&ucmp->Compare(*user_key, f->smallest.user_key()) < 0);
   }
   

9. GetStats

   这个结构反应的是查询到的一些附加结果可以用来指导compaction.

   struct GetStats {FileMetaData* seek_file; //被seek到的文件int seek_file_level; //以及这个文件所处level.
   };
   

10. NewestFirst

    按照file_number进行排序.逆序.越迟生成的file那么是最新的。

    static bool NewestFirst(FileMetaData* a, FileMetaData* b) {return a->number > b->number;
    }
    

11. Get

    Status Version::Get(const ReadOptions& options,const LookupKey& k,std::string* value,GetStats* stats) {Slice ikey = k.internal_key();Slice user_key = k.user_key();const Comparator* ucmp = vset_->icmp_.user_comparator();Status s;stats->seek_file = NULL;stats->seek_file_level = -1;FileMetaData* last_file_read = NULL;int last_file_read_level = -1;//We can search level-by-level since entries never hop across//levels.  Therefore we are guaranteed that if we find data//in an smaller level, later levels are irrelevant.std::vector<FileMetaData*> tmp;FileMetaData* tmp2;for (int level = 0; level < config::kNumLevels; level++) {size_t num_files = files_[level].size();if (num_files == 0) continue;//Get the list of files to search in this levelFileMetaData* const* files = &files_[level][0];if (level == 0) {//Level-0 files may overlap each other.  Find all files that//overlap user_key and process them in order from newest to oldest.//因为level0存在重叠，所以需要都进行搜索.tmp.reserve(num_files);for (uint32_t i = 0; i < num_files; i++) {FileMetaData* f = files[i];if (ucmp->Compare(user_key, f->smallest.user_key()) >= 0 &&ucmp->Compare(user_key, f->largest.user_key()) <= 0) {tmp.push_back(f);}}if (tmp.empty()) continue;//然后按照进行排序.最新生成的file放在最前面.std::sort(tmp.begin(), tmp.end(), NewestFirst);files = &tmp[0];num_files = tmp.size();} else {//对于其他level的话越低level越新.并且level内部没有overlap//那么可以通过简单的二分法来判断哪个文件.只会存在一个文件.//Binary search to find earliest index whose largest key >= ikey.uint32_t index = FindFile(vset_->icmp_, files_[level], ikey);if (index >= num_files) {files = NULL;num_files = 0;} else {tmp2 = files[index];if (ucmp->Compare(user_key, tmp2->smallest.user_key()) < 0) {//All of "tmp2" is past any data for user_keyfiles = NULL;num_files = 0;} else {files = &tmp2;num_files = 1;}}}for (uint32_t i = 0; i < num_files; ++i) {if (last_file_read != NULL && stats->seek_file == NULL) {//We have had more than one seek for this read.  Charge the 1st file.//stats这里只保留第一次读取的文件.stats->seek_file = last_file_read;stats->seek_file_level = last_file_read_level;}FileMetaData* f = files[i];last_file_read = f;last_file_read_level = level;//通过iterator的seek方式来进行查找.Iterator* iter = vset_->table_cache_->NewIterator(options,f->number,f->file_size);iter->Seek(ikey);//seek只是一个大概位置这里需要精确比较返回值.后面我们仔细分析GetValue.const bool done = GetValue(ucmp, iter, user_key, value, &s);if (!iter->status().ok()) {s = iter->status();delete iter;return s;} else {delete iter;if (done) {return s;}}}}return Status::NotFound(Slice());  //Use an empty error message for speed
    }
    

12. GetValue

    GetValue根据某个iter和key判断这个key是否为想查询的key.

    //If "*iter" points at a value or deletion for user_key, store
    //either the value, or a NotFound error and return true.
    //Else return false.
    static bool GetValue(const Comparator* cmp,Iterator* iter, const Slice& user_key,std::string* value,Status* s) {if (!iter->Valid()) {return false;}ParsedInternalKey parsed_key;//对于iterator里面是internal key.解析出parsed internal key出来.if (!ParseInternalKey(iter->key(), &parsed_key)) {*s = Status::Corruption("corrupted key for ", user_key);return true;}//比较user key内容是否一致.这里没有考虑sequence number.if (cmp->Compare(parsed_key.user_key, user_key) != 0) {return false;}//然后判断这个key是否标记删除.switch (parsed_key.type) {case kTypeDeletion:*s = Status::NotFound(Slice());  //Use an empty error message for speedbreak;case kTypeValue: {Slice v = iter->value();value->assign(v.data(), v.size());break;}}return true;
    }
    

13. UpdateStats

    根据stats来更新version内部状态.设置将要compaction文件以及对应的level是什么.

    bool Version::UpdateStats(const GetStats& stats) {FileMetaData* f = stats.seek_file; //如果stats里面标记了数据的话.if (f != NULL) {f->allowed_seeks--; //if (f->allowed_seeks <= 0 && file_to_compact_ == NULL) {file_to_compact_ = f;file_to_compact_level_ = stats.seek_file_level;return true;}}return false;
    }
    

14. GetOverlappingInputs

    得到某个level下面所有和[begin,end]有overlap的文件.语义不知道是否理解正确.因为代码里面有个地方没有太看懂.

    //Store in "*inputs" all files in "level" that overlap [begin,end]
    void Version::GetOverlappingInputs(int level,const InternalKey* begin,const InternalKey* end,std::vector<FileMetaData*>* inputs) {inputs->clear();Slice user_begin, user_end;if (begin != NULL) {user_begin = begin->user_key();}if (end != NULL) {user_end = end->user_key();}const Comparator* user_cmp = vset_->icmp_.user_comparator();for (size_t i = 0; i < files_[level].size(); ) {FileMetaData* f = files_[level][i++];const Slice file_start = f->smallest.user_key();const Slice file_limit = f->largest.user_key();if (begin != NULL && user_cmp->Compare(file_limit, user_begin) < 0) {//"f" is completely before specified range; skip it} else if (end != NULL && user_cmp->Compare(file_start, user_end) > 0) {//"f" is completely after specified range; skip it} else {inputs->push_back(f);if (level == 0) {//todo:实现上可能存在问题吧.看上去对于level0的话似乎在不断地过大范围.//Level-0 files may overlap each other.  So check if the newly//added file has expanded the range.  If so, restart search.if (begin != NULL && user_cmp->Compare(file_start, user_begin) < 0) {user_begin = file_start;inputs->clear();i = 0;} else if (end != NULL && user_cmp->Compare(file_limit, user_end) > 0) {user_end = file_limit;inputs->clear();i = 0;}}}}
    }
    

15. OverlapInLevel

    判断某个level里面是否有文件和[small,large]这个范围内容的key重叠

    bool Version::OverlapInLevel(int level,const Slice* smallest_user_key,const Slice* largest_user_key) {//(level>0)的话表示每个文件没有重叠的部分.return SomeFileOverlapsRange(vset_->icmp_, (level > 0), files_[level],smallest_user_key, largest_user_key);
    }
    

16. PickLevelForMemTableOutput

    根据[small,large]这个范畴选择某个level来作为memtable的output.这个range应该就是memtable的range.这个策略比较精巧。这个策略在常数部分定义注释里面给出来了。

    //db/dbformat.h//如果level0过低的话那么会造成过多compaction
    //但是level0多高的话那么会浪费磁盘空间因为key的重复.
    //Maximum level to which a new compacted memtable is pushed if it
    //does not create overlap.  We try to push to level 2 to avoid the
    //relatively expensive level 0=>1 compactions and to avoid some
    //expensive manifest file operations.  We do not push all the way to
    //the largest level since that can generate a lot of wasted disk
    //space if the same key space is being repeatedly overwritten.
    static const int kMaxMemCompactLevel = 2;//db/version_set.cc
    static const int kTargetFileSize = 2 * 1048576;//Maximum bytes of overlaps in grandparent (i.e., level+2) before we
    //stop building a single file in a level->level+1 compaction.
    static const int64_t kMaxGrandParentOverlapBytes = 10 * kTargetFileSize;
    

    #=BEGIN_SRC C++int Version::PickLevelForMemTableOutput( const Slice& smallest_user_key, const Slice& largest_user_key) { int level = 0; // 首先判断和level0是否有overlap.如果有overlapd的话那么选择level0. 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) {/ 判断和其他level是否有overlap.如果存在overlap的那么也选择. / >0的level是不允许overlap的. if (OverlapInLevel(level + 1, &smallest_user_key, &largest_user_key)) { break; }/ 如果没有overlap的话那么判断和grandparent level重叠文件.判断重叠文件大小. / 这个策略没有太明白. / 关于这个策略估计可以从Compaction::IsTrivialMove的注释里面看到.这里应该是担心grand parent本身就存在很多overlap / 如果放在level+1做compaction的话，会造成grand parent这个部分合并时间过长./ 从值上来看意思应该是存在overlap的文件数目不应该超过10个(假设每个overlap文件都是MaxSizeForLevel的). / Avoid a move if there is lots of overlapping grandparent data./ Otherwise, the move could create a parent file that will require / a very expensive merge later on. GetOverlappingInputs(level + 2, &start, &limit, &overlaps); const int64_t sum = TotalFileSize(overlaps); if (sum > kMaxGrandParentOverlapBytes) { break; } level++; } } return level;}#+END_SRC

17. DebugString

    DebugString作为Version的调试信息打印出来.我们可以稍微得到一点启发吧.尤其如果是自己调试的话.

    std::string Version::DebugString() const {std::string r;for (int level = 0; level < config::kNumLevels; level++) {//E.g.,//--- level 1 ---//17:123['a' .. 'd']//20:43['e' .. 'g']r.append("--- level ");AppendNumberTo(&r, level);r.append(" ---\n");const std::vector<FileMetaData*>& files = files_[level];for (size_t i = 0; i < files.size(); i++) {r.push_back(' ');AppendNumberTo(&r, files[i]->number); //file_numberr.push_back(':');AppendNumberTo(&r, files[i]->file_size); //file大小.r.append("[");r.append(files[i]->smallest.DebugString()); //最小和最大的key的打印.r.append(" .. ");r.append(files[i]->largest.DebugString());r.append("]\n");}}return r;
    }
    

10.3.12 VersionSet

class VersionSet {public:VersionSet(const std::string& dbname,const Options* options,TableCache* table_cache,const InternalKeyComparator*);~VersionSet();private:Env* const env_;const std::string dbname_;const Options* const options_;TableCache* const table_cache_;const InternalKeyComparator icmp_;uint64_t next_file_number_; //下一个分配的file numberuint64_t manifest_file_number_; //manifest file number.uint64_t last_sequence_; //下次操作的sequence.//从LogNumber和PrevLogNumber注释可以看出来//分别表示当前使用的log number以及之前使用的log number(当前正在被压缩).uint64_t log_number_; //uint64_t prev_log_number_;  //0 or backing store for memtable being compacted//Opened lazilyWritableFile* descriptor_file_; //manifest文件log::Writer* descriptor_log_; //manifest以log形式打开.Version dummy_versions_;  //Head of circular doubly-linked list of versions. // 为了组织称为链表.Version* current_;        //== dummy_versions_.prev_ // 当前version//Per-level key at which the next compaction at that level should start.//Either an empty string, or a valid InternalKey.//对于每层的话都会设置一个key.只有>这层设置的key才允许进行compaction.std::string compact_pointer_[config::kNumLevels]; //每个level进行compaction使用的internal key//todo:似乎每个level进行compaction都配上了一个internal key//DONE:记录每层的最大key.
};

1. VersionSet

   首先看看构造函数和析构函数.

   VersionSet::VersionSet(const std::string& dbname,const Options* options,TableCache* table_cache,const InternalKeyComparator* cmp): env_(options->env),dbname_(dbname),options_(options),table_cache_(table_cache),icmp_(*cmp),next_file_number_(2),//下面这些字段都是通过Recover恢复的.manifest_file_number_(0),  //Filled by Recover()last_sequence_(0),log_number_(0),prev_log_number_(0),descriptor_file_(NULL),descriptor_log_(NULL),dummy_versions_(this),current_(NULL) {AppendVersion(new Version(this)); //添加一个当前version.
   }
   VersionSet::~VersionSet() {current_->Unref();//析构时候必须确保里面没有任何版本.assert(dummy_versions_.next_ == &dummy_versions_);  //List must be emptydelete descriptor_log_;delete descriptor_file_;
   }
   

2. NeedsCompaction

   当前versionset是否需要触发compaction操作.

   //Returns true iff some level needs a compaction.
   bool NeedsCompaction() const {Version* v = current_;return (v->compaction_score_ >= 1) || (v->file_to_compact_ != NULL);
   }
   

3. AppendVersion

   添加version.非常简单修改引用计数挂载链表上.将version挂在version_set下面的话可以使得version_set了解到哪些文件依然是被正在使用的.

   void VersionSet::AppendVersion(Version* v) {//Make "v" currentassert(v->refs_ == 0);assert(v != current_);if (current_ != NULL) {current_->Unref();}current_ = v;v->Ref();//Append to linked listv->prev_ = dummy_versions_.prev_;v->next_ = &dummy_versions_;v->prev_->next_ = v;v->next_->prev_ = v;
   }
   

4. NumLevelFiles

   得到某个level的文件数目.以current这个Version来计算的

   int VersionSet::NumLevelFiles(int level) const {assert(level >= 0);assert(level < config::kNumLevels);return current_->files_[level].size();
   }
   

5. NumLevelBytes

   某个level的文件大小.以current这个Version来计算的.

   int64_t VersionSet::NumLevelBytes(int level) const {assert(level >= 0);assert(level < config::kNumLevels);return TotalFileSize(current_->files_[level]);
   }
   

6. TotalFileSize

   根据file meta data得到所有文件大小.非常简单因为里面有file_size.

   static int64_t TotalFileSize(const std::vector<FileMetaData*>& files) {int64_t sum = 0;for (size_t i = 0; i < files.size(); i++) {sum += files[i]->file_size;}return sum;
   }
   

7. MarkFileNumberUsed

   标记当前使用了file_number进度.number表示已经使用的进度,所以下次分配必须+1.

   void VersionSet::MarkFileNumberUsed(uint64_t number) {if (next_file_number_ <= number) {next_file_number_ = number + 1;}
   }
   

8. ApproximateOffsetOf

   判断这个key在数据库内部大致偏移是多少.从这个实现里面我们可以看到.对于level-0的文件并不是排序的对于level>0的文件都是按照range进行排序的。并且这个排序是按照smallest来进行排序的。

   uint64_t VersionSet::ApproximateOffsetOf(Version* v, const InternalKey& ikey) {uint64_t result = 0;for (int level = 0; level < config::kNumLevels; level++) {const std::vector<FileMetaData*>& files = v->files_[level];for (size_t i = 0; i < files.size(); i++) {if (icmp_.Compare(files[i]->largest, ikey) <= 0) { //如果比largest key大的话那么这个//那么偏移需要算上整个文件.但是这里还包括了index block,meta block以及footer大小.//所以只能够说是大致大小.//Entire file is before "ikey", so just add the file sizeresult += files[i]->file_size;} else if (icmp_.Compare(files[i]->smallest, ikey) > 0) { //如果smallest比当前key大的话那么取消.//Entire file is after "ikey", so ignoreif (level > 0) { //如果是>0的level那么可以直接取消.因为这个部分的file都是按照smallest排序的.//Files other than level 0 are sorted by meta->smallest, so//no further files in this level will contain data for//"ikey".break;}} else { //如果存在交集的话那么那么可以构造出Table对象找到这个key大致偏移.//"ikey" falls in the range for this table.  Add the//approximate offset of "ikey" within the table.Table* tableptr;Iterator* iter = table_cache_->NewIterator(ReadOptions(), files[i]->number, files[i]->file_size, &tableptr);if (tableptr != NULL) {result += tableptr->ApproximateOffsetOf(ikey.Encode());}delete iter;}}}return result;
   }
   

9. AddLiveFiles

   将version set里面所有version所持有的文件收集起来.非常简单:).

   void VersionSet::AddLiveFiles(std::set<uint64_t>* live) {for (Version* v = dummy_versions_.next_;v != &dummy_versions_;v = v->next_) {for (int level = 0; level < config::kNumLevels; level++) {const std::vector<FileMetaData*>& files = v->files_[level];for (size_t i = 0; i < files.size(); i++) {live->insert(files[i]->number);}}}
   }
   

10. LevelSummary

    每层level的文件个数分别是多少.

    const char* VersionSet::LevelSummary(LevelSummaryStorage* scratch) const {//Update code if kNumLevels changesassert(config::kNumLevels == 7);snprintf(scratch->buffer, sizeof(scratch->buffer),"files[ %d %d %d %d %d %d %d ]",int(current_->files_[0].size()),int(current_->files_[1].size()),int(current_->files_[2].size()),int(current_->files_[3].size()),int(current_->files_[4].size()),int(current_->files_[5].size()),int(current_->files_[6].size()));return scratch->buffer;
    }
    

11. WriteSnapshot

    将当前current version写入到磁盘记为log.方式是将内容copy到version edit对象里面去然后持久化.过程还是非常简单吧:).

    Status VersionSet::WriteSnapshot(log::Writer* log) {//TODO: Break up into multiple records to reduce memory usage on recovery?//Save metadataVersionEdit edit;edit.SetComparatorName(icmp_.user_comparator()->Name());//Save compaction pointersfor (int level = 0; level < config::kNumLevels; level++) {if (!compact_pointer_[level].empty()) {InternalKey key;key.DecodeFrom(compact_pointer_[level]);edit.SetCompactPointer(level, key);}}//Save filesfor (int level = 0; level < config::kNumLevels; level++) {const std::vector<FileMetaData*>& files = current_->files_[level];for (size_t i = 0; i < files.size(); i++) {const FileMetaData* f = files[i];edit.AddFile(level, f->number, f->file_size, f->smallest, f->largest);}}std::string record;edit.EncodeTo(&record);return log->AddRecord(record);
    }
    

12. Finalize

    所谓的Finalize的含义应该是，如果我们不希望修改version这个结构之后我们应该做的事情。可能叫做Finish会更好。相当于针对这个version而言的话，最高一级的compaction level是什么，分数是多少.

    void VersionSet::Finalize(Version* v) {//Precomputed best level for next compactionint best_level = -1;double best_score = -1;for (int level = 0; level < config::kNumLevels-1; level++) {double score;if (level == 0) {//We treat level-0 specially by bounding the number of files//instead of number of bytes for two reasons:////(1) With larger write-buffer sizes, it is nice not to do too//many level-0 compactions.////(2) The files in level-0 are merged on every read and//therefore we wish to avoid too many files when the individual//file size is small (perhaps because of a small write-buffer//setting, or very high compression ratios, or lots of//overwrites/deletions).score = v->files_[level].size() /static_cast<double>(config::kL0_CompactionTrigger);} else {//Compute the ratio of current size to size limit.const uint64_t level_bytes = TotalFileSize(v->files_[level]);score = static_cast<double>(level_bytes) / MaxBytesForLevel(level);}if (score > best_score) {best_level = level;best_score = score;}}v->compaction_level_ = best_level;v->compaction_score_ = best_score;
    }
    

13. Recover

    从db_impl.cc里面的DB::Open可以看到,这里的Recover并没有将剩余的version_edit回放完成.回放这个部分在LogAndApply里面完成.所以Recover可能只是恢复到以前某个状态.将当前的CURRENT里面的manifest文件回放之后就得到最新的内容.然后根据里面的log信息回放后面的内容.

    Status VersionSet::Recover() {struct LogReporter : public log::Reader::Reporter {Status* status;virtual void Corruption(size_t bytes, const Status& s) {if (this->status->ok()) *this->status = s;}};//读取当前current这个内容,得到最新的manifest文件.//Read "CURRENT" file, which contains a pointer to the current manifest filestd::string current;Status s = ReadFileToString(env_, CurrentFileName(dbname_), &current);if (!s.ok()) {return s;}if (current.empty() || current[current.size()-1] != '\n') {return Status::Corruption("CURRENT file does not end with newline");}current.resize(current.size() - 1);std::string dscname = dbname_ + "/" + current;SequentialFile* file;s = env_->NewSequentialFile(dscname, &file);if (!s.ok()) {return s;}bool have_log_number = false;bool have_prev_log_number = false;bool have_next_file = false;bool have_last_sequence = false;uint64_t next_file = 0;uint64_t last_sequence = 0;uint64_t log_number = 0;uint64_t prev_log_number = 0;//相当于从empty status来构建current version.Builder builder(this, current_);{LogReporter reporter;reporter.status = &s;log::Reader reader(file, &reporter, true/*checksum*/, 0/*initial_offset*/);Slice record;std::string scratch;//按照每条log内容读取.里面的内容都是version edit.while (reader.ReadRecord(&record, &scratch) && s.ok()) {VersionEdit edit;s = edit.DecodeFrom(record);if (s.ok()) {if (edit.has_comparator_ &&edit.comparator_ != icmp_.user_comparator()->Name()) {s = Status::InvalidArgument(edit.comparator_ + "does not match existing comparator ",icmp_.user_comparator()->Name());}}//直接apply上去.if (s.ok()) {builder.Apply(&edit);}if (edit.has_log_number_) {log_number = edit.log_number_;have_log_number = true;}if (edit.has_prev_log_number_) {prev_log_number = edit.prev_log_number_;have_prev_log_number = true;}if (edit.has_next_file_number_) {next_file = edit.next_file_number_;have_next_file = true;}if (edit.has_last_sequence_) {last_sequence = edit.last_sequence_;have_last_sequence = true;}}}delete file;file = NULL;if (s.ok()) {if (!have_next_file) {s = Status::Corruption("no meta-nextfile entry in descriptor");} else if (!have_log_number) {s = Status::Corruption("no meta-lognumber entry in descriptor");} else if (!have_last_sequence) {s = Status::Corruption("no last-sequence-number entry in descriptor");}if (!have_prev_log_number) {prev_log_number = 0;}MarkFileNumberUsed(prev_log_number);MarkFileNumberUsed(log_number);}if (s.ok()) {Version* v = new Version(this);builder.SaveTo(v);//Install recovered versionFinalize(v);//增加这么一个version作为当前current version.AppendVersion(v);manifest_file_number_ = next_file;next_file_number_ = next_file + 1;last_sequence_ = last_sequence;log_number_ = log_number;prev_log_number_ = prev_log_number;}return s;
    }
    

14. LogAndApply

    Status VersionSet::LogAndApply(VersionEdit* edit, port::Mutex* mu) {if (edit->has_log_number_) {assert(edit->log_number_ >= log_number_);assert(edit->log_number_ < next_file_number_);} else {edit->SetLogNumber(log_number_);}if (!edit->has_prev_log_number_) {edit->SetPrevLogNumber(prev_log_number_);}edit->SetNextFile(next_file_number_);edit->SetLastSequence(last_sequence_);//这个过程应该是将edit的修改放在current_里面//然后将edit作用在current上面，将current_保存到v里面//然后对v进行finish.Version* v = new Version(this);{Builder builder(this, current_);builder.Apply(edit);builder.SaveTo(v);}Finalize(v);//Initialize new descriptor log file if necessary by creating//a temporary file that contains a snapshot of the current version.std::string new_manifest_file;Status s;if (descriptor_log_ == NULL) {//No reason to unlock *mu here since we only hit this path in the//first call to LogAndApply (when opening the database).assert(descriptor_file_ == NULL);new_manifest_file = DescriptorFileName(dbname_, manifest_file_number_);edit->SetNextFile(next_file_number_);s = env_->NewWritableFile(new_manifest_file, &descriptor_file_);if (s.ok()) {descriptor_log_ = new log::Writer(descriptor_file_);//将current做snapshot保存在descriptor log里面.s = WriteSnapshot(descriptor_log_);}}//Unlock during expensive MANIFEST log write{mu->Unlock();//Write new record to MANIFEST log//将这个修改也保存在log里面.//现在比较担心最后的结构.似乎现在snapshot以及log内容都混在一起了.//之前在descriptor_log里面填写了current version内容//而这里还添加了edit增量内容//这个格式现在看上去还好.到时候可能需要主要看看Recover内容.if (s.ok()) {std::string record;edit->EncodeTo(&record);s = descriptor_log_->AddRecord(record);if (s.ok()) {s = descriptor_file_->Sync(); //注意这里我们需要做sync.}}//If we just created a new descriptor file, install it by writing a//new CURRENT file that points to it.if (s.ok() && !new_manifest_file.empty()) {s = SetCurrentFile(env_, dbname_, manifest_file_number_);}mu->Lock();}//Install the new version//作为当前current version存在.if (s.ok()) {AppendVersion(v);log_number_ = edit->log_number_;prev_log_number_ = edit->prev_log_number_;} else {delete v;if (!new_manifest_file.empty()) {delete descriptor_log_;delete descriptor_file_;descriptor_log_ = NULL;descriptor_file_ = NULL;env_->DeleteFile(new_manifest_file);}}return s;
    }
    

15. GetRange

    得到input files里面的最大和最小key.

    //Stores the minimal range that covers all entries in inputs in
    //*smallest, *largest.
    //REQUIRES: inputs is not empty
    void VersionSet::GetRange(const std::vector<FileMetaData*>& inputs,InternalKey* smallest,InternalKey* largest) {assert(!inputs.empty());smallest->Clear();largest->Clear();for (size_t i = 0; i < inputs.size(); i++) {FileMetaData* f = inputs[i];if (i == 0) {*smallest = f->smallest;*largest = f->largest;} else {if (icmp_.Compare(f->smallest, *smallest) < 0) {*smallest = f->smallest;}if (icmp_.Compare(f->largest, *largest) > 0) {*largest = f->largest;}}}
    }
    

16. PickCompaction

    选择一个Compaction来进行操作.对于选择Compaction应该比较具有策略性的。然后丢给DB去执行这个Compaction操作.PickCompaction和CompactRange都是compaction行为.不过PickCompaction是自动触发的，而CompactRange是用户自己手动触发的.

    Compaction* VersionSet::PickCompaction() {Compaction* c;int level;//We prefer compactions triggered by too much data in a level over  // the compactions triggered by seeks.//根据当前version的统计结果判断是否需要compaction.const bool size_compaction = (current_->compaction_score_ >= 1);const bool seek_compaction = (current_->file_to_compact_ != NULL);if (size_compaction) {level = current_->compaction_level_;assert(level >= 0);assert(level+1 < config::kNumLevels);c = new Compaction(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 spacec->inputs_[0].push_back(current_->files_[level][0]);}} else if (seek_compaction) {level = current_->file_to_compact_level_;c = new Compaction(level);c->inputs_[0].push_back(current_->file_to_compact_);} else {return NULL;}//现在已经选择好了一个文件来进行compaction.//对当前的version +ref count.c->input_version_ = current_;c->input_version_->Ref();//Files in level 0 may overlap each other, so pick up all overlapping onesif (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());}//选择了level的文件之后需要选择level+1的文件//把其他的input files字段填上.SetupOtherInputs(c);return c;
    }
    

17. CompactRange

    DBImpl里面提供CompactRange打包成为一个Manual Compaction的请求提交到后端.然后判断如果为Manual Compaction的话那么实际调用的还是VersionSet::CompactRange这个方法产生一个Compaction对象真正进行执行.

    Compaction* VersionSet::CompactRange(int level,const InternalKey* begin,const InternalKey* end) {std::vector<FileMetaData*> inputs;current_->GetOverlappingInputs(level, begin, end, &inputs);if (inputs.empty()) {return NULL;}//Avoid compacting too much in one shot in case the range is large.//压缩范围的话可以提交多个input files作为底层level输入.//但是通过控制大小来限制输入文件多少.const uint64_t limit = MaxFileSizeForLevel(level);uint64_t total = 0;for (int i = 0; i < inputs.size(); i++) {uint64_t s = inputs[i]->file_size;total += s;if (total >= limit) {inputs.resize(i + 1);break;}}Compaction* c = new Compaction(level);c->input_version_ = current_;c->input_version_->Ref();c->inputs_[0] = inputs;SetupOtherInputs(c);return c;
    }
    

18. GetRange2

    得到input1和input2两个文件compaction之后的最小和最大key.

    //Stores the minimal range that covers all entries in inputs1 and inputs2
    //in *smallest, *largest.
    //REQUIRES: inputs is not empty
    void VersionSet::GetRange2(const std::vector<FileMetaData*>& inputs1,const std::vector<FileMetaData*>& inputs2,InternalKey* smallest,InternalKey* largest) {std::vector<FileMetaData*> all = inputs1;all.insert(all.end(), inputs2.begin(), inputs2.end());GetRange(all, smallest, largest);
    }
    

19. SetupOtherInputs

    将compaction的另外一层内容(input files)补齐.

    void VersionSet::SetupOtherInputs(Compaction* c) {const int level = c->level();InternalKey smallest, largest;GetRange(c->inputs_[0], &smallest, &largest);//判断level+1里面有哪些文件是存在重叠的.current_->GetOverlappingInputs(level+1, &smallest, &largest, &c->inputs_[1]);//Get entire range covered by compaction//得到整个compaction所覆盖的范围InternalKey all_start, all_limit;GetRange2(c->inputs_[0], c->inputs_[1], &all_start, &all_limit);//See if we can grow the number of inputs in "level" without//changing the number of "level+1" files we pick up.//观察是否可以扩大level input files数量但是不造成level+1 input files数量变化.if (!c->inputs_[1].empty()) {std::vector<FileMetaData*> expanded0;current_->GetOverlappingInputs(level, &all_start, &all_limit, &expanded0);if (expanded0.size() > c->inputs_[0].size()) {InternalKey new_start, new_limit;GetRange(expanded0, &new_start, &new_limit);std::vector<FileMetaData*> expanded1;current_->GetOverlappingInputs(level+1, &new_start, &new_limit,&expanded1);if (expanded1.size() == c->inputs_[1].size()) {Log(options_->info_log,"Expanding@%d %d+%d to %d+%d\n",level,int(c->inputs_[0].size()),int(c->inputs_[1].size()),int(expanded0.size()),int(expanded1.size()));smallest = new_start;largest = new_limit;c->inputs_[0] = expanded0;c->inputs_[1] = expanded1;GetRange2(c->inputs_[0], c->inputs_[1], &all_start, &all_limit);}}}//Compute the set of grandparent files that overlap this compaction//(parent == level+1; grandparent == level+2)if (level + 2 < config::kNumLevels) {//假设这两个合并之后,和level+2会有哪些文件出现重叠.//保存在grandparents里面.current_->GetOverlappingInputs(level + 2, &all_start, &all_limit,&c->grandparents_);}if (false) {Log(options_->info_log, "Compacting %d '%s' .. '%s'",level,smallest.DebugString().c_str(),largest.DebugString().c_str());}//Update the place where we will do the next compaction for this level.//We update this immediately instead of waiting for the VersionEdit//to be applied so that if the compaction fails, we will try a different//key range next time.//保存到本层最大key.compact_pointer_[level] = largest.Encode().ToString();//将level的largest保存到edit里面.这个可能对于恢复起来之后可能有用.c->edit_.SetCompactPointer(level, largest);
    }
    

20. MakeInputIterator

    针对Compaction而言我们可能需要读取多个文件设计的遍历器.底层应该是MergeIterator实现.

    Iterator* VersionSet::MakeInputIterator(Compaction* c) {ReadOptions options;options.verify_checksums = options_->paranoid_checks;options.fill_cache = false;//Level-0 files have to be merged together.  For other levels,//we will make a concatenating iterator per level.//TODO(opt): use concatenating iterator for level-0 if there is no overlap//对于>0的level来说的话可以使用LevelFileNumIterator来当做这一层的iterator.//而对==0的level来说的话每一个文件作为一个iterator存在.const int space = (c->level() == 0 ? c->inputs_[0].size() + 1 : 2);Iterator** list = new Iterator*[space];int num = 0;for (int which = 0; which < 2; which++) {if (!c->inputs_[which].empty()) {if (c->level() + which == 0) {const std::vector<FileMetaData*>& files = c->inputs_[which];for (size_t i = 0; i < files.size(); i++) {list[num++] = table_cache_->NewIterator(options, files[i]->number, files[i]->file_size);}} else {//Create concatenating iterator for the files from this levellist[num++] = NewTwoLevelIterator(new Version::LevelFileNumIterator(icmp_, &c->inputs_[which]),&GetFileIterator, table_cache_, options);}}}assert(num <= space);//然后作为一个merge iterator存在.Iterator* result = NewMergingIterator(&icmp_, list, num);delete[] list;return result;
    }