levelDB源码剖析(6)--levelDB组件

levelDB的组件Comparator、Status、Env、Options

levelDB中有很多常用的组件，下面我们来对其进行介绍。

• Comparator : 定义了比较的规则，是一个虚基类。
• Status : 定义函数执行的结果信息。
• Env : 封装系统相关的调用，比如文件操作，线程操作。
• Options : 指定数据库选项。

Comparator

源码位置与说明：

utils/comparator.h utils/comparator.cc : 虚基类与BytewiseComparatorImpl类头文件与实现

db/dbformat.h db/format.cc : InternalKeyComparator类头文件与实现

Comparator定义了比较规则，这个是一个虚基类，所以如果想要实现自定义的接口需要实现自己的类。

其中levelDB基于Comparator实现了一些内置比较类：BytewiseComparatorImpl和InternalKeyComparator，两者的作用是不同的。

• BytewiseComparatorImpl : 实现key的按二进制来进行比较。
• InternalKeyComparator : 用于内部的key比较器，基于BytewiseComparatorImpl。

由于levelDB是key-value的结构，其中key就是Slice对象。所以BytewiseComparatorImpl用于key之间的比较，这个key称为usr-key。

但是levelDB内部的存储的key结构并不是usr-key，所以需要默认定义两种比较类。

Comparator源码剖析

首先我们来看一下虚基类的定义。

class LEVELDB_EXPORT Comparator
{
public:
    // 虚析构函数
    virtual ~Comparator();
    // 进行比较的方法
    virtual int Compare(const Slice &a, const Slice &b) const = 0;

    // 返回comparator的名字，用于确认
    virtual const char *Name() const = 0;

    // 将start更改为一个位于[start, limit)里的最短的字符串。
    // 也就是找到start和limit中的公共字符字串，并把这个字串最后一个字符ascii+1
    // 这主要是为了优化SSTable里的Index Block里的索引项的长度，使得索引更短。
    // 因为每一个Data Block对应的索引项大于等于这个Data Block的最后一个项，
    // 而小于下一个Data Block的第一个项，通过这个函数可以减小索引项的长度；
    virtual void FindShortestSeparator(std::string *start,
                                       const Slice &limit) const = 0;

    // 将key更改为大于key的最短的key
    // 也就是取出key的第一个字节，然后+1
    // 这也是为了减小索引项的长度
    // 不过这是优化一个SSTable里最后一个索引项的。
    virtual void FindShortSuccessor(std::string *key) const = 0;
};

其中FindShortestSeparator和FindShortSuccessor可能会令人感到疑惑，这两个方法的作用是缩短索引。因为levelDB内部是有序的，假如有这样的两个索引"Aaaaa"和"Acccc"，其实我们只需要"Ab"就能将两者分开了；另外，假如一个表中的最后一项是"Yaaaa"，我们只需要"Z"就能确定这个项的上界了。这就是这两个函数能减短索引的原理。

接下来我们来看一下BytewiseComparatorImpl类

class BytewiseComparatorImpl : public Comparator
{
public:
    BytewiseComparatorImpl() = default;
    const char *Name() const override { return "leveldb.BytewiseComparator"; }
    int Compare(const Slice &a, const Slice &b) const override
    {
        return a.compare(b);
    }

    void FindShortestSeparator(std::string *start,
                               const Slice &limit) const override
    {
        // 先找到两者之间的最长子串
        size_t min_length = std::min(start->size(), limit.size());
        size_t diff_index = 0;
        while ((diff_index < min_length) &&
               ((*start)[diff_index] == limit[diff_index]))
        {
            diff_index++;
        }

        if (diff_index >= min_length)
        {
            // 如果一个串是另一个串的前缀就什么也不做
        }
        else
        {
            // 否则把start的最后一个字节+1(如果没超过255的话)
            uint8_t diff_byte = static_cast<uint8_t>((*start)[diff_index]);
            if (diff_byte < static_cast<uint8_t>(0xff) &&
                diff_byte + 1 < static_cast<uint8_t>(limit[diff_index]))
            {
                (*start)[diff_index]++;
                start->resize(diff_index + 1);
                assert(Compare(*start, limit) < 0);
            }
        }
    }

    void FindShortSuccessor(std::string *key) const override
    {
        // 找到第一个能够+1的字节
        size_t n = key->size();
        for (size_t i = 0; i < n; i++)
        {
            const uint8_t byte = (*key)[i];
            if (byte != static_cast<uint8_t>(0xff))
            {
                (*key)[i] = byte + 1;
                key->resize(i + 1);
                return;
            }
        }
        // 如果全是0xff，就不管了
    }
};

// 这个函数返回一个BytewiseComparatorImpl实例
// 之所以存在这个函数是为了保证线程安全
// 可以看https://www.zhihu.com/question/267013757
const Comparator *BytewiseComparator()
{
    static NoDestructor<BytewiseComparatorImpl> singleton;
    return singleton.get();
}

这部分比较的规则其实就是我们前面看到的虚基类的具体化。需要注意它存在一个为了保证线程安全的方法，进一步了解可以看此处。

接下来我们来看一下InternalKeyComparator的头文件和实现文件。

// 头文件

class InternalKeyComparator : public Comparator
{
private:
    // 多了一个私有成员，存放usr-key的比较函数
    const Comparator *user_comparator_;

public:
    // 前面这些接口都非常简单
    explicit InternalKeyComparator(const Comparator *c) : user_comparator_(c) {}
    const char *Name() const override;
    int Compare(const Slice &a, const Slice &b) const override;
    void FindShortestSeparator(std::string *start,
                               const Slice &limit) const override;
    void FindShortSuccessor(std::string *key) const override;

    // 返回usr-key的比较器，应该是用于判断
    const Comparator *user_comparator() const { return user_comparator_; }
    // 这个也是比较的，InternalKey就是内部的key类型
    int Compare(const InternalKey &a, const InternalKey &b) const;
};

// 实现文件

const char *InternalKeyComparator::Name() const
{
    return "leveldb.InternalKeyComparator";
}

// 比较key的类，调用的是usr-key的比较
int InternalKeyComparator::Compare(const Slice &akey, const Slice &bkey) const
{
    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)
        {
            r = -1;
        }
        else if (anum < bnum)
        {
            r = +1;
        }
    }
    return r;
}

// 找到最短的分割函数
void InternalKeyComparator::FindShortestSeparator(std::string *start,
                                                  const Slice &limit) const
{
    Slice user_start = ExtractUserKey(*start);
    Slice user_limit = ExtractUserKey(limit);
    std::string tmp(user_start.data(), user_start.size());
    user_comparator_->FindShortestSeparator(&tmp, user_limit);
    if (tmp.size() < user_start.size() &&
        user_comparator_->Compare(user_start, tmp) < 0)
    {
        PutFixed64(&tmp,
                   PackSequenceAndType(kMaxSequenceNumber, kValueTypeForSeek));
        assert(this->Compare(*start, tmp) < 0);
        assert(this->Compare(tmp, limit) < 0);
        start->swap(tmp);
    }
}

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);
    if (tmp.size() < user_key.size() &&
        user_comparator_->Compare(user_key, tmp) < 0)
    {
        PutFixed64(&tmp,
                   PackSequenceAndType(kMaxSequenceNumber, kValueTypeForSeek));
        assert(this->Compare(*key, tmp) < 0);
        key->swap(tmp);
    }
}

// 比较内部key的函数
inline int InternalKeyComparator::Compare(const InternalKey &a,
                                          const InternalKey &b) const
{
    return Compare(a.Encode(), b.Encode());
}

我们可以看到InternalKeyComparator的比较其实也是基于usr-key的，不过由于内部key与usr-key存在差异，剩余的代码主要是处理差异的，具体我们先不展开。

Status

源码位置： leveldb/include/status.h util/status.cc

这个类主要定义了很多操作的返回码，很多操作需要通过返回的status来判断下一步的行为。

Status源码剖析

我们接下来来看一下它的头文件。

class LEVELDB_EXPORT Status
{
private:
    //返回码
    enum Code
    {
        kOk = 0,
        kNotFound = 1,
        kCorruption = 2,
        kNotSupported = 3,
        kInvalidArgument = 4,
        kIOError = 5
    };

    Code code() const
    {
        return (state_ == nullptr) ? kOk : static_cast<Code>(state_[4]);
    }
    Status(Code code, const Slice &msg, const Slice &msg2);
    static const char *CopyState(const char *s);

    // 成功的时候state_为空指针，否则指向一个由new[]分配的数组
    // 其中[0:3]为信息的长度
    // [4]为错误码
    // [5:]为信息
    const char *state_;

public:
    // 构造与析构函数等等一个类的必须的函数
    Status() noexcept : state_(nullptr) {}
    ~Status() { delete[] state_; }
    Status(const Status &rhs);
    Status &operator=(const Status &rhs);
    Status(Status &&rhs) noexcept : state_(rhs.state_) { rhs.state_ = nullptr; }
    Status &operator=(Status &&rhs) noexcept;

    // 返回成功或错误类型
    static Status OK() { return Status(); }
    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);
    }

    // state_为空就是成功
    bool ok() const { return (state_ == nullptr); }

    // 否则根据code()返回值来判断错误类型
    bool IsNotFound() const { return code() == kNotFound; }
    bool IsCorruption() const { return code() == kCorruption; }
    bool IsIOError() const { return code() == kIOError; }
    bool IsNotSupportedError() const { return code() == kNotSupported; }
    bool IsInvalidArgument() const { return code() == kInvalidArgument; }

    // 根据错误类型返回一个字符串用于打印
    std::string ToString() const;
};

由于这个类主要就是处理错误类型的，所以其实现也是非常简单的，此处就不展开了。

不过实现内部有一个小技巧：用memcpy去进行赋值。这个似乎会更快一些，不过我觉得这个还是用于大数组会比较好吧，毕竟针对内存块有SIMD之类的优化。

Env

源码位置：

include/leveldb/env.h : env相关的接口定义

util/env_posix.cc util/posix_logger.h : Posix系统相关的封装，包括文件操作，文件锁，后台线程创建,Posix写日志

util/env_windows.cc util/windows_logger.h : Windows相关的实现

LevelDB是一个数据库函数库，数据库总是需要操作文件和线程，这就需要做很多系统调用。各个操作系统的系统调用方式不一样，为了跨平台支持，LevelDB对这些系统调用做了一层封装，提供了统一的接口来操作，并且提供了Posix和Windows两种实现，如果需要实现其他的系统，只需要根据系统实现相应的Env即可。

不过由于这些代码实在是太多，而且也就是一些系统调用，所以此处不展开了。

里面感觉有用的就是关于多线程的一个锁的问题，但是也相对比较简单。

Options

源码位置：leveldb/include/options.h util/options.cc

Options定义了操作数据库的选项，定义了3个struct来操作：

• Options定义打开数据库的选项
• ReadOptions定义读操作相关的选项
• WriteOptions定义写操作相关的选项

同样的，这部分源码也非常简单，也不展开了

总结

编程小技巧

• 可以定义虚基类来规定接口。
• 需要处理线程安全问题。
• memcpy进行数组赋值会更快。