[era_check (rust)] First code drop

src/bin/pdata_tools.rs

@@ -34,6 +34,8 @@ fn main_() -> Result<()> {
         cache_repair::run(&new_args);
     } else if name_eq(name, "cache_restore") {
         cache_restore::run(&new_args);
+    } else if name_eq(name, "era_check") {
+        era_check::run(&new_args);
     } else if name_eq(name, "thin_check") {
         thin_check::run(&new_args);
     } else if name_eq(name, "thin_dump") {

src/commands/era_check.rs

@@ -0,0 +1,84 @@
+extern crate clap;
+
+use atty::Stream;
+use clap::{App, Arg};
+use std::path::Path;
+use std::process;
+use std::sync::Arc;
+
+use crate::commands::utils::*;
+use crate::era::check::{check, EraCheckOptions};
+use crate::report::*;
+
+//------------------------------------------
+
+pub fn run(args: &[std::ffi::OsString]) {
+    let parser = App::new("era_check")
+        .version(crate::version::tools_version())
+        // flags
+        .arg(
+            Arg::with_name("ASYNC_IO")
+                .help("Force use of io_uring for synchronous io")
+                .long("async-io")
+                .hidden(true),
+        )
+        .arg(
+            Arg::with_name("AUTO_REPAIR")
+                .help("Auto repair trivial issues.")
+                .long("auto-repair"),
+        )
+        .arg(
+            Arg::with_name("IGNORE_NON_FATAL")
+                .help("Only return a non-zero exit code if a fatal error is found.")
+                .long("ignore-non-fatal-errors"),
+        )
+        .arg(
+            Arg::with_name("QUIET")
+                .help("Suppress output messages, return only exit code.")
+                .short("q")
+                .long("quiet"),
+        )
+        .arg(
+            Arg::with_name("SB_ONLY")
+                .help("Only check the superblock.")
+                .long("super-block-only"),
+        )
+        // arguments
+        .arg(
+            Arg::with_name("INPUT")
+                .help("Specify the input device to check")
+                .required(true)
+                .index(1),
+        );
+
+    let matches = parser.get_matches_from(args);
+    let input_file = Path::new(matches.value_of("INPUT").unwrap());
+
+    let report = if matches.is_present("QUIET") {
+        std::sync::Arc::new(mk_quiet_report())
+    } else if atty::is(Stream::Stdout) {
+        std::sync::Arc::new(mk_progress_bar_report())
+    } else {
+        Arc::new(mk_simple_report())
+    };
+
+    check_input_file(input_file, &report);
+    check_file_not_tiny(input_file, &report);
+    check_not_xml(input_file, &report);
+
+    let opts = EraCheckOptions {
+        dev: input_file,
+        async_io: matches.is_present("ASYNC_IO"),
+        sb_only: matches.is_present("SB_ONLY"),
+        ignore_non_fatal: matches.is_present("IGNORE_NON_FATAL"),
+        auto_repair: matches.is_present("AUTO_REPAIR"),
+        report: report.clone(),
+    };
+
+    if let Err(reason) = check(&opts) {
+        report.fatal(&format!("{}", reason));
+        process::exit(1);
+    }
+}
+
+//------------------------------------------

src/commands/mod.rs

@@ -2,6 +2,7 @@ pub mod cache_check;
 pub mod cache_dump;
 pub mod cache_repair;
 pub mod cache_restore;
+pub mod era_check;
 pub mod thin_check;
 pub mod thin_dump;
 pub mod thin_metadata_pack;

src/era/check.rs

@@ -0,0 +1,153 @@
+use anyhow::{anyhow, Result};
+use std::path::Path;
+use std::sync::Arc;
+
+use crate::era::superblock::*;
+use crate::era::writeset::*;
+use crate::io_engine::{AsyncIoEngine, IoEngine, SyncIoEngine};
+use crate::pdata::array::{self, ArrayBlock, ArrayError};
+use crate::pdata::array_walker::*;
+use crate::pdata::bitset::*;
+use crate::pdata::btree_walker::*;
+use crate::pdata::space_map::*;
+use crate::report::*;
+
+//------------------------------------------
+
+const MAX_CONCURRENT_IO: u32 = 1024;
+
+//------------------------------------------
+
+fn inc_superblock(sm: &ASpaceMap) -> anyhow::Result<()> {
+    let mut sm = sm.lock().unwrap();
+    sm.inc(SUPERBLOCK_LOCATION, 1)?;
+    Ok(())
+}
+
+//------------------------------------------
+
+struct EraChecker {
+    current_era: u32,
+}
+
+impl EraChecker {
+    pub fn new(current_era: u32) -> EraChecker {
+        EraChecker { current_era }
+    }
+}
+
+impl ArrayVisitor<u32> for EraChecker {
+    fn visit(&self, index: u64, b: ArrayBlock<u32>) -> array::Result<()> {
+        let mut errs: Vec<ArrayError> = Vec::new();
+
+        let dbegin = index as u32 * b.header.max_entries;
+        let dend = dbegin + b.header.max_entries;
+        for (era, dblock) in b.values.iter().zip(dbegin..dend) {
+            if era > &self.current_era {
+                errs.push(array::value_err(format!(
+                    "invalid era value at data block {}: {}",
+                    dblock, era
+                )));
+            }
+        }
+
+        match errs.len() {
+            0 => Ok(()),
+            1 => Err(errs[0].clone()),
+            _ => Err(array::aggregate_error(errs)),
+        }
+    }
+}
+
+//------------------------------------------
+
+pub struct EraCheckOptions<'a> {
+    pub dev: &'a Path,
+    pub async_io: bool,
+    pub sb_only: bool,
+    pub ignore_non_fatal: bool,
+    pub auto_repair: bool,
+    pub report: Arc<Report>,
+}
+
+struct Context {
+    report: Arc<Report>,
+    engine: Arc<dyn IoEngine + Send + Sync>,
+}
+
+fn mk_context(opts: &EraCheckOptions) -> anyhow::Result<Context> {
+    let engine: Arc<dyn IoEngine + Send + Sync>;
+
+    if opts.async_io {
+        engine = Arc::new(AsyncIoEngine::new(opts.dev, MAX_CONCURRENT_IO, false)?);
+    } else {
+        let nr_threads = std::cmp::max(8, num_cpus::get() * 2);
+        engine = Arc::new(SyncIoEngine::new(opts.dev, nr_threads, false)?);
+    }
+
+    Ok(Context {
+        report: opts.report.clone(),
+        engine,
+    })
+}
+
+fn check_superblock(sb: &Superblock) -> anyhow::Result<()> {
+    if sb.version > 1 {
+        return Err(anyhow!("unknown superblock version"));
+    }
+    Ok(())
+}
+
+pub fn check(opts: &EraCheckOptions) -> Result<()> {
+    let ctx = mk_context(opts)?;
+    let engine = &ctx.engine;
+    let report = &ctx.report;
+    let mut fatal = false;
+
+    report.set_title("Checking era metadata");
+
+    let metadata_sm = core_sm(engine.get_nr_blocks(), u8::MAX as u32);
+    inc_superblock(&metadata_sm)?;
+
+    let sb = read_superblock(engine.as_ref(), SUPERBLOCK_LOCATION)?;
+    check_superblock(&sb)?;
+
+    if opts.sb_only {
+        return Ok(());
+    }
+
+    let mut path = vec![0];
+    let writesets = btree_to_map::<Writeset>(
+        &mut path,
+        engine.clone(),
+        opts.ignore_non_fatal,
+        sb.writeset_tree_root,
+    )?;
+
+    for ws in writesets.values() {
+        let (_bs, err) = read_bitset_with_sm(
+            engine.clone(),
+            ws.root,
+            ws.nr_bits as usize,
+            metadata_sm.clone(),
+            opts.ignore_non_fatal,
+        )?;
+        if err.is_some() {
+            ctx.report.fatal(&format!("{}", err.unwrap()));
+            fatal = true;
+        }
+    }
+
+    let w = ArrayWalker::new_with_sm(engine.clone(), metadata_sm.clone(), opts.ignore_non_fatal)?;
+    let mut c = EraChecker::new(sb.current_era);
+    if let Err(e) = w.walk(&mut c, sb.era_array_root) {
+        ctx.report.fatal(&format!("{}", e));
+        fatal = true;
+    }
+
+    if fatal {
+        Err(anyhow!("fatal errors in metadata"))
+    } else {
+        Ok(())
+    }
+}

src/era/mod.rs

@@ -0,0 +1,3 @@
+pub mod check;
+pub mod superblock;
+pub mod writeset;

src/era/superblock.rs

@@ -0,0 +1,153 @@
+use anyhow::{anyhow, Result};
+use byteorder::{LittleEndian, WriteBytesExt};
+use nom::{bytes::complete::*, number::complete::*, IResult};
+use std::io::Cursor;
+
+use crate::checksum::*;
+use crate::era::writeset::Writeset;
+use crate::io_engine::*;
+
+//------------------------------------------
+
+pub const SPACE_MAP_ROOT_SIZE: usize = 128;
+pub const SUPERBLOCK_LOCATION: u64 = 0;
+
+const MAGIC: u64 = 0o17660203573; // 0x7EC1077B in hex
+const UUID_SIZE: usize = 16;
+
+//------------------------------------------
+
+#[derive(Debug, Clone)]
+pub struct SuperblockFlags {
+    pub clean_shutdown: bool,
+}
+
+#[derive(Debug, Clone)]
+pub struct Superblock {
+    pub flags: SuperblockFlags,
+    pub block: u64,
+    pub version: u32,
+
+    pub metadata_sm_root: Vec<u8>,
+
+    pub data_block_size: u32,
+    pub nr_blocks: u32,
+
+    pub current_era: u32,
+    pub current_writeset: Writeset,
+
+    pub writeset_tree_root: u64,
+    pub era_array_root: u64,
+
+    pub metadata_snap: u64,
+}
+
+fn unpack(data: &[u8]) -> IResult<&[u8], Superblock> {
+    let (i, _csum) = le_u32(data)?;
+    let (i, flags) = le_u32(i)?;
+    let (i, block) = le_u64(i)?;
+    let (i, _uuid) = take(16usize)(i)?;
+    let (i, _magic) = le_u64(i)?;
+    let (i, version) = le_u32(i)?;
+
+    let (i, metadata_sm_root) = take(SPACE_MAP_ROOT_SIZE)(i)?;
+    let (i, data_block_size) = le_u32(i)?;
+    let (i, _metadata_block_size) = le_u32(i)?;
+    let (i, nr_blocks) = le_u32(i)?;
+
+    let (i, current_era) = le_u32(i)?;
+    let (i, nr_bits) = le_u32(i)?;
+    let (i, root) = le_u64(i)?;
+
+    let (i, writeset_tree_root) = le_u64(i)?;
+    let (i, era_array_root) = le_u64(i)?;
+    let (i, metadata_snap) = le_u64(i)?;
+
+    Ok((
+        i,
+        Superblock {
+            flags: SuperblockFlags {
+                clean_shutdown: (flags & 0x1) != 0,
+            },
+            block,
+            version,
+            metadata_sm_root: metadata_sm_root.to_vec(),
+            data_block_size,
+            nr_blocks,
+            current_era,
+            current_writeset: Writeset { nr_bits, root },
+            writeset_tree_root,
+            era_array_root,
+            metadata_snap,
+        },
+    ))
+}
+
+pub fn read_superblock(engine: &dyn IoEngine, loc: u64) -> Result<Superblock> {
+    let b = engine.read(loc)?;
+
+    if metadata_block_type(b.get_data()) != BT::ERA_SUPERBLOCK {
+        return Err(anyhow!("bad checksum in superblock"));
+    }
+
+    if let Ok((_, sb)) = unpack(b.get_data()) {
+        Ok(sb)
+    } else {
+        Err(anyhow!("couldn't unpack superblock"))
+    }
+}
+
+//------------------------------------------
+
+fn pack_superblock<W: WriteBytesExt>(sb: &Superblock, w: &mut W) -> Result<()> {
+    // checksum, which we don't know yet
+    w.write_u32::<LittleEndian>(0)?;
+
+    // flags
+    let mut flags: u32 = 0;
+    if sb.flags.clean_shutdown {
+        flags |= 0x1;
+    }
+    w.write_u32::<LittleEndian>(flags)?;
+    w.write_u64::<LittleEndian>(sb.block)?;
+
+    w.write_all(&[0; UUID_SIZE])?;
+    w.write_u64::<LittleEndian>(MAGIC)?;
+    w.write_u32::<LittleEndian>(sb.version)?;
+
+    w.write_all(&sb.metadata_sm_root)?;
+    w.write_u32::<LittleEndian>(sb.data_block_size)?;
+    // metadata block size
+    w.write_u32::<LittleEndian>((BLOCK_SIZE >> SECTOR_SHIFT) as u32)?;
+    w.write_u32::<LittleEndian>(sb.nr_blocks)?;
+
+    w.write_u32::<LittleEndian>(sb.current_era)?;
+    w.write_u32::<LittleEndian>(sb.current_writeset.nr_bits)?;
+    w.write_u64::<LittleEndian>(sb.current_writeset.root)?;
+
+    w.write_u64::<LittleEndian>(sb.writeset_tree_root)?;
+    w.write_u64::<LittleEndian>(sb.era_array_root)?;
+
+    w.write_u64::<LittleEndian>(sb.metadata_snap)?;
+
+    Ok(())
+}
+
+pub fn write_superblock(engine: &dyn IoEngine, _loc: u64, sb: &Superblock) -> Result<()> {
+    let b = Block::zeroed(SUPERBLOCK_LOCATION);
+
+    // pack the superblock
+    {
+        let mut cursor = Cursor::new(b.get_data());
+        pack_superblock(sb, &mut cursor)?;
+    }
+
+    // calculate the checksum
+    write_checksum(b.get_data(), BT::ERA_SUPERBLOCK)?;
+
+    // write
+    engine.write(&b)?;
+    Ok(())
+}
+
+//------------------------------------------

src/era/writeset.rs

@@ -0,0 +1,35 @@
+use anyhow::Result;
+use byteorder::{LittleEndian, WriteBytesExt};
+use nom::{number::complete::*, IResult};
+
+use crate::pdata::unpack::*;
+
+//------------------------------------------
+
+#[derive(Clone, Copy, Debug)]
+pub struct Writeset {
+    pub nr_bits: u32,
+    pub root: u64,
+}
+
+impl Unpack for Writeset {
+    fn disk_size() -> u32 {
+        12
+    }
+
+    fn unpack(i: &[u8]) -> IResult<&[u8], Writeset> {
+        let (i, nr_bits) = le_u32(i)?;
+        let (i, root) = le_u64(i)?;
+        Ok((i, Writeset { nr_bits, root }))
+    }
+}
+
+impl Pack for Writeset {
+    fn pack<W: WriteBytesExt>(&self, w: &mut W) -> Result<()> {
+        w.write_u32::<LittleEndian>(self.nr_bits)?;
+        w.write_u64::<LittleEndian>(self.root)?;
+        Ok(())
+    }
+}
+
+//------------------------------------------

src/lib.rs

@@ -18,6 +18,7 @@ extern crate quickcheck_macros;
 pub mod cache;
 pub mod checksum;
 pub mod commands;
+pub mod era;
 pub mod file_utils;
 pub mod io_engine;
 pub mod math;