Prince's Blog

0%

Day6: Write Path

Posted on Edited on In
Symbols count in article: 1.2k Reading time ≈ 4 mins.

Week1 Day6 的任务：

  1. 实现有 L0 flush 的 LSM 写路径
  2. 实现更新 LSM 状态的正确逻辑

week1 day6

Tasks

Task 1: Flush Memtable to SST

现在需要进一步实现的是实现从内存写入到磁盘的逻辑（称为 flush）。

首先是 memtable flush 进 SST。只需要将 memtable 中的数据利用 SST 的 builder 写到 SST 中即可。

1
2
3
4
5
6
pub fn flush(&self, _builder: &mut SsTableBuilder) -> Result<()> {
    for entry in self.map.iter() {
        _builder.add(KeySlice::from_slice(&entry.key()[..]), &entry.value()[..]);
    }
    Ok(())
}

然后是将最早创建的（最旧的）不可变的 memtable flush 的逻辑。首先获取状态锁，确保只有一个线程对状态进行修改。然后获取最早的 imm_memtable 的克隆（记得获取读锁并释放，这里采用生命周期的方式自动释放）。然后调用其 flush 方法进行 flush。最后获取写锁，将新的状态写入（同样利用生命周期自动释放写锁）。

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
/// Force flush the earliest-created immutable memtable to disk
pub fn force_flush_next_imm_memtable(&self) -> Result<()> {
    let state_lock = self.state_lock.lock();

    let flush_memtable;
    {
        let snapshot = self.state.read();
        flush_memtable = snapshot
            .imm_memtables
            .last()
            .expect("no memtable to flush")
            .clone();
    }

    let mut builder = SsTableBuilder::new(self.options.block_size);
    flush_memtable.flush(&mut builder)?;

    let sstable_id = flush_memtable.id();
    let sstable = Arc::new(builder.build(
        sstable_id,
        Some(self.block_cache.clone()),
        self.path_of_sst(sstable_id),
    )?);

    // add flushed l0 table to the list
    {
        let mut guard = self.state.write();
        let mut snapshot = guard.as_ref().clone();
        // remove the last imm memtable
        let mem = snapshot.imm_memtables.pop().unwrap();
        debug_assert_eq!(mem.id(), sstable_id);
        snapshot.l0_sstables.insert(0, sstable_id);
        snapshot.sstables.insert(sstable_id, sstable);
        *guard = Arc::new(snapshot);
    }

    Ok(())
}

Task 2: Flush Trigger

当可变与不可变的 memtable 到达数量限制的时候，就需要将最早的不可变 memtable flush 到磁盘。直接获取当前状态快照，判断不可变 memtable 的长度是否达到限制，如果是就触发 flush。

1
2
3
4
5
6
7
8
fn trigger_flush(&self) -> Result<()> {
    let snapshot = self.state.read().clone();
    if snapshot.imm_memtables.len() >= self.options.num_memtable_limit {
        self.force_flush_next_imm_memtable()?;
    }

    Ok(())
}

当关闭 MiniLSM 的时候，要等待 flush 线程结束，并且将可变 memtable 转为不可变，将所有的不可变的 memtable flush 到磁盘。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
pub fn close(&self) -> Result<()> {
    let mut flush_thread = self.flush_thread.lock();
    if let Some(flush_thread) = flush_thread.take() {
        flush_thread
            .join()
            .map_err(|e| anyhow::anyhow!("{:?}", e))?;
    };

    let snapshot = &self.inner.state.read().clone();
    if !snapshot.memtable.is_empty() {
        self.inner
            .freeze_memtable(Arc::new(MemTable::create(self.inner.next_sst_id())))?;
    }

    while !snapshot.imm_memtables.is_empty() {
        self.inner.force_flush_next_imm_memtable()?;
    }

    Ok(())
}

Task 3: Filter the SSTs

目前已经拥有了一个完整的可以工作的存储引擎。最后实现一个小优化，将不包含 key 范围的 SST 剔除不搜索。之前实现了 scan 时判断 key 是否在 SST 范围内的函数 range_bound，现在实现 get 时判断 key 是否在 SST 范围内的函数 key_within。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
fn key_within(user_key: &[u8], table_begin: KeySlice, table_end: KeySlice) -> bool {
    table_begin.raw_ref() <= user_key && user_key <= table_end.raw_ref()
}

pub fn get(&self, _key: &[u8]) -> Result<Option<Bytes>> {
    ...
    for table_id in state.l0_sstables.iter() {
        let table = state.sstables[table_id].clone();
        if Self::key_within(
            _key,
            table.first_key().as_key_slice(),
            table.last_key().as_key_slice(),
        ) {
            l0_iters.push(Box::new(SsTableIterator::create_and_seek_to_key(
                table,
                KeySlice::from_slice(_key),
            )?));
        }
    }
    ...
}

Result

result

Test Your Understanding

  1. What happens if a user requests to delete a key twice?
    不会发生什么特殊的事情，因为删除操作是通过插入值为空的键值对实现的，每次删除就插入了一个值为空的键值对，每次获取最新的键值对都是空，也就是被删除。

  2. How much memory (or number of blocks) will be loaded into memory at the same time when the iterator is initialized?
    在迭代器初始化的时候，在查询范围内的 SST 会被加载到内存中。

  3. Some crazy users want to fork their LSM tree. They want to start the engine to ingest some data, and then fork it, so that they get two identical dataset and then operate on them separately. An easy but not efficient way to implement is to simply copy all SSTs and the in-memory structures to a new directory and start the engine. However, note that we never modify the on-disk files, and we can actually reuse the SST files from the parent engine. How do you think you can implement this fork functionality efficiently without copying data? (Check out Neon Branching).
    核心是利用 SST 的不可变性和懒加载的特性。在 fork 的时候上锁，创建子引擎目录，将内存中的 memtable 深拷贝到子引擎中；而对于 SST 文件，可以通过硬链接的方式指向父引擎的 SST 文件，这样就不需要拷贝数据，只需要拷贝元数据。修改的时候直接创建新的 SST 文件。

  4. Imagine you are building a multi-tenant LSM system where you host 10k databases on a single 128GB memory machine. The memtable size limit is set to 256MB. How much memory for memtable do you need for this setup?
    每个数据库至少有一个可变的 memtable，也就是 \(10k \times 256MB = 2560 GB = 2.5TB\)。

    • Obviously, you don’t have enough memory for all these memtables. Assume each user still has their own memtable, how can you design the memtable flush policy to make it work? Does it make sense to make all these users share the same memtable (i.e., by encoding a tenant ID as the key prefix)?
    • 可以设计动态的调度策略（例如 LRU），将一段时间未使用的 memtable 刷出内存，空出内存给新的 memtable。
    • 将 tenant ID 作为 key 的前缀，这样不同 tenant 的 key 不会冲突，可以共享同一个 memtable。但是可能导致用户隔离性下降。