LevelDB project 1 10225501460 林子骥 10211900416 郭夏辉
Nie możesz wybrać więcej, niż 25 tematów Tematy muszą się zaczynać od litery lub cyfry, mogą zawierać myślniki ('-') i mogą mieć do 35 znaków.

435 wiersze
21 KiB

2 tygodni temu
  1. # 实验报告:在LevelDB中实现TTL功能
  2. 10225501460 林子骥
  3. 10211900416 郭夏辉
  4. ## 1.实验目的
  5. - 深入了解LevelDB的内部原理和数据结构。
  6. - 掌握TTL(Time To Live,生存时间)功能的设计与实现方法。
  7. - 学习如何在开源项目中添加新功能,提升代码阅读和修改能力。
  8. ## 2.实验要求
  9. - 在LevelDB中实现键值对的TTL功能,使得过期的数据在**读取时自动失效**,并在适当的时候**被合并清理**。
  10. - 修改LevelDB的源码,实现对TTL的支持,包括数据的写入、读取和过期数据的清理。
  11. - 编写测试用例,验证TTL功能的正确性和稳定性。
  12. ## 3.实验背景
  13. **TTL(Time To Live)**,即生存时间,是指数据在存储系统中的有效期。设置TTL可以使得过期的数据自动失效,减少存储空间占用,提高系统性能。**为什么需要TTL功能:**
  14. - **数据自动过期**:无需手动删除过期数据,简化数据管理。
  15. - **节省存储空间**:定期清理无效数据,优化资源利用。
  16. - **提高性能**:减少无效数据的干扰,提升读写效率。
  17. **在LevelDB中添加TTL功能的方案:**
  18. 1. **数据编码方式修改**:在键或值中增加过期时间的信息。
  19. 2. **读取时判断过期**:在Get操作时,检查数据是否过期,过期则返回NotFound。
  20. 3. **Compaction清理**:在数据压缩过程中,删除过期的数据。
  21. ## 4.实验步骤
  22. ### 4.1 数据格式调整及支持TTL的Put方法
  23. 原本的数据样貌大致是[key, value],为了方便进一步的操作,在样貌不变的前提下,我们直接将过期时间放在了实际数据后面,一起存储到 value 中。
  24. ```cpp
  25. Status DBImpl::Put(const WriteOptions& options, const Slice& key,
  26. const Slice& value, uint64_t ttl) {
  27. std::string val_with_ts;
  28. val_with_ts.reserve(value.size() + kTSLength);
  29. char ts_string[kTSLength];
  30. TIMESTAMP expiration_time = this->env_->NowMicros() + ttl * 1000000;
  31. EncodeFixed64(ts_string,expiration_time);
  32. //assert(sizeof(expiration_time) == sizeof(TIMESTAMP ));
  33. // 追加原始 value 到 val_with_ts
  34. val_with_ts.append(value.data(), value.size());
  35. // 将 expiration_time 追加到 val_with_ts
  36. val_with_ts.append(ts_string,kTSLength);
  37. return DB::Put(options, key, Slice(val_with_ts));
  38. }
  39. ```
  40. 最开始,我们拓展了`DBImpl::Put`使得它支持TTL。从前到后的逻辑很简单:
  41. 1. 创建一个空的 `std::string val_with_ts`,存储要写入的带过期时间的value,并且为其预留足够的空间(value和时间戳的长度之和)
  42. 2. 利用`env_`提供的`NowMicros()`方法,以微秒为单位计算过期时间(`当前时间+TTL*1000000`)这里我们尽量将所有底层操作交给`env_`完成,而DBImpl有该对象可以调用相关查找时间的函数
  43. 3. 利用`EncodeFixed64`编码函数将`expiration_time`编码为一个固定长度的字符串,并存储在`ts_string`中。利用LevelDB自带的而不是自己手写的编码函数可以提升一些运行效率。
  44. 4. 追加原始`value`到`val_with_ts`以及将`expiration_time` 追加到 `val_with_ts`只需要使用字符串的`append`方法从前往后拼接即可。
  45. 5. 最后将刚刚处理得到的`val_with_ts`打包成一个`Slice`,然后再调用原始的`DB::Put`方法。
  46. 为了适配之后的实验,提升兼容性,我们将上面这套逻辑重复了一遍,也完成了新的`Status DB::Put(const WriteOptions& options, const Slice& key, const Slice& value, uint64_t ttl) `,使得它支持TTL。
  47. ```cpp
  48. Status DB::Put(const WriteOptions& options, const Slice& key,
  49. const Slice& value, uint64_t ttl) {
  50. // 将 value 和 expiration_time 合并到一起,形成带 TTL 的 value
  51. std::string val_with_ts;
  52. val_with_ts.reserve(value.size() + sizeof(uint64_t));
  53. uint64_t expiration_time = std::chrono::duration_cast<std::chrono::milliseconds>(
  54. std::chrono::system_clock::now().time_since_epoch())
  55. .count() + ttl * 1000;
  56. // 追加原始 value 到 val_with_ts
  57. val_with_ts.append(value.data(), value.size());
  58. // 将 expiration_time 追加到 val_with_ts
  59. val_with_ts.append(reinterpret_cast<const char*>(&expiration_time), sizeof(expiration_time));
  60. WriteBatch batch;
  61. batch.Put(key, Slice(val_with_ts));
  62. return Write(options, &batch);
  63. }
  64. ```
  65. 只不过在这里计算当前时间不能再使用`env_`的相关方法了,使用的是`std::chrono::system_clock::now()`获取当前时间,其他的逻辑是一样的。
  66. 实现很简单,就是将TTL值转为到期的时间戳,加入value尾部,这样可以尽可能降低整个项目的耦合程度,我们无需对ExtractUserkey(提取具体key值相关的函数),FindShortSeperator(简化key长度,这里还可能导致时间戳被修改)等函数进行复杂的重写。此外,还可以节约空间。
  67. 这里基于原先的项目举几个例子来说明为什么不讲时间戳作为key的一部分写入,原先的index_block就是为了尽量缩短key的长度,而如果将时间戳加入key的尾部,其时间戳本身的长度就会大于key的长度。并且由于其内部的时间戳与key不同,是无序存储的,因此很难使用重启点来减少存储时间戳的空间开销。
  68. ![image-20241107080532982](assets/image-20241107080532982.png)
  69. ### 4.2支持TTL的Get方法
  70. Get操作的核心便是检查数据是否过期,只要实现这里就能让读取顺利进行了。
  71. ```cpp
  72. uint64_t DBImpl::GetTS(std::string* val) {
  73. auto expiration_time = DecodeFixed64(val->data() + val->size() - kTSLength);
  74. val->resize(val->size() - kTSLength);//保证返回值没有ttl
  75. return expiration_time;
  76. }
  77. Status DBImpl::CheckIsExpire(std::string* value) {
  78. //debug 用
  79. auto a = env_->NowMicros();
  80. auto b = GetTS(value);
  81. if(a > b){
  82. return Status::Expire("Expire",Slice());
  83. }
  84. return Status();
  85. }
  86. Status DBImpl::Get(const ReadOptions& options, const Slice& key,
  87. std::string* value) {
  88. // other code...
  89. // Unlock while reading from files and memtables
  90. // Unlock while reading from files and memtables
  91. {
  92. mutex_.Unlock();
  93. // First look in the memtable, then in the immutable memtable (if any).
  94. LookupKey lkey(key, snapshot);
  95. if (mem->Get(lkey, value, &s)) {
  96. // Done
  97. } else if (imm != nullptr && imm->Get(lkey, value, &s)) {
  98. // Done
  99. } else {
  100. stats.now_ts = this->env_->NowMicros();
  101. s = current->Get(options, lkey, value, &stats);
  102. have_stat_update = true;
  103. }
  104. mutex_.Lock();
  105. }
  106. if(s.ok()){
  107. s = CheckIsExpire(value);
  108. }
  109. if (have_stat_update && current->UpdateStats(stats,s.IsExpire())) {
  110. MaybeScheduleCompaction();//有可能寻址过多,导致allow_seek为0,触发合并。
  111. }
  112. //......
  113. }
  114. bool Version::UpdateStats(const GetStats& stats,bool is_expire) {
  115. FileMetaData* f = stats.seek_file;
  116. if (f != nullptr) {
  117. f->allowed_seeks--;
  118. if(is_expire)f->allowed_seeks--;
  119. if (f->allowed_seeks <= 0 && file_to_compact_ == nullptr) {
  120. file_to_compact_ = f;
  121. file_to_compact_level_ = stats.seek_file_level;
  122. return true;
  123. }
  124. }
  125. return false;
  126. }
  127. ```
  128. 与编码时候的`EncodeFixed64`相对应,我们使用`DecodeFixed64`将过期时间戳从value里面提取出来,并实现了`DBImpl::GetTS`方法。在此基础上我们检查value是否应该被丢弃了,实现了`DBImpl::CheckIsExpire`方法,如果当前时间已经超过了时间戳,那么直接返回过期状态`Status::Expire`。(其实这里也纠结了一下要不要按照实验文档中要求的那样返回状态`Status::NotFound`,经过测试两者都是可以的,然后严谨一些,实际情况是key已经存在,但由于某种原因不可再访问,所以最终还是使用了`Status::Expire`)
  129. 在此我们进行一些时间分析,对于涉及内存(memtable内,缓存命中)的查找速度很快;而这里可能会想到,有许多过期数据都在缓存池内要怎么办,实际上,只要是在缓存池内的数据,即使过期了,它也是有信息价值的(提供了过期这一信息)。因此,我们实际需要解决的是,如何降低IO层面的对于过期数据的查找。这里我们在基于leveldb采用的自动合并优化上(即一个文件访问多次后,它将会作为input传给compaction,然后后续过滤过期数据),对于在缓存池中,读取到的所过期的数据所在的文件,其`allowed_seek`次数再次减一(也可以基于需求减去其他值),这样的目的是,期望尽快对于其key值所在的文件触发合并,减少底层过期的数据量。
  130. 当然,这样做对于此次的查找帮助不大,主要是为了以后再次查找时,能加快扫描速度。接下里我们可能会想,如何减少cache内减少IO操作的次数。先回忆整个cache的读取流程:获得布隆过滤器,若key值存在,从硬盘中读取IndexBlock,然后再从硬盘读取DataBlock。其中IndexBlock的设计在之前已经提到,为了节约空间开销,不能添加时间戳。因此我们只能依靠于bloomfilter类似的功能。这里我们以manifest的元数据作为过滤的基础。
  131. 我们为每个文件的元数据增加新字段
  132. ```cpp
  133. struct FileMetaData {
  134. FileMetaData() : refs(0), allowed_seeks(1 << 30), file_size(0),oldest_ts(UINT64_MAX),newer_ts(0) {}
  135. int refs;
  136. int allowed_seeks; // Seeks allowed until compaction
  137. uint64_t number;
  138. uint64_t file_size; // File size in bytes
  139. InternalKey smallest; // Smallest internal key served by table
  140. InternalKey largest; // Largest internal key served by table
  141. TIMESTAMP oldest_ts; //新字段:该文件内最小的时间戳,
  142. TIMESTAMP newer_ts; //新字段:该文件内最大的时间戳
  143. };
  144. ```
  145. 接下来就是对相关的MANIFEST文件的元数据写入进行修改以及在写入key,value对时,对时间戳更新。包括`VersionEdit::AddFile`,`VersionEdit::EncodeTo`Status `VersionEdit::DecodeFrom`,`builder.h:Status BuildTable`这里不在报告中举出。然后最后的过滤操作,则在`Status Version::Get`内的`match`函数
  146. ```cpp
  147. //1.调用Status Version::Get(const ReadOptions& options, const LookupKey& k,
  148. // std::string* value, GetStats* stats)接口,进入非memtable部分查找
  149. //2调用void Version::ForEachOverlapping(Slice user_key, Slice internal_key, void* arg,
  150. // bool (*func)(void*, int, FileMetaData*))查找涉及到的key以及对应的文件元数据
  151. //3.调用match函数,寻址
  152. static bool Match(void* arg, int level, FileMetaData* f) {
  153. State* state = reinterpret_cast<State*>(arg);
  154. if (state->stats->seek_file == nullptr && state->last_file_read != nullptr) {
  155. // We have had more than one seek for this read. Charge the 1st file.
  156. state->stats->seek_file = state->last_file_read;
  157. state->stats->seek_file_level = state->last_file_read_level;
  158. }
  159. state->last_file_read = f;
  160. state->last_file_read_level = level;
  161. if(state->stats->now_ts > f->newer_ts)return false;//在访问缓存池前,判断是否过期
  162. state->s = state->vset->table_cache_->Get(*state->options, f->number, f->file_size, state->ikey,&state->saver, SaveValue);
  163. // other code...
  164. }
  165. ```
  166. 这里有尝试`if(state->stats->now_ts > f->newer_ts)return false;`根据文件内的最新时间戳来进行过滤。但是实际上,对于多次的批量插入以及一个batch内不同的TTL,在合并后,我们无法很好的预测出一个文件的新旧情况,即每一个文件的元数据的new_ts有可能都十分大,最终还是要进入cache内查找。
  167. **经过刚刚的修改,我们初步完成了读取功能,第一个测试点经过一些修改也能顺利通过了。**
  168. 在这时曾发现ttl_test启动后,无论如何修改ReadTTL都显示 failed,经过助教提醒发现是随机种子的问题,如果采用`static_cast<unsigned int>(time(0))`根据当前的时间戳生成随机数,那么`InsertData`插入数据的 key 和查找时所使用的 key 不一致导致读取失败。解决的办法就是`srand(0)`固定随机种子。
  169. ReadTTL之后,CompactionTTL还未开始,直接显示 open DB failed,出现这个原因是ReadTTL之后打开的数据库没有清除,只需要在ReadTTL的末尾加上`delete db;`就解决了问题。
  170. 还有就是`delete db;`之后需要睡眠一秒,否则还是会出现read DB failed。这里的另外一个问题就是ReadTTL测试用例在我的电脑上一直出现open db fail,但是在另一位队友的电脑上就不会,这是因为在我这里程序跑的太慢了,超过它所设定的20s,这里酌情提高之后就能解决了。
  171. ### 4.3支持TTL的Compaction
  172. 在数据Compaction过程中,需要删除过期的数据。大合并操作都调用`DoCompactionWork`进行,结合之前的授课我们可以知道它本身就会检查数据是否有效,因此我们只需要在此多加入一些关于TTL判断的逻辑就行。比较巧的是,`DoCompactionWork`自带drop变量来标记数据是否需要被丢弃,我们只需要利用好它就行。
  173. ```cpp
  174. Status DBImpl::DoCompactionWork(CompactionState* compact) {
  175. // other code...
  176. while (input->Valid() && !shutting_down_.load(std::memory_order_acquire)) {
  177. // other code...
  178. if (!ParseInternalKey(key, &ikey)) {
  179. // other code...
  180. } else {
  181. if (!has_current_user_key ||
  182. user_comparator()->Compare(ikey.user_key, Slice(current_user_key)) !=
  183. 0) {
  184. // other code...
  185. }
  186. if (last_sequence_for_key <= compact->smallest_snapshot) {
  187. // other code...
  188. } else if (ikey.type == kTypeDeletion &&
  189. ikey.sequence <= compact->smallest_snapshot &&
  190. compact->compaction->IsBaseLevelForKey(ikey.user_key)) {
  191. // other code...
  192. }else if((ts = DecodeFixed64(input->value().data() + input->value().size() - kTSLength)) < env_->NowMicros()){
  193. // TTL
  194. drop = true;
  195. }
  196. last_sequence_for_key = ikey.sequence;
  197. }
  198. // other code...
  199. return status;
  200. }
  201. ```
  202. 其实判断的逻辑和之前`Get`和`Put`时候是一样的,如果当前时间超出了过期时间,则将 drop 标记为 true,表示丢弃该数据;若当前时间尚未超出过期时间则将数据视为有效的。
  203. 这里有一个小问题是TEST中无法完全删除所有过期数据。具体的例子如下,未触发手动合并前:level0文件数量:31个;level1数量:1;level2数量:1个;然后手动合并后,它会先尝试合并level0数据,然后在合并的过程中,它会将level0和level1的所有数据都当做input,放入一个mergeIterator。因为时间过期了,它会将这些文件全删掉,只剩下level2的一个文件。然后因为此时我们最大的遍历层只能到level1,level2的数据删除依赖于level1在合并时,其key的范围覆盖到level2,然后删除。此外,即使level1还剩下一个文件,也需要其所有。对于这个问题的解决,我们最开始是修改`kL0_CompactionTrigger `等参数,加大level0的文件数量,使得所有数据都放在level0中,这样可以确保所有数据都在compaction中作为input,但这会导致在今后正常运行时(非测试情况),其IO开销较大(因为level0层会覆盖很多key)。
  204. 在这里经过助教提示发现了一个小问题,就是手动合并可能无法保证合并所有数据,导致无法完全丢弃过期数据。解决方法就是修改 CompactRange 的流程使得所有数据得到合并。原始代码并没有去合并最大的含有文件的level(max_level_with_files)的文件,只合并了在此之前的level,我们的修改是这样的:
  205. ```cpp
  206. void DBImpl::CompactRange(const Slice* begin, const Slice* end) {
  207. // other code...
  208. TEST_CompactMemTable(); // TODO(sanjay): Skip if memtable does not overlap
  209. for (int level = 0; level < max_level_with_files; level++) {
  210. TEST_CompactRange(level, begin, end);
  211. }
  212. TEST_CompactRange(max_level_with_files, begin, end);
  213. }
  214. ```
  215. ## 5.测试和结果
  216. ### 5.1测试用例
  217. ```cpp
  218. #include "gtest/gtest.h"
  219. #include "leveldb/env.h"
  220. #include "leveldb/db.h"
  221. using namespace leveldb;
  222. constexpr int value_size = 2048;
  223. constexpr int data_size = 128 << 20;
  224. Status OpenDB(std::string dbName, DB **db) {
  225. Options options;
  226. options.create_if_missing = true;
  227. return DB::Open(options, dbName, db);
  228. }
  229. void InsertData(DB *db, uint64_t ttl/* second */) {
  230. WriteOptions writeOptions;
  231. int key_num = data_size / value_size;
  232. srand(0);
  233. for (int i = 0; i < key_num; i++) {
  234. int key_ = rand() % key_num+1;
  235. std::string key = std::to_string(key_);
  236. std::string value(value_size, 'a');
  237. db->Put(writeOptions, key, value, ttl);
  238. }
  239. }
  240. void GetData(DB *db, int size = (1 << 30)) {
  241. ReadOptions readOptions;
  242. int key_num = data_size / value_size;
  243. // 点查
  244. srand(0);
  245. for (int i = 0; i < 100; i++) {
  246. int key_ = rand() % key_num+1;
  247. std::string key = std::to_string(key_);
  248. std::string value;
  249. db->Get(readOptions, key, &value);
  250. }
  251. }
  252. //
  253. TEST(TestTTL, ReadTTL) {
  254. DB *db;
  255. if(OpenDB("testdb", &db).ok() == false) {
  256. std::cerr << "open db failed" << std::endl;
  257. abort();
  258. }
  259. // 如果出现open db fail,请酌情提高这里
  260. uint64_t ttl = 20;
  261. InsertData(db, ttl);
  262. ReadOptions readOptions;
  263. Status status;
  264. int key_num = data_size / value_size;
  265. srand(0);
  266. for (int i = 0; i < 100; i++) {
  267. int key_ = rand() % key_num+1;
  268. std::string key = std::to_string(key_);
  269. std::string value;
  270. status = db->Get(readOptions, key, &value);
  271. ASSERT_TRUE(status.ok());
  272. }
  273. Env::Default()->SleepForMicroseconds(ttl * 1000000);
  274. for (int i = 0; i < 100; i++) {
  275. int key_ = rand() % key_num+1;
  276. std::string key = std::to_string(key_);
  277. std::string value;
  278. status = db->Get(readOptions, key, &value);
  279. ASSERT_FALSE(status.ok());
  280. }
  281. delete db;
  282. Env::Default()->SleepForMicroseconds( 1000);
  283. }
  284. TEST(TestTTL, CompactionTTL) {
  285. DestroyDB("testdb", Options());
  286. DB *db;
  287. if(OpenDB("testdb", &db).ok() == false) {
  288. std::cerr << "open db failed" << std::endl;
  289. abort();
  290. }
  291. uint64_t ttl = 20;
  292. InsertData(db, ttl);
  293. leveldb::Range ranges[1];
  294. ranges[0] = leveldb::Range("-", "A");
  295. uint64_t sizes[1];
  296. db->GetApproximateSizes(ranges, 1, sizes);
  297. ASSERT_GT(sizes[0], 0);
  298. Env::Default()->SleepForMicroseconds(ttl * 1000000);
  299. db->CompactRange(nullptr, nullptr);
  300. leveldb::Range ranges2[1];
  301. ranges2[0] = leveldb::Range("-", "A");
  302. uint64_t sizes2[1];
  303. db->GetApproximateSizes(ranges2, 1, sizes2);
  304. ASSERT_EQ(sizes2[0], 0);
  305. }
  306. int main(int argc, char** argv) {
  307. // All tests currently run with the same read-only file limits.
  308. testing::InitGoogleTest(&argc, argv);
  309. return RUN_ALL_TESTS();
  310. }
  311. ```
  312. ### 5.2实验结果
  313. 在我的电脑上,我适当调大了ReadTTL中TTL之后(调整到了50),结果如下所示:
  314. ![image-20241107082315854](assets/image-20241107082315854.png)
  315. ## 6.本实验ttl实现与已有ttl实现的比较
  316. **RocksDB**
  317. 本身支持一种数据过期淘汰方案,该方案是通过特定的 API 打开 DB,对写入该 DB 的全部 key 都遵循一个 TTL 过期策略,例如 TTL 为 3 天,那么写入该 DB 的 key 都会在写入的三天后自动过期。但是该方案不灵活,无法针对每一条 key 设置 TTL。
  318. 此外,rocksdb该方案底层也是通过 compaction filter 实现的,也就是说过期数据虽然对用户不可见,但是磁盘空间并不会及时回收,根本原因就是数据堆积在某一层,而该层没有触发 compaction,那么我们可以手动调用 RocksDB 的 CompactionRange 函数,来触发 compaction filter,达到快速回收磁盘空间的目的。但是主动调用 CompactionRange 会导致 RocksDB 自身的 compaction 暂停,这会触发 Write Stall,造成非常严重的后果。
  319. **Parker**:
  320. 采用`Periodic compaction + dynamic compaction`。Periodic compaction 的主要原理是增加一个 periodic_compaction_seconds 参数,并记录每个SST 文件的创建时间,每隔 periodic_compaction_seconds 秒,主动对这个 SST 文件进行 compaction 操作,从而回收沉底的 SST 文件;而 dynamic compaction 则是通过设置level_compaction_dynamic_level_bytes 为 true,进行动态合并,而不是按 level 的顺序合并到下一层,这使得 compaction 更加频繁。
  321. **业内有赞 KV :**
  322. 具体做法是,保持现有存储列族不变,另外增加一个列族,专门存储 key 的 TTL,然后通过一个 goroutine 根据当前时间戳,按照过期数据删除策略可以是定时触发,例如凌晨1点,或者每个一段时间触发等。该列族中 key 的编码规则如下:
  323. - key:时间戳 + key 的类型 + key 值
  324. - value:存储一个字节,代表不同数据类型
  325. 这个方案优点是回收速度快且回收时间可控,但是缺点就是实现复杂,极端情况下会降低 50% 的TPS。
  326. **本实验的ttl**
  327. 在数据的 value 的尾部添加八个字节的 TTL,实现灵活。
  328. ![屏幕截图 2024-11-07 110211](./assets/key_value_pair.jpg)
  329. 基于ttl过期的case,显著降低allowed_seek次数,使得对于指定的过期文件,能够加快合并速度。
  330. ## 7.实验总结和思考
  331. 通过本次实验,我们对LevelDB有了更深入的理解。超脱于之前的理论讲解,我们透过代码掌握了各个部分是如何实现的,也对整体样貌再次有了一个概览,为日后的数据管理系统的学习打下了坚实基础。