Nydus 原始碼解讀：nydus-image create

$ mkdir -p fs/dir1
$ mkdir -p fs/dir2
$ touch fs/foo.txt
$ echo abcde > fs/dir1/bar.txt

$ mkdir -p fs/dir1/dir1-1
$ touch fs/dir1/dir1-1/foo
$ echo abc > fs/dir1/dir1-1/hello

$ tree fs/
fs/
├── dir1
│   ├── bar.txt
│   └── dir1-1
│       ├── foo
│       └── hello
├── dir2
└── foo.txt

3 directories, 4 files

$ mkdir output
$ nydus-image create \
    --fs-version 5 \
    --bootstrap output/bootstrap \
    --blob-dir output \
    fs

builder = Box::new(DirectoryBuilder::new())

builder.build(&mut build_ctx, &mut bootstrap_mgr, &mut blob_mgr)

fn build(
    &mut self,
    ctx: &mut BuildContext,
    bootstrap_mgr: &mut BootstrapManager,
    blob_mgr: &mut BlobManager,
) -> Result<BuildOutput> {
    // 建立一個 BootstrapContext，在構建中將 bootstrap 資料儲存在記憶體中
    let mut bootstrap_ctx = bootstrap_mgr.create_ctx(ctx.inline_bootstrap)?;

    // 基於輸入原始檔夾在記憶體中構建檔案樹結構
    let mut tree = self.build_tree_from_fs(ctx, &mut bootstrap_ctx, layer_idx)?;

    let mut bootstrap = Bootstrap::new()?;

    // 如果是多層構建，當層資料需要依賴父層進行計算
    if bootstrap_ctx.layered {
        // Merge with lower layer if there's one, do not prepare `prefetch` list during merging.
        bootstrap.build(ctx, &mut bootstrap_ctx, &mut tree)?;
        // 本例中不會走入這個分支，這裡我們就不介紹 apply 方法了。
        tree = bootstrap.apply(ctx, &mut bootstrap_ctx, bootstrap_mgr, blob_mgr, None)?;
    }

    // 將樹狀結構轉換為一個平鋪的陣列
    // 儲存到 `bootstrap_ctx.nodes` 中
    timing_tracer!(
        { bootstrap.build(ctx, &mut bootstrap_ctx, &mut tree) },
        "build_bootstrap"
    )?;

    // Dump blob file
    let mut blob_ctx = BlobContext::new()?;
    let blob_index = blob_mgr.alloc_index()?;
    let mut blob = Blob::new();

    // blob.dump 將記憶體的資料寫入磁碟 blob 檔案
    let blob_exists = timing_tracer!(
        {
            blob.dump(
                ctx,
                &mut blob_ctx,
                blob_index,
                &mut bootstrap_ctx.nodes,
                &mut blob_mgr.chunk_dict_cache,
            )
        },
        "dump_blob"
    )?;

    let mut blob_writer = blob_ctx.writer.take().unwrap();
    let blob_id = blob_ctx.blob_id();

    // 這裡的 blob_exists 表示 compressed_blob_size  > 0，即 blob 檔案大小大於 0 
    if blob_exists {
        blob_writer.finalize(blob_id.clone())?;
        // Add new blob to blob table.
        blob_mgr.add(blob_ctx);
    }

    // 將 bootstrap 寫入檔案
    let blob_table = blob_mgr.to_blob_table(ctx)?;
    bootstrap.dump(ctx, &mut bootstrap_ctx, &blob_table)?;

    bootstrap_mgr.add(bootstrap_ctx);
}

pub(crate) struct Tree {
    /// Filesystem node.
    pub node: Node,
    /// Children tree nodes.
    pub children: Vec<Tree>,
}

pub fn build(
    &mut self,
    ctx: &mut BuildContext,
    bootstrap_ctx: &mut BootstrapContext,
    tree: &mut Tree,
) -> Result<()> {
    // 設定為 root 節點
    tree.node.index = RAFS_ROOT_INODE;
    tree.node.inode.set_ino(RAFS_ROOT_INODE);

    bootstrap_ctx.inode_map.insert(
        (tree.node.layer_idx, tree.node.src_ino, tree.node.src_dev),
        vec![tree.node.index],
    );

    let mut nodes = Vec::with_capacity(0x10000);
    // 將根節點 push 進去
    nodes.push(tree.node.clone());

    // 呼叫 build_rafs 構建子節點
    self.build_rafs(ctx, bootstrap_ctx, tree, &mut nodes)?;

    // 將結果儲存到 bootstrap_ctx.nodes 中
    bootstrap_ctx.nodes = nodes;

    Ok(())
}

fn build_rafs(
    &mut self,
    ctx: &mut BuildContext,
    bootstrap_ctx: &mut BootstrapContext,
    tree: &mut Tree,
    nodes: &mut Vec<Node>,
) -> Result<()> {
    // 分配 index，新的值為 nodes 陣列長度 + 1
    let index = nodes.len() as u32 + 1;
    // 每個 node 都會有這個 index 屬性，方便從 nodes 陣列獲取該節點
    let parent = &mut nodes[tree.node.index as usize - 1];

    // Maybe the parent is not a directory in multi-layers build scenario, so we check here.
    if parent.is_dir() {
        parent.inode.set_child_index(index);
        parent.inode.set_child_count(tree.children.len() as u32);
    }

    let mut dirs: Vec<&mut Tree> = Vec::new();
    let parent_ino = parent.inode.ino();

    for child in tree.children.iter_mut() {
        let index = nodes.len() as u64 + 1;
        child.node.index = index;
        child.node.inode.set_parent(parent_ino);

        // 這裡刪除了 hardlink 處理的程式碼
        if !hardlink {
            // 分配 ino，也就是陣列索引
            child.node.inode.set_ino(index);
            child.node.inode.set_nlink(1);
        }

        // 這裡處理 whiteout 型別
        match (
            bootstrap_ctx.layered,
            child.node.whiteout_type(ctx.whiteout_spec),
        ) {
            (true, Some(whiteout_type)) => {
                // 這裡省略 layered 的處理
            }
            (false, Some(whiteout_type)) => {
                // Remove overlayfs opaque xattr for single layer build
                if whiteout_type == WhiteoutType::OverlayFsOpaque {
                    child
                        .node
                        .remove_xattr(&OsString::from(OVERLAYFS_WHITEOUT_OPAQUE));
                }
                nodes.push(child.node.clone());
            }
            _ => {
                // 這個例子中走預設分支，
                // 將 child.node 新增到 nodes 陣列
                nodes.push(child.node.clone());
            }
        }

        // 如果當前 child 是資料夾，還需要再次遞迴處理
        // 所以將該資料夾儲存到 dirs 陣列
        if child.node.is_dir() {
            dirs.push(child);
        }
    }

    // 迴圈完當前節點的 children 之後
    // 還需要再次遞迴處理子資料夾
    for dir in dirs {
        self.build_rafs(ctx, bootstrap_ctx, dir, nodes)?;
    }

}

pub fn dump<'a, T: ChunkDict>(
    &mut self,
    ctx: &BuildContext,
    blob_ctx: &'a mut BlobContext,
    blob_index: u32,
    nodes: &mut Vec<Node>,
    chunk_dict: &mut T,
) -> Result<bool> {
    match ctx.source_type {
        SourceType::Directory | SourceType::Diff => {
            // layout_blob_simple 函式用於從 Vec<Node> 型別的 nodes
            // 返回 Vec<usize> 型別的 inodes
            // prefetch 相關的變數和預拉取有關，這裡我們暫時忽略
            let (inodes, prefetch_entries) = blob_ctx
                .blob_layout
                .layout_blob_simple(&ctx.prefetch, nodes)?;

            // inodes 型別為 Vec<usize> ，inode 即 usize，
            // 用這個 inode 作為陣列索引，從 nodes 陣列獲取 node 元素
            // 然後再呼叫 node 的 dump_blob 將 node 的資訊寫到磁碟
            for (idx, inode) in inodes.iter().enumerate() {
                let node = &mut nodes[*inode];
                let size = node
                    .dump_blob(ctx, blob_ctx, blob_index, chunk_dict)
                    .context("failed to dump blob chunks")?;
            }
        }
        SourceType::StargzIndex => {
            // 省略 stargz 處理的 case
        }
    }

    // 如果沒有指定 blob_id，則自動從 blob_hash 建立 ID
    if blob_ctx.blob_id.is_empty() {
        blob_ctx.blob_id = format!("{:x}", blob_ctx.blob_hash.clone().finalize());
    }

    // compressed_blob_size > 0 才需要寫到磁碟，
    // 這裡返回的 blob_exists 即是該標誌
    let blob_exists = blob_ctx.compressed_blob_size > 0;

    Ok(blob_exists)
}

pub fn dump_blob<T: ChunkDict>(
    self: &mut Node,
    ctx: &BuildContext,
    blob_ctx: &mut BlobContext,
    blob_index: u32,
    chunk_dict: &mut T,
) -> Result<u64> {
    if self.is_dir() {
        return Ok(0);
    } else if self.is_symlink() {
      // 針對 symlink 的特殊處理
        return Ok(0);
    } else if self.is_special() {
      // 針對 special 檔案的特殊處理
        return Ok(0);
    }

    let mut file = File::open(&self.path)
        .with_context(|| format!("failed to open node file {:?}", self.path))?;
    let mut inode_hasher = RafsDigest::hasher(ctx.digester);
    let mut blob_size = 0u64;

    // `child_count` of regular file is reused as `chunk_count`.
    // 一個檔案過大的話，大於 chunk size，就要被拆分為多個 chunk 儲存。
    for i in 0..self.inode.child_count() {
        let chunk_size = blob_ctx.chunk_size;
        let file_offset = i as u64 * chunk_size as u64;
        let chunk_size = if i == self.inode.child_count() - 1 {
          // 最後一個 chunk，可能實際的 chunk size 小於 blob_ctx.chunk_size
          // 這裡檢查下是否 chunk size 不足
            (self.inode.size() as u64)
                .checked_sub((chunk_size * i) as u64)
                .ok_or_else(|| {
                    anyhow!("the rest chunk size of inode is bigger than chunk_size")
                })? as u32
        } else {
            chunk_size
        };

        // 使用 read_exact 讀取一個 chunk 的資料到 chunk_data
        let mut chunk_data = &mut blob_ctx.chunk_data_buf[0..chunk_size as usize];
        file.read_exact(&mut chunk_data)
            .with_context(|| format!("failed to read node file {:?}", self.path))?;

        // 生成 chunk_id
        let chunk_id = RafsDigest::from_buf(chunk_data, ctx.digester);
        inode_hasher.digest_update(chunk_id.as_ref());

        // 生成 chunk 物件，對本例來說這是一個 RafsV5ChunkInfo 型別的資料結構
        let mut chunk = self.inode.create_chunk();
        chunk.set_id(chunk_id);

        // 通過對比 chunk digest 來判斷是否該 chunk 已經存在
        // 注意這裡從兩個快取的地方獲取，
        // 一個是 blob_ctx.chunk_dict，
        // 一個是輸入引數的 chunk_dict
        let exist_chunk = match blob_ctx.chunk_dict.get_chunk(&chunk_id) {
            Some(v) => Some((v, true)),
            None => chunk_dict.get_chunk(&chunk_id).map(|v| (v, false)),
        };

        if let Some((cached_chunk, from_dict)) = exist_chunk {
            if cached_chunk.uncompressed_size() == 0
                || cached_chunk.uncompressed_size() == chunk_size
            {
              // 從 cached_chunk 拷貝資料
                chunk.copy_from(cached_chunk);
                chunk.set_file_offset(file_offset);
                if from_dict {
                  // 如果是 blob_ctx.chunk_dict 裡已存在該 chunk，
                  // 則設定 blob index 為該 chunk 的 blob id
                    let idx = blob_ctx.chunk_dict.get_real_blob_idx(chunk.blob_index());
                    chunk.set_blob_index(idx);
                }

                let source = if from_dict {
                    ChunkSource::Dict
                } else {
                    ChunkSource::Build
                };
                self.chunks.push(NodeChunk {
                    source,
                    inner: chunk,
                });

                continue;
            }
        }

        // 沒有從任何快取中獲得該 chunk，則建立該 chunk
        // 首先先對資料進行壓縮處理
        let (compressed, is_compressed) = compress::compress(&chunk_data, ctx.compressor)
            .with_context(|| format!("failed to compress node file {:?}", self.path))?;
        let compressed_size = compressed.len();

        // 4k 對齊
        let aligned_chunk_size = if ctx.aligned_chunk {
            try_round_up_4k(chunk_size).unwrap()
        } else {
            chunk_size
        };

        // 更新 blob_ctx 中的一些資訊
        let pre_decompress_offset = blob_ctx.decompress_offset;
        let pre_compress_offset = blob_ctx.compress_offset;

        blob_ctx.compress_offset += compressed_size as u64;
        blob_ctx.decompressed_blob_size =
            blob_ctx.decompress_offset + aligned_chunk_size as u64;
        blob_ctx.compressed_blob_size += compressed_size as u64;
        blob_ctx.decompress_offset += aligned_chunk_size as u64;
        blob_ctx.blob_hash.update(&compressed);

        // 將壓縮後的 chunk 資料寫入到 blob 檔案
        if let Some(writer) = &mut blob_ctx.writer {
            writer
                .write_all(&compressed)
                .context("failed to write blob")?;
        }

        // 更新 chunk 物件的一些屬性
        let chunk_index = blob_ctx.alloc_index()?;
        chunk.set_chunk_info(
            blob_index,
            chunk_index,
            file_offset,
            pre_decompress_offset,
            pre_compress_offset,
            compressed_size,
            chunk_size,
            is_compressed,
        )?;

        blob_ctx.add_chunk_meta_info(&chunk)?;

        // 將 chunk 物件加入到 chunk_dict，即該方法的輸入引數
        chunk_dict.add_chunk(chunk.clone());

      // 將 chunk 物件新增到 self.chunks
        self.chunks.push(NodeChunk {
            source: ChunkSource::Build,
            inner: chunk,
        });
        blob_size += compressed_size as u64;
    }

    Ok(blob_size)
}

fn dump_rafsv5(
    &mut self,
    ctx: &mut BuildContext,
    bootstrap_ctx: &mut BootstrapContext,
    blob_table: &RafsV5BlobTable,
) -> Result<()> {
    // 計算資料夾 inode 的 digest
    for idx in (0..bootstrap_ctx.nodes.len()).rev() {
        self.digest_node(ctx, bootstrap_ctx, idx);
    }

    // Set inode table
    let super_block_size = size_of::<RafsV5SuperBlock>();
    let inode_table_entries = bootstrap_ctx.nodes.len() as u32;
    // 建立 inode table
    let mut inode_table = RafsV5InodeTable::new(inode_table_entries as usize);
    let inode_table_size = inode_table.size();

    // Set blob table, use sha256 string (length 64) as blob id if not specified
    let prefetch_table_offset = super_block_size + inode_table_size;
    let blob_table_offset = prefetch_table_offset + prefetch_table_size;
    let blob_table_size = blob_table.size();
    let extended_blob_table_offset = blob_table_offset + blob_table_size;
    let extended_blob_table_size = blob_table.extended.size();
    let extended_blob_table_entries = blob_table.extended.entries();

    // 建立 super block
    let mut super_block = RafsV5SuperBlock::new();
    let inodes_count = bootstrap_ctx.inode_map.len() as u64;
    super_block.set_inodes_count(inodes_count);
    super_block.set_inode_table_offset(super_block_size as u64);
    super_block.set_inode_table_entries(inode_table_entries);
    super_block.set_blob_table_offset(blob_table_offset as u64);
    super_block.set_blob_table_size(blob_table_size as u32);
    super_block.set_extended_blob_table_offset(extended_blob_table_offset as u64);
    super_block.set_extended_blob_table_entries(u32::try_from(extended_blob_table_entries)?);
    super_block.set_prefetch_table_offset(prefetch_table_offset as u64);
    super_block.set_prefetch_table_entries(prefetch_table_entries);
    super_block.set_compressor(ctx.compressor);
    super_block.set_digester(ctx.digester);
    super_block.set_chunk_size(ctx.chunk_size);

    // Set inodes and chunks
    let mut inode_offset = (super_block_size
        + inode_table_size
        + prefetch_table_size
        + blob_table_size
        + extended_blob_table_size) as u32;

    for node in &mut bootstrap_ctx.nodes {
        inode_table.set(node.index, inode_offset)?;
        // Add inode size
        inode_offset += node.inode.inode_size() as u32;
        // Add chunks size
        if node.is_reg() {
            inode_offset += node.inode.child_count() * size_of::<RafsV5ChunkInfo>() as u32;
        }
    }

    // Dump super block
    super_block
        .store(bootstrap_ctx.writer.as_mut())
        .context("failed to store superblock")?;

    // Dump inode table
    inode_table
        .store(bootstrap_ctx.writer.as_mut())
        .context("failed to store inode table")?;

    // Dump blob table
    blob_table
        .store(bootstrap_ctx.writer.as_mut())
        .context("failed to store blob table")?;

    // Dump extended blob table
    blob_table
        .store_extended(bootstrap_ctx.writer.as_mut())
        .context("failed to store extended blob table")?;

    // Dump inodes and chunks
    timing_tracer!(
        {
            for node in &bootstrap_ctx.nodes {
                node.dump_bootstrap_v5(&ctx, bootstrap_ctx.writer.as_mut())
                    .context("failed to dump bootstrap")?;
            }

            Ok(())
        },
        "dump_bootstrap",
        Result<()>
    )?;

    bootstrap_ctx
        .writer
        .finalize(Some(bootstrap_ctx.name.to_string()))?;

    Ok(())
}