Lower Layers (3): RocksDB Inside BlueStore

25-1 Introduction: why RocksDB?#

BlueStore writes object data directly to the raw device, but all of its metadata, from onodes and omaps to the freelist and deferred writes, lives in RocksDB. RocksDB is a key-value database based on an LSM-tree: writes are first recorded in a sequential log (the WAL) and an in-memory structure (the memtable), and later, in the background, turned into sorted, immutable files (SSTs) and merged. This structure makes small writes cheap and sequential, which is ideal for BlueStore's frequently changing metadata. The cost is paid in compaction and write amplification. This chapter shows layer by layer how Ceph configures and shards RocksDB and places it on BlueFS.

25-2 From a BlueStore transaction to a WriteBatch#

BlueStore::_txc_finalize_kv and _kv_sync_thread gather ready transactions. Each KeyValueDB::Transaction in RocksDBStore is a rocksdb::WriteBatch, and submit_transaction_sync hands it to RocksDB with WriteOptions.sync = true. Several BlueStore transactions are submitted back to back in one round of _kv_sync_thread, but only the last of them is synchronous. So several txcs become durable with one fsync on the WAL (group commit).

25-3 The WAL: record format#

RocksDB's WAL (the *.log files in the BlueFS directory named db.wal) is divided into fixed 32-kilobyte blocks (kBlockSize = 32768). Each record has a 7-byte header (kHeaderSize = 4 + 2 + 1): a masked crc32c checksum (4 bytes), the length (2 bytes) and the type (1 byte). The type is one of: kFullType=1 (the whole record in this block), kFirstType=2, kMiddleType=3 or kLastType=4 (a record broken across several blocks). In recyclable mode, a 4-byte log number is added and the header becomes 11 bytes.

WAL record (logic from RocksDB db/log_format.h, Apache-2.0/GPL-2.0; drawing from this book)
 +--------+--------+------+---------------------------+
 | CRC(4) | Len(2) | T(1) |  payload: WriteBatch part  |
 +--------+--------+------+---------------------------+
 WriteBatch = seq(8) | count(4) | {type, [cf id], key, value}*

25-4 memtable and flush#

The memtable is a skiplist by default. With Ceph's defaults, each memtable has write_buffer_size=16M, and up to max_write_buffer_number=64 memtables (that is at most 1 gigabyte) can be in memory. min_write_buffer_number_to_merge=6 means a flush only happens when at least six immutable memtables are ready. Then all six are merged together, and keys that were written and deleted in the meantime (such as deferred writes and temporary pg log records) never reach an SST. For the L and P CFs this number is raised to 32 in bluestore_rocksdb_cfs, because their keys are very short-lived.

25-5 SST: file format#

25-6 Compaction and levels#

With compaction_style=kCompactionStyleLevel, each level Ln (n ≥ 1) is a set of SSTs with no key overlap. When the number of L0 files reaches level0_file_num_compaction_trigger=8, an L0→L1 compaction starts. When a level's size exceeds its target, part of it is merged into the next level. Ceph's defaults are max_bytes_for_level_base=1G, max_bytes_for_level_multiplier=8 and max_background_jobs=4 (flush and compaction together).

Numerical example of DB size: an OSD with 10 million 4M objects (about 40 terabytes) and onodes of about 800 bytes has about 8 gigabytes of data just for O. With omap and the pg log it may reach 20 to 30 gigabytes. With the defaults, L1+L2+L3 = 1 + 8 + 64 = 73G is needed on the DB to keep all level 3 data on the fast device. If block.db is only 30G, the use_some_extra policy puts part of L3 on the DB and the rest on the slow device, and BLUEFS_SPILLOVER is reported.

25-7 Column family sharding#

The definition syntax is <prefix>(<shards>[,<l>-<h>])[=<options>], which parse_sharding_def in RocksDBStore.cc interprets. get_key_cf hashes the [hash_l, hash_h) part of the key with ceph_str_hash_rjenkins and returns hash % shards.handles.size(). In p(3,0-12), bytes 0 to 12 of the key (that is the pool, the PG hash and the start of the nid) are used, so that a PG's omaps stay in one shard and a range scan runs on one CF.

Why shard? (1) Each CF has its own memtable and compaction. So frequently changing data (L and P) is not merged into the same SST as stable onodes (O), and write amplification drops. (2) Compaction runs in parallel across several CFs. (3) A separate cache for O (block_cache={type=binned_lru}) prevents onodes from being evicted by omap.

# OSD stopped:
ceph-bluestore-tool show-sharding --path /var/lib/ceph/osd/ceph-4
ceph-bluestore-tool reshard --path /var/lib/ceph/osd/ceph-4 \
    --sharding "m(3) p(3,0-12) O(3,0-13)=block_cache={type=binned_lru} L P"
ceph-kvstore-tool bluestore-kv /var/lib/ceph/osd/ceph-4 stats
# OSD running:
ceph tell osd.4 compact
ceph daemon osd.4 bluefs stats

25-8 Tuning and diagnosis#

ceph daemon osd.4 perf dump rocksdb | jq '.rocksdb | {get_latency, submit_latency, submit_sync_latency, compact, compact_running}'
ceph config set osd bluestore_rocksdb_options_annex "max_background_jobs=6"   # only the changed keys

25-9 A WriteBatch at the byte level#

To see what a BlueStore transaction looks like on the WAL, we hand-encode a hypothetical WriteBatch with two operations: a Put in the CF with ID 3 (key Ok1o and value hello) and a SingleDelete in the CF with ID 7 (key L42). The sequence number is 1234567. The type constants come from db/dbformat.h in RocksDB: kTypeColumnFamilyValue=0x5 and kTypeColumnFamilySingleDeletion=0x8. The encoder is our own.

These 32 bytes are then written as the payload of a WAL record, with a 7-byte header (crc + length + type kFullType), into the current 32K block. If the payload crosses the end of the block, it is split into First/Middle/Last records. If fewer than 7 bytes remain to the end of the block, those few bytes are filled with zeros.

25-10 BlueRocksEnv: RocksDB without the kernel file system#

BlueRocksEnv inherits from rocksdb::EnvWrapper and sends RocksDB's file calls to BlueFS:

BlueFS directories express the physical location: db (the DB device), db.wal (the WAL device) and db.slow (the main device). RocksDB creates files in these directories through the db_paths and wal_dir options, and the BlueFS volume selector (previous section) makes the final decision about the device.

pseudocode: fsync of a WAL file (this book's summary)
  BlueFS::fsync(h):
      _flush_F(h)                         # write the buffer to extents with aio
      if h.file.fnode.size changed (or allocated changed):
          log.t.op_file_update_inc(h.file.fnode)
          _flush_and_sync_log_LD()        # the second synchronous write: the BlueFS log
      else:
          _flush_bdev(h)                  # only flush the device

25-11 Algorithms inside RocksDB at the code level#

The previous sections described what RocksDB does inside BlueStore. This section examines seven key algorithms with the code of RocksDB 9.11, that is the version the src/rocksdb submodule on the main branch points to (the ceph/rocksdb repository). The Reef and Squid versions point to other commits. Defaults that changed across versions are marked separately.

25-11-1 The skiplist in the memtable#

The default memtable is an InlineSkipList. Ceph does not change the memtable factory, so this same structure is used for all column families. Its constructor has max_height = 12 and branching_factor = 4.

pseudocode
RandomHeight(): h = 1
    while h < 12 and rnd.Next() < 2^32/4: h++          # probability 1/4 for each higher level
FindGreaterOrEqual(key):
    x = head ; level = max_height − 1
    loop: next = x.next[level]
          cmp = (next == null) ? +1 : compare(next.key, key)
          if cmp == 0 or (cmp > 0 and level == 0): return next
          if cmp < 0: x = next                       # move forward
          else: level −= 1                           # move down
Insert(key): height h ; find prev[] and next[] from top to bottom ;
             build the node with one allocation from the arena (pointers before the key) ;
             link from level 0 upward with CAS or store-release

Numerical example. With write_buffer_size = 16 MiB (Ceph's default) and entries of about 100 bytes, each memtable has about 160000 nodes. The expected number of nodes whose height is at least h is $n \cdot 4^{-(h-1)}$: 160000, 40000, 10000, 2500, 625, 156, 39, 10, 2.4 for h = 1..9. So the effective height is about 9, and a Seek walks about 9 levels with 1 to 3 horizontal steps per level, that is fewer than 30 key comparisons. The average height is 4/3, so each node has on average 8 × 4/3 ≈ 10.7 bytes of pointers. A separate simulation with 160000 nodes gave an average of 1.335.

25-11-2 The write group (group commit)#

pseudocode
JoinBatchGroup(w): change newest_writer_ to w with CAS
    if the queue was empty: w is the leader
    else: wait on w.state (short spin, then futex/condvar)
EnterAsBatchGroupLeader(leader):
    max = max_write_batch_group_size_bytes                # 1 MiB
    if size(leader) ≤ max/8: max = size(leader) + max/8   # 128 KiB
    for w from oldest to newest:
        if (w.sync and not leader.sync) or w.disable_wal ≠ leader.disable_wal
           or w.no_slowdown ≠ … or size + size(w) > max: cut
        else: add w to the group

Numerical example. The leader has a sixty-kilobyte batch. Because 60K ≤ 128K, the group cap becomes 60K + 128K = 188K. The next three writers have 40K, 50K and 70K. 60 + 40 + 50 = 150K is accepted, but with 70K the sum becomes 220K, which exceeds 188K, so the group is cut there. If the leader has sync=false and the second writer sync=true, the group is cut before the second writer, because a leader without sync cannot guarantee an fsync for another writer.

25-11-3 WAL record format and fragmentation#

pseudocode
left = len(record) ; begin = true
do:
    leftover = 32768 − block_offset
    if leftover < 7: write zeros to the end of the block ; block_offset = 0
    avail = 32768 − block_offset − 7
    frag = min(left, avail) ; end = (frag == left)
    type = begin&&end ? FULL(1) : begin ? FIRST(2) : end ? LAST(4) : MIDDLE(3)
    EmitPhysicalRecord(type, ptr, frag)      # masked crc32c(type‖payload), len (2B), type (1B)
    left −= frag ; begin = false
while left > 0

Numerical example. A seventy-thousand-byte WriteBatch starts at block_offset = 30000:

• leftover = 2768 and avail = 2761, so FIRST with 2761 bytes.
• Next block: avail = 32761, so MIDDLE.
• Third block: MIDDLE again with 32761 bytes.
• The remainder is 70000 − 2761 − 2×32761 = 1717 bytes, so LAST at the start of the fourth block.

The next block_offset is 7 + 1717 = 1724. The total overhead is 4 × 7 = 28 bytes.

25-11-4 The SST data block: key deltas and restarts#

pseudocode
Add(key, value):
    if counter == 16: restarts.push(buffer.size) ; shared = 0 ; counter = 0
    else: shared = length of the prefix shared with the previous key
    PutVarint32 ×3 (shared, len(key)−shared, len(value)) ; append key[shared:] ; append value
Finish(): append restarts[] as u32 ; append Pack(index_type, num_restarts)   # bit31
Seek(target): binary search over restarts comparing each restart's full key
              then a linear walk, rebuilding key = prev[0:shared] + delta
EncodeVarint32(v): while v ≥ 128: emit (v & 127) | 128 ; v ≫= 7 ; emit v

Numerical example. With rocksdb_block_size = 4K (Ceph's default) and entries of about 300 bytes, each block has about 13 entries, that is fewer than 16, so only one restart. Seek within the block is effectively linear, but 13 comparisons on data that is in cache. In metadata blocks where values are small (for example 100 keys of 40 bytes), there are 7 restarts and the binary search takes 3 comparisons plus at most 15 linear steps. The varint encoding of 300 is the two bytes AC 02, and of 18 the single byte 12.

25-11-5 The cache-line-local Bloom filter#

With rocksdb_bloom_bits_per_key = 20, Ceph calls NewBloomFilterPolicy(20) (RocksDBStore::load_rocksdb_options). In RocksDB 9.11 the default is format_version = 6, so the filter implementation is FastLocalBloomImpl: all probes of a key fall within one five-hundred-and-twelve-bit cache line.

pseudocode
k = ChooseNumProbes(1000·b)            # empirical table: 20000 millibits ⇒ 11
AddHash(h1, h2): line = FastRange32(h1, len/64) · 64     # choose the cache line
    h = h2
    repeat k: bit = h ≫ 23 (top 9 bits ⇒ 0..511) ; line[bit] = 1 ; h ×= 0x9e3779b9
MayMatch: the same path ; if any bit is zero ⇒ "definitely not present"
FP ≈ the average of a standard Bloom for (keys per line ± standard deviation)

Numerical example.

• b = 20: k = 11, and each cache line has on average 512/20 = 25.6 keys. The standard deviation is $\sqrt{25.6} \approx 5.06$, so the crowded and sparse cases have 30.66 and 20.54 keys per line. The estimated FP is about $1.7 \times 10^{-4}$ (0.017%), versus $6.8 \times 10^{-5}$ for a standard Bloom with the same memory.
• b = 10: k = 6 and FP about 0.95%.
• Memory: an SST with 1 million keys has $20 \times 10^6$ bits, that is 2.5 MB of filter.

The price of locality is a slightly higher FP. In return, each query has only one cache miss instead of 11.

25-11-6 The compaction score and dynamic level sizes#

pseudocode
CalculateBaseBytes (dynamic):
    max_level_size = the largest level ; cur = max_level_size
    for i = num_levels−2 down to first_non_empty: cur /= multiplier
    base_level = first_non_empty
    while base_level > 1 and cur > base_bytes_max: base_level−− ; cur /= multiplier
    target[base_level] = max(cur, base_bytes_max) ; target[i+1] = target[i]·multiplier (floor base_bytes_max)
ComputeCompactionScore: for level in 0 .. num_levels−2:
    L0: score = max(files/trigger, size/max_bytes_for_level_base …) and ×10 if > 1
    L>0: score = bytes_not_compacting / target   (or / (target + incoming) ×10 if ≥ 1)
highest score ⇒ first compaction

Numerical example (Ceph's defaults: base = 1 GiB, multiplier = 8, seven levels). Suppose L6 has two hundred gigabytes. From the bottom up: L5 = 25, L4 = 3.125 and L3 = 0.39 GiB. Because 0.39 is larger than base_bytes_min = 1/8 GiB, base_level = L3 and the targets are L3 = max(0.39, 1) = 1 GiB, L4 = 3.125, L5 = 25 and L6 = 200. With actual sizes of 0.31, 3.0 and 24, the scores are 0.31, 0.96 and 0.96. None has reached 1. If L0 has eight files, its score is 8/8 = 1, which becomes 10 after multiplying by kScoreScale, so L0→L3 has priority. L1 and L2 stay empty: this is exactly the dynamic level bytes behaviour, which takes data straight to the base level.

25-12 Supplementary and up to date#

• Sharding has been the default for new OSDs since Pacific. Compare main's default string, m(3) p(3,0-12) O(3,0-13)=block_cache={type=binned_lru} L=min_write_buffer_number_to_merge=32 P=min_write_buffer_number_to_merge=32, with ceph config help bluestore_rocksdb_cfs on your version.
• bluefs_spillover_cleaner (present in Tentacle and main, default false) is a background thread that moves files that spilled over to the slow device back to the DB, if there is space.
• According to the documentation, Pacific brought experimental support for level_compaction_dynamic_level_bytes (together with the fit_to_fast policy).

25-13 Hands-on exercise#

1. 1. Extract a test OSD's RocksDB files with ceph-bluestore-tool bluefs-export, see the CFs with ldb --db=... list_column_families, and read the sizes of data, index and filter with sst_dump --file=... --show_properties.
2. 2. Read the last 8 bytes of an SST with tail -c 8 | xxd and confirm the magic number (mind the little-endian order).
3. 3. For your DB size (for example 60G), compute which levels fit with the defaults. Then read bluefs stats and compare it with your prediction.
4. 4. Compare submit_sync_latency before and after moving the WAL to NVMe.

25-14 Review questions#

1. 1. How does group commit in _kv_sync_thread reduce the number of fsyncs?
2. 2. How many bytes is a WAL record header, and what fields does it have?
3. 3. Why is min_write_buffer_number_to_merge=6 important for Ceph?
4. 4. Why is this number 32 for the L and P CFs?
5. 5. What role does an SST's filter block play on the read path?
6. 6. With the defaults, how large is L3?
7. 7. How does use_some_extra differ from rocksdb_original?
8. 8. How is the CF of a per-pg omap key chosen?
9. 9. Why should a PG's omap stay in one shard?
10. 10. Why is changing the whole bluestore_rocksdb_options string dangerous, and what is the alternative?

Answers to these questions are in Appendix E: Answers to Review Questions. Answers to this chapter's questions →

25-15 Sources for this chapter#

• RocksDBStore
  https://github.com/ceph/ceph/blob/main/src/kv/RocksDBStore.cc
• BlueStore configuration and sharding
  https://docs.ceph.com/en/latest/rados/configuration/bluestore-config-ref/
• RocksDB WAL format
  https://github.com/facebook/rocksdb/wiki/Write-Ahead-Log-File-Format
• Block-based table format
  https://github.com/facebook/rocksdb/wiki/Rocksdb-BlockBasedTable-Format
• Leveled compaction
  https://github.com/facebook/rocksdb/wiki/Leveled-Compaction
• Column families
  https://github.com/facebook/rocksdb/wiki/Column-Families
• log_format.h
  https://github.com/facebook/rocksdb/blob/main/db/log_format.h
• LSM-tree (O'Neil et al., 1996
  https://www.cs.umb.edu/~poneil/lsmtree.pdf