C++从零开始实现LSM-Tree-KV存储-05-文件IO

项目
,

这一小节实现基于之前的Block和SST完成Block的读取和SST的写入。

代码仓库：https://github.com/ToniXWD/toni-lsm

1 文件IO设计

这里先回顾之前我们在之前实现SST和Block中对文件IO的使用:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
std::shared_ptr<SST>
SSTBuilder::build(size_t sst_id, const std::string &path,
                  std::shared_ptr<BlockCache> block_cache) {
  // 数据准备代码
  ...
  
  // 创建文件
  FileObj file = FileObj::create_and_write(path, file_content);

  // 返回SST对象
  auto res = std::make_shared<SST>();

  res->sst_id = sst_id;
  res->file = std::move(file);
  res->first_key = meta_entries.front().first_key;
  res->last_key = meta_entries.back().last_key;
  res->meta_block_offset = meta_offset;
  res->meta_entries = std::move(meta_entries);
  res->block_cache = block_cache;

  return res;
}

可以看到, 在创建SST时, 会将SST整个写入文件系统, 并将一个文件描述对象作为SST对象的一个成员变量, 之后通过这个SST访问数据时, 是以Block为基础的IO单位进行访问的，这里的逻辑是：

  1. 初始化时, 先从SST文件中读取BlockMeta数据, 然后常驻内存
  2. 后续进行某个key的查找时, 从BlockMeta数据找出其对应的Block在文件中的偏移位置, 然后从文件中读取数据

因此核心的代码只有一个read_to_slice, 我们在之前的初始化代码中这样使用它:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
// 数据库初始化时打开文件
std::shared_ptr<SST> SST::open(size_t sst_id, FileObj file,
                               std::shared_ptr<BlockCache> block_cache) {
  ...

  // 读取文件末尾的元数据块
  // 读取元数据块的偏移量（最后8字节）
  size_t file_size = sst->file.size();

  auto offset_bytes =
      sst->file.read_to_slice(file_size - sizeof(uint32_t), sizeof(uint32_t));
  memcpy(&sst->meta_block_offset, offset_bytes.data(), sizeof(uint32_t));

  // 读取并解码元数据块 && 设置首尾key
  ...

  return sst;
}

在读取Block时这样使用:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
std::shared_ptr<Block> SST::read_block(size_t block_idx) {
  if (block_idx >= meta_entries.size()) {
    throw std::out_of_range("Block index out of range");
  }

  const auto &meta = meta_entries[block_idx];
  size_t block_size;

  // 计算block大小
  if (block_idx == meta_entries.size() - 1) {
    block_size = meta_block_offset - meta.offset;
  } else {
    block_size = meta_entries[block_idx + 1].offset - meta.offset;
  }

  // 读取block数据
  auto block_data = file.read_to_slice(meta.offset, block_size);
  
  // 其他逻辑
  ...

  return block_res;
}

这里的核心就是一个惰性读取, 因此只需要对标准库的文件读写进行一些简单封装即可。但另一方面，实现一个全新的项目时，Debug时设计文件数据的校验一定是非常困难的, 因此这里我除了实现了一个标准库的文件读写包装类之外, 我还实现了基于mmap的文件IO, 其将整个SST文件进行内存映射, 这样Debug时可以直接通过内存地址访问文件数据, 方便很多。并且, 尽管mmap提供了文件驻于内存中的抽象, 实际上其还是会由操作系统管理内存, 发现内存Page Fault时, 会从磁盘中读取数据到内存中, 本质上也是一个懒加载的实现。那为什么索性不直接使用mmap呢? 是因为我们后续要实现一个Block的缓存池, 我们想通过自己的方案来实现内存管理, 但实际上用mmap也没有什么大问题就是了。

2 代码实现

2.1 FileObj 文件封装类

由于我们需要同时支持标准文件库和mmap, 这里再定义一层封装:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
// include/utils/files.h
#pragma once

#include "mmap_file.h"
#include "std_file.h"
#include <cstddef>
#include <cstdint>
#include <memory>
#include <string>
#include <vector>

class FileObj {
private:
  std::unique_ptr<StdFile> m_file; // StdFile可以换为MmapFile
  size_t m_size;

public:
  FileObj();
  ~FileObj();

  // 禁用拷贝
  FileObj(const FileObj &) = delete;
  FileObj &operator=(const FileObj &) = delete;

  // 实现移动语义
  FileObj(FileObj &&other) noexcept;

  FileObj &operator=(FileObj &&other) noexcept;

  // 文件大小
  size_t size() const;

  // 设置文件大小
  void set_size(size_t size);

  // 创建文件对象, 并写入到磁盘
  static FileObj create_and_write(const std::string &path,
                                  std::vector<uint8_t> buf);

  // 打开文件对象
  static FileObj open(const std::string &path);

  // 读取并返回切片
  std::vector<uint8_t> read_to_slice(size_t offset, size_t length);
};

这里的FileObj类中, 我们使用了一个std::unique_ptr来指向一个StdFile或者MmapFile对象, 成员函数的操作就是基于这个指针指向的对象完成的。

这里照理说应该使用虚基类和多态, 但我偷懒了…

2.2 StdFile 标准文件IO

简单定义一下头文件:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
// include/utils/std_file.h
#pragma once

#include <cstddef>
#include <filesystem>
#include <fstream>
#include <string>
#include <vector>

class StdFile {

private:
  std::fstream file_;
  std::filesystem::path filename_;

public:
  StdFile() {}
  ~StdFile() {
    if (file_.is_open()) {
      close();
    }
  }

  // 打开文件并映射到内存
  bool open(const std::string &filename, bool create);

  // 创建文件
  bool create(const std::string &filename, std::vector<uint8_t> &buf);

  // 关闭文件
  void close();

  // 获取文件大小
  size_t size();

  // 写入数据
  bool write(size_t offset, const void *data, size_t size);

  // 读取数据
  std::vector<uint8_t> read(size_t offset, size_t length);

  // 同步到磁盘
  bool sync();
};

这里的代码都很简单, 没有运用什么高级手段, 就是对C++标准库的简单使用, 这里不详细展开了, 记得写入后sync就行。具体的实现为:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
#include "../../include/utils/std_file.h"

bool StdFile::open(const std::string &filename, bool create) {
  filename_ = filename;

  if (create) {
    file_.open(filename, std::ios::in | std::ios::out | std::ios::binary |
                             std::ios::trunc);
  } else {
    file_.open(filename, std::ios::in | std::ios::out | std::ios::binary);
  }

  return file_.is_open();
}

bool StdFile::create(const std::string &filename, std::vector<uint8_t> &buf) {
  if (!this->open(filename, true)) {
    throw std::runtime_error("Failed to open file for writing");
  }
  write(0, buf.data(), buf.size());
  return true;
}

void StdFile::close() {
  if (file_.is_open()) {
    sync();
    file_.close();
  }
}

size_t StdFile::size() {
  file_.seekg(0, std::ios::end);
  return file_.tellg();
}

std::vector<uint8_t> StdFile::read(size_t offset, size_t length) {
  std::vector<uint8_t> buf(length);
  file_.seekg(offset, std::ios::beg);
  if (!file_.read(reinterpret_cast<char *>(buf.data()), length)) {
    throw std::runtime_error("Failed to read from file");
  }
  return buf;
}

bool StdFile::write(size_t offset, const void *data, size_t size) {
  file_.seekg(offset, std::ios::beg);
  file_.write(static_cast<const char *>(data), size);
  this->sync();
  return true;
}

bool StdFile::sync() {
  if (!file_.is_open()) {
    return false;
  }
  file_.flush();
  return file_.good();
}

2.3 mmap文件IO

这里首先还是简单介绍下mmap

mmap（Memory-Mapped File I/O）是一种将文件或设备映射到进程的地址空间中的机制。通过mmap，文件的内容可以直接作为内存的一部分进行访问，而不需要使用传统的读写系统调用。这种方式简化了文件操作，并且在某些情况下可以显著提高性能。

mmap的工作原理

  • 内存映射：mmap将文件的部分或全部内容映射到进程的虚拟地址空间中，使得文件内容可以通过指针直接访问。
  • 按需加载：只有当程序实际访问映射区域时，操作系统才会从磁盘加载相应的页面到物理内存中。
  • 数据同步：对映射区域的修改会自动反映到文件中，反之亦然（取决于映射模式）。

我这里使用mmap的最大原因还是方便排查问题, 在调试过程中，可以直接通过内存地址访问文件内容, 这对于理解文件结构和数据流非常有帮助。并且mmap将内存管理直接交给操作系统完成, 如果你不想实现自己的缓存池的话, 直接用mmap就对了, 但介于我们项目的完整性, 后续还是实现了缓存池。

但mmap缺点也存在, 一方面, 将内存管理完全交给操作系统无法利用我们对其访问特性的利用(虽然我觉得自己写的缓存池还不如操作系统…), 另一方面就是mmap只能在类Unix系统上运行, 在Windows上是不支持的。至于打开的文件描述符数量过多倒不是问题，因为每个SST文件大小都不小, 如果文件描述符不够用了, 那数据库已经很庞大了, 已经不是我们这个玩具项目该做的事情了。

这里给出mmap文件IO类的实现:

头文件定义:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
#pragma once

#include <cstddef>
#include <cstdint>
#include <fcntl.h>
#include <string>
#include <sys/mman.h>
#include <unistd.h>
#include <vector>

class MmapFile {
private:
  int fd_;               // 文件描述符
  void *mapped_data_;    // 映射的内存地址
  size_t file_size_;     // 文件大小
  std::string filename_; // 文件名

  // 获取映射的内存指针
  void *data() const { return mapped_data_; }

  // 创建文件并映射到内存
  bool create_and_map(const std::string &path, size_t size);

public:
  MmapFile() : fd_(-1), mapped_data_(nullptr), file_size_(0) {}
  ~MmapFile() { close(); }

  // 打开文件并映射到内存
  bool open(const std::string &filename, bool create = false);

  // 创建文件
  bool create(const std::string &filename, std::vector<uint8_t> &buf);

  // 关闭文件
  void close();

  // 获取文件大小
  size_t size() const { return file_size_; }

  // 写入数据
  bool write(size_t offset, const void *data, size_t size);

  // 读取数据
  std::vector<uint8_t> read(size_t offset, size_t length);

  // 同步到磁盘
  bool sync();

private:
  // 禁止拷贝
  MmapFile(const MmapFile &) = delete;
  MmapFile &operator=(const MmapFile &) = delete;
};

实现:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
#include "../../include/utils/mmap_file.h"
#include <cstdint>
#include <errno.h>
#include <stdexcept>
#include <string.h>
#include <sys/stat.h>
#include <vector>

bool MmapFile::open(const std::string &filename, bool create) {
  filename_ = filename;

  // 打开或创建文件
  int flags = O_RDWR;
  if (create) {
    flags |= O_CREAT;
  }

  fd_ = ::open(filename.c_str(), flags, 0644);
  if (fd_ == -1) {
    return false;
  }

  // 获取文件大小
  struct stat st;
  if (fstat(fd_, &st) == -1) {
    close();
    return false;
  }
  file_size_ = st.st_size;

  // 映射文件
  if (file_size_ > 0) {
    mapped_data_ =
        mmap(nullptr, file_size_, PROT_READ | PROT_WRITE, MAP_SHARED, fd_, 0);
    if (mapped_data_ == MAP_FAILED) {
      close();
      return false;
    }
  }

  return true;
}

bool MmapFile::create(const std::string &filename, std::vector<uint8_t> &buf) {
  // 创建文件，设置大小并映射到内存
  if (!create_and_map(filename, buf.size())) {
    // throw std::runtime_error("Failed to create or wrte file: " + filename);
    return false;
  }

  // 写入数据
  memcpy(this->data(), buf.data(), buf.size());
  this->sync();
  return true;
}
void MmapFile::close() {
  if (mapped_data_ != nullptr && mapped_data_ != MAP_FAILED) {
    munmap(mapped_data_, file_size_);
    mapped_data_ = nullptr;
  }

  if (fd_ != -1) {
    ::close(fd_);
    fd_ = -1;
  }

  file_size_ = 0;
}

bool MmapFile::write(size_t offset, const void *data, size_t size) {
  // 调整文件大小以包含 offset + size
  size_t new_size = offset + size;
  if (ftruncate(fd_, new_size) == -1) {
    return false;
  }

  // 如果已经映射，先解除映射
  if (mapped_data_ != nullptr && mapped_data_ != MAP_FAILED) {
    munmap(mapped_data_, file_size_);
  }

  // 重新映射
  file_size_ = new_size;
  mapped_data_ =
      mmap(nullptr, new_size, PROT_READ | PROT_WRITE, MAP_SHARED, fd_, 0);
  if (mapped_data_ == MAP_FAILED) {
    mapped_data_ = nullptr;
    return false;
  }

  // 写入数据
  memcpy(static_cast<uint8_t *>(mapped_data_) + offset, data, size);
  this->sync();
  return true;
}

std::vector<uint8_t> MmapFile::read(size_t offset, size_t length) {
  // 从映射的内存中复制数据
  // 创建结果vector
  // 创建结果vector
  std::vector<uint8_t> result(length);

  // 从映射的内存中复制数据
  const uint8_t *data = static_cast<const uint8_t *>(this->data());
  memcpy(result.data(), data + offset, length);

  return result;
}

bool MmapFile::sync() {
  if (mapped_data_ != nullptr && mapped_data_ != MAP_FAILED) {
    return msync(mapped_data_, file_size_, MS_SYNC) == 0;
  }
  return true;
}

// ********************* private *********************

bool MmapFile::create_and_map(const std::string &path, size_t size) {
  // 创建并打开文件
  fd_ = ::open(path.c_str(), O_RDWR | O_CREAT | O_TRUNC, 0644);
  if (fd_ == -1) {
    return false;
  }

  // 先调整文件大小
  if (ftruncate(fd_, size) == -1) {
    close();
    return false;
  }

  // 映射与文件大小相同的空间
  mapped_data_ =
      mmap(nullptr, size, PROT_READ | PROT_WRITE, MAP_SHARED, fd_, 0);
  if (mapped_data_ == MAP_FAILED) {
    close();
    return false;
  }

  file_size_ = size;
  return true;
}

这里的data()函数直接返回内存映射的指针, 有了这个指针后, 可以直接通过内存访问完成文件读写 是不是很方便:

1
2
3
4
5
6
7
8
9
10
11
12
std::vector<uint8_t> MmapFile::read(size_t offset, size_t length) {
  // 从映射的内存中复制数据
  // 创建结果vector
  // 创建结果vector
  std::vector<uint8_t> result(length);

  // 从映射的内存中复制数据
  const uint8_t *data = static_cast<const uint8_t *>(this->data());
  memcpy(result.data(), data + offset, length);

  return result;
}

3 调试工具分享

实现这个文件类之后, 已经可以将整个存储引擎串联起来了。吗，目前写到这里，最痛苦的就是文件编码调试， 二进制的数据编解码真的很头疼，这里推荐一个二进制文件查看工具xxd

安装

1
sudo apt install xxd

查看二进制文件

1
2
3
4
5
6
(base) ➜  bin git:(master) ✗ xxd example_data/sst_0000
00000000: 0400 6b65 7931 0a00 6e65 775f 7661 6c75  ..key1..new_valu
00000010: 6531 0400 6b65 7932 0000 0400 6b65 7933  e1..key2....key3
00000020: 0600 7661 6c75 6533 0000 1200 1a00 0300  ..value3........
00000030: 1e7a 8b4f 0100 0000 0000 0000 0400 6b65  .z.O..........ke
00000040: 7931 0400 6b65 7933 7419 9439 3400 0000  y1..key3t..94...