[era_restore (rust)] First code drop

src/bin/pdata_tools.rs

@@ -38,6 +38,8 @@ fn main_() -> Result<()> {
         era_check::run(&new_args);
     } else if name_eq(name, "era_dump") {
         era_dump::run(&new_args);
+    } else if name_eq(name, "era_restore") {
+        era_restore::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_restore.rs

@@ -0,0 +1,64 @@
+extern crate clap;
+
+use clap::{App, Arg};
+use std::path::Path;
+use std::process;
+
+use crate::commands::utils::*;
+use crate::era::restore::{restore, EraRestoreOptions};
+
+pub fn run(args: &[std::ffi::OsString]) {
+    let parser = App::new("era_restore")
+        .version(crate::version::tools_version())
+        .about("Convert XML format metadata to binary.")
+        // 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("QUIET")
+                .help("Suppress output messages, return only exit code.")
+                .short("q")
+                .long("quiet"),
+        )
+        // options
+        .arg(
+            Arg::with_name("INPUT")
+                .help("Specify the input xml")
+                .short("i")
+                .long("input")
+                .value_name("FILE")
+                .required(true),
+        )
+        .arg(
+            Arg::with_name("OUTPUT")
+                .help("Specify the output device to check")
+                .short("o")
+                .long("output")
+                .value_name("FILE")
+                .required(true),
+        );
+
+    let matches = parser.get_matches_from(args);
+    let input_file = Path::new(matches.value_of("INPUT").unwrap());
+    let output_file = Path::new(matches.value_of("OUTPUT").unwrap());
+
+    let report = mk_report(matches.is_present("QUIET"));
+    check_input_file(input_file, &report);
+    check_output_file(output_file, &report);
+
+    let opts = EraRestoreOptions {
+        input: input_file,
+        output: output_file,
+        async_io: matches.is_present("ASYNC_IO"),
+        report: report.clone(),
+    };
+
+    if let Err(reason) = restore(opts) {
+        report.fatal(&format!("{}", reason));
+        process::exit(1);
+    }
+}

src/commands/mod.rs

@@ -4,6 +4,7 @@ pub mod cache_repair;
 pub mod cache_restore;
 pub mod era_check;
 pub mod era_dump;
+pub mod era_restore;
 pub mod thin_check;
 pub mod thin_dump;
 pub mod thin_metadata_pack;

src/era/mod.rs

@@ -1,6 +1,7 @@
 pub mod check;
 pub mod dump;
 pub mod ir;
+pub mod restore;
 pub mod superblock;
 pub mod writeset;
 pub mod xml;

src/era/restore.rs

@@ -0,0 +1,284 @@
+use anyhow::{anyhow, Result};
+
+use std::collections::BTreeMap;
+use std::fs::OpenOptions;
+use std::path::Path;
+use std::sync::Arc;
+
+use crate::era::ir::{self, MetadataVisitor, Visit};
+use crate::era::superblock::*;
+use crate::era::writeset::Writeset;
+use crate::era::xml;
+use crate::io_engine::*;
+use crate::math::*;
+use crate::pdata::array_builder::*;
+use crate::pdata::btree_builder::*;
+use crate::pdata::space_map_common::pack_root;
+use crate::pdata::space_map_metadata::*;
+use crate::report::*;
+use crate::write_batcher::*;
+
+//------------------------------------------
+
+const MAX_CONCURRENT_IO: u32 = 1024;
+
+//------------------------------------------
+
+pub struct EraRestoreOptions<'a> {
+    pub input: &'a Path,
+    pub output: &'a Path,
+    pub async_io: bool,
+    pub report: Arc<Report>,
+}
+
+struct Context {
+    _report: Arc<Report>,
+    engine: Arc<dyn IoEngine + Send + Sync>,
+}
+
+fn mk_context(opts: &EraRestoreOptions) -> anyhow::Result<Context> {
+    let engine: Arc<dyn IoEngine + Send + Sync>;
+
+    if opts.async_io {
+        engine = Arc::new(AsyncIoEngine::new(opts.output, MAX_CONCURRENT_IO, true)?);
+    } else {
+        let nr_threads = std::cmp::max(8, num_cpus::get() * 2);
+        engine = Arc::new(SyncIoEngine::new(opts.output, nr_threads, true)?);
+    }
+
+    Ok(Context {
+        _report: opts.report.clone(),
+        engine,
+    })
+}
+
+//------------------------------------------
+
+#[derive(PartialEq)]
+enum Section {
+    None,
+    Superblock,
+    Writeset,
+    EraArray,
+    Finalized,
+}
+
+pub struct Restorer<'a> {
+    w: &'a mut WriteBatcher,
+    sb: Option<ir::Superblock>,
+    writesets: BTreeMap<u32, Writeset>,
+    writeset_builder: Option<ArrayBuilder<u64>>, // bitset
+    current_writeset: Option<ir::Writeset>,
+    era_array_builder: Option<ArrayBuilder<u32>>,
+    writeset_buf: (u32, u64), // (index in u64 array, value)
+    in_section: Section,
+}
+
+impl<'a> Restorer<'a> {
+    pub fn new(w: &'a mut WriteBatcher) -> Restorer<'a> {
+        Restorer {
+            w,
+            sb: None,
+            writesets: BTreeMap::new(),
+            writeset_builder: None,
+            current_writeset: None,
+            era_array_builder: None,
+            writeset_buf: (0, 0),
+            in_section: Section::None,
+        }
+    }
+
+    fn finalize(&mut self) -> Result<()> {
+        let src_sb;
+        if let Some(sb) = self.sb.take() {
+            src_sb = sb;
+        } else {
+            return Err(anyhow!("not in superblock"));
+        }
+
+        // build the writeset tree
+        let mut tree_builder = BTreeBuilder::<Writeset>::new(Box::new(NoopRC {}));
+        let mut writesets = BTreeMap::<u32, Writeset>::new();
+        std::mem::swap(&mut self.writesets, &mut writesets);
+        for (era, ws) in writesets {
+            tree_builder.push_value(self.w, era as u64, ws)?;
+        }
+        let writeset_tree_root = tree_builder.complete(self.w)?;
+
+        // complete the era array
+        let era_array_root;
+        if let Some(builder) = self.era_array_builder.take() {
+            era_array_root = builder.complete(self.w)?;
+        } else {
+            return Err(anyhow!("internal error. couldn't find era array"));
+        }
+
+        // build metadata space map
+        let metadata_sm_root = build_metadata_sm(self.w)?;
+
+        let sb = Superblock {
+            flags: SuperblockFlags {
+                clean_shutdown: true,
+            },
+            block: SUPERBLOCK_LOCATION,
+            version: 1,
+            metadata_sm_root,
+            data_block_size: src_sb.block_size,
+            nr_blocks: src_sb.nr_blocks,
+            current_era: src_sb.current_era,
+            current_writeset: Writeset {
+                nr_bits: src_sb.nr_blocks,
+                root: 0,
+            },
+            writeset_tree_root,
+            era_array_root,
+            metadata_snap: 0,
+        };
+        write_superblock(self.w.engine.as_ref(), SUPERBLOCK_LOCATION, &sb)?;
+
+        self.in_section = Section::Finalized;
+        Ok(())
+    }
+}
+
+impl<'a> MetadataVisitor for Restorer<'a> {
+    fn superblock_b(&mut self, sb: &ir::Superblock) -> Result<Visit> {
+        if self.in_section != Section::None {
+            return Err(anyhow!("duplicated superblock"));
+        }
+
+        self.sb = Some(sb.clone());
+        let b = self.w.alloc()?;
+        if b.loc != SUPERBLOCK_LOCATION {
+            return Err(anyhow!("superblock was occupied"));
+        }
+
+        self.writeset_builder = None;
+        self.era_array_builder = Some(ArrayBuilder::new(sb.nr_blocks as u64));
+        self.in_section = Section::Superblock;
+
+        Ok(Visit::Continue)
+    }
+
+    fn superblock_e(&mut self) -> Result<Visit> {
+        self.finalize()?;
+        Ok(Visit::Continue)
+    }
+
+    fn writeset_b(&mut self, ws: &ir::Writeset) -> Result<Visit> {
+        if self.in_section != Section::Superblock {
+            return Err(anyhow!("not in superblock"));
+        }
+        self.writeset_builder = Some(ArrayBuilder::new(div_up(ws.nr_bits as u64, 64)));
+        self.writeset_buf.0 = 0;
+        self.writeset_buf.1 = 0;
+        self.current_writeset = Some(ws.clone());
+        self.in_section = Section::Writeset;
+        Ok(Visit::Continue)
+    }
+
+    fn writeset_e(&mut self) -> Result<Visit> {
+        if self.in_section != Section::Writeset {
+            return Err(anyhow!("not in writeset"));
+        }
+
+        if let Some(mut builder) = self.writeset_builder.take() {
+            if let Some(ws) = self.current_writeset.take() {
+                // push the trailing bits
+                builder.push_value(self.w, self.writeset_buf.0 as u64, self.writeset_buf.1)?;
+
+                let root = builder.complete(self.w)?;
+                self.writesets.insert(
+                    ws.era,
+                    Writeset {
+                        root,
+                        nr_bits: ws.nr_bits,
+                    },
+                );
+                self.in_section = Section::Superblock;
+            } else {
+                return Err(anyhow!("internal error. couldn't find writeset"));
+            }
+        } else {
+            return Err(anyhow!("internal error. couldn't find writeset"));
+        }
+
+        Ok(Visit::Continue)
+    }
+
+    fn writeset_bit(&mut self, wbit: &ir::WritesetBit) -> Result<Visit> {
+        if wbit.value {
+            let index = wbit.block >> 6;
+            let mask = 1 << (wbit.block & 63);
+            if index == self.writeset_buf.0 {
+                self.writeset_buf.1 |= mask;
+            } else {
+                let builder = self.writeset_builder.as_mut().unwrap();
+                builder.push_value(self.w, self.writeset_buf.0 as u64, self.writeset_buf.1)?;
+                self.writeset_buf.0 = index;
+                self.writeset_buf.1 = mask;
+            }
+        }
+
+        Ok(Visit::Continue)
+    }
+
+    fn era_b(&mut self) -> Result<Visit> {
+        if self.in_section != Section::Superblock {
+            return Err(anyhow!("not in superblock"));
+        }
+        self.in_section = Section::EraArray;
+        Ok(Visit::Continue)
+    }
+
+    fn era_e(&mut self) -> Result<Visit> {
+        if self.in_section != Section::EraArray {
+            return Err(anyhow!("not in era array"));
+        }
+        self.in_section = Section::Superblock;
+        Ok(Visit::Continue)
+    }
+
+    fn era(&mut self, era: &ir::Era) -> Result<Visit> {
+        let builder = self.era_array_builder.as_mut().unwrap();
+        builder.push_value(self.w, era.block as u64, era.era)?;
+        Ok(Visit::Continue)
+    }
+
+    fn eof(&mut self) -> Result<Visit> {
+        if self.in_section != Section::Finalized {
+            return Err(anyhow!("incompleted source metadata"));
+        }
+        Ok(Visit::Continue)
+    }
+}
+
+//------------------------------------------
+
+fn build_metadata_sm(w: &mut WriteBatcher) -> Result<Vec<u8>> {
+    let r = write_metadata_sm(w)?;
+    let sm_root = pack_root(&r, SPACE_MAP_ROOT_SIZE)?;
+
+    Ok(sm_root)
+}
+
+//------------------------------------------
+
+pub fn restore(opts: EraRestoreOptions) -> Result<()> {
+    let input = OpenOptions::new()
+        .read(true)
+        .write(false)
+        .open(opts.input)?;
+
+    let ctx = mk_context(&opts)?;
+
+    let sm = core_metadata_sm(ctx.engine.get_nr_blocks(), u32::MAX);
+    let mut w = WriteBatcher::new(ctx.engine.clone(), sm.clone(), ctx.engine.get_batch_size());
+
+    let mut restorer = Restorer::new(&mut w);
+    xml::read(input, &mut restorer)?;
+
+    Ok(())
+}
+
+//------------------------------------------

src/era/xml.rs

@@ -1,7 +1,8 @@
-use anyhow::Result;
+use anyhow::{anyhow, Result};
 use quick_xml::events::{BytesEnd, BytesStart, Event};
-use quick_xml::Writer;
-use std::io::Write;
+use quick_xml::{Reader, Writer};
+use std::io::{BufRead, BufReader};
+use std::io::{Read, Write};
 
 use crate::era::ir::*;
 use crate::xml::*;
@@ -91,3 +92,146 @@ impl<W: Write> MetadataVisitor for XmlWriter<W> {
 }
 
 //------------------------------------------
+
+fn parse_superblock(e: &BytesStart) -> Result<Superblock> {
+    let tag = "superblock";
+    let mut uuid: Option<String> = None;
+    let mut block_size: Option<u32> = None;
+    let mut nr_blocks: Option<u32> = None;
+    let mut current_era: Option<u32> = None;
+
+    for a in e.attributes() {
+        let kv = a.unwrap();
+        match kv.key {
+            b"uuid" => uuid = Some(string_val(&kv)),
+            b"block_size" => block_size = Some(u32_val(&kv)?),
+            b"nr_blocks" => nr_blocks = Some(u32_val(&kv)?),
+            b"current_era" => current_era = Some(u32_val(&kv)?),
+            _ => return bad_attr(tag, kv.key),
+        }
+    }
+
+    Ok(Superblock {
+        uuid: check_attr(tag, "uuid", uuid)?,
+        block_size: check_attr(tag, "block_size", block_size)?,
+        nr_blocks: check_attr(tag, "nr_cache_blocks", nr_blocks)?,
+        current_era: check_attr(tag, "current_era", current_era)?,
+    })
+}
+
+fn parse_writeset(e: &BytesStart) -> Result<Writeset> {
+    let tag = "writeset";
+    let mut era: Option<u32> = None;
+    let mut nr_bits: Option<u32> = None;
+
+    for a in e.attributes() {
+        let kv = a.unwrap();
+        match kv.key {
+            b"era" => era = Some(u32_val(&kv)?),
+            b"nr_bits" => nr_bits = Some(u32_val(&kv)?),
+            _ => return bad_attr(tag, kv.key),
+        }
+    }
+
+    Ok(Writeset {
+        era: check_attr(tag, "era", era)?,
+        nr_bits: check_attr(tag, "nr_bits", nr_bits)?,
+    })
+}
+
+fn parse_writeset_bit(e: &BytesStart) -> Result<WritesetBit> {
+    let tag = "bit";
+    let mut block: Option<u32> = None;
+    let mut value: Option<bool> = None;
+
+    for a in e.attributes() {
+        let kv = a.unwrap();
+        match kv.key {
+            b"block" => block = Some(u32_val(&kv)?),
+            b"value" => value = Some(bool_val(&kv)?),
+            _ => return bad_attr(tag, kv.key),
+        }
+    }
+
+    Ok(WritesetBit {
+        block: check_attr(tag, "block", block)?,
+        value: check_attr(tag, "value", value)?,
+    })
+}
+
+fn parse_era(e: &BytesStart) -> Result<Era> {
+    let tag = "era";
+    let mut block: Option<u32> = None;
+    let mut era: Option<u32> = None;
+
+    for a in e.attributes() {
+        let kv = a.unwrap();
+        match kv.key {
+            b"block" => block = Some(u32_val(&kv)?),
+            b"era" => era = Some(u32_val(&kv)?),
+            _ => return bad_attr(tag, kv.key),
+        }
+    }
+
+    Ok(Era {
+        block: check_attr(tag, "block", block)?,
+        era: check_attr(tag, "era", era)?,
+    })
+}
+
+fn handle_event<R, M>(reader: &mut Reader<R>, buf: &mut Vec<u8>, visitor: &mut M) -> Result<Visit>
+where
+    R: Read + BufRead,
+    M: MetadataVisitor,
+{
+    match reader.read_event(buf) {
+        Ok(Event::Start(ref e)) => match e.name() {
+            b"superblock" => visitor.superblock_b(&parse_superblock(e)?),
+            b"writeset" => visitor.writeset_b(&parse_writeset(e)?),
+            b"era_array" => visitor.era_b(),
+            _ => return Err(anyhow!("Parse error at byte {}", reader.buffer_position())),
+        },
+        Ok(Event::End(ref e)) => match e.name() {
+            b"superblock" => visitor.superblock_e(),
+            b"writeset" => visitor.writeset_e(),
+            b"era_array" => visitor.era_e(),
+            _ => return Err(anyhow!("Parse error at byte {}", reader.buffer_position())),
+        },
+        Ok(Event::Empty(ref e)) => match e.name() {
+            b"bit" => visitor.writeset_bit(&parse_writeset_bit(e)?),
+            b"era" => visitor.era(&parse_era(e)?),
+            _ => return Err(anyhow!("Parse error at byte {}", reader.buffer_position())),
+        },
+        Ok(Event::Text(_)) => Ok(Visit::Continue),
+        Ok(Event::Comment(_)) => Ok(Visit::Continue),
+        Ok(Event::Eof) => {
+            visitor.eof()?;
+            Ok(Visit::Stop)
+        }
+        Ok(_) => return Err(anyhow!("Parse error at byte {}", reader.buffer_position())),
+        Err(e) => {
+            return Err(anyhow!(
+                "Parse error at byte {}: {:?}",
+                reader.buffer_position(),
+                e
+            ))
+        }
+    }
+}
+
+pub fn read<R, M>(input: R, visitor: &mut M) -> Result<()>
+where
+    R: Read,
+    M: MetadataVisitor,
+{
+    let input = BufReader::new(input);
+    let mut reader = Reader::from_reader(input);
+
+    reader.trim_text(true);
+    let mut buf = Vec::new();
+
+    while let Visit::Continue = handle_event(&mut reader, &mut buf, visitor)? {}
+    Ok(())
+}
+
+//------------------------------------------