Merge pull request #22 from urbit/philip/persist

[ares] Add jam-based snapshots

hoon/scaffolding/baby.hoon

@@ -0,0 +1,48 @@
+::  A trivial working pill which requires no jets
+!.
+=>  ~
+=/  core
+  !=
+  =>  ~  =>
+  |%
+  +$  card  (cask)
+  ++  cask  |$  [a]  (pair mark a)
+  +$  knot  @ta
+  ++  list  |$  [item]  $@(~ [i=item t=(list item)])
+  +$  mark  @tas
+  +$  ovum  [=wire =card]
+  ++  pair  |$  [head tail]  [p=head q=tail]
+  +$  path  (list knot)
+  +$  wire  path
+  --  =>
+  ::
+  |%
+  ++  load  !!
+  ++  peek  _~
+  ++  wish  !!
+  ++  poke
+    |=  [now=@da ovo=ovum]
+    ^-  ^
+    :: ~>  %slog.[0 'got']
+    :: ~>  %slog.[0 -.card.ovo]
+    =/  fec  [//term/1 %blit [%put "effect"] [%nel ~] ~]
+    [[fec ~] ..poke]
+  --
+  ::
+  |=  [now=@da ovo=ovum]
+  ^-  *
+  .(+> +:(poke now ovo))
+::
+|%
+++  aeon
+  ^-  *
+  =>  *[arvo=* epic=*]
+  !=
+  |-  ^-  *
+  ?@  epic  arvo
+  %=  $
+    epic  +.epic
+    arvo  .*([arvo -.epic] [%9 2 %10 [6 %0 3] %0 2])
+  ==
+--
+[%pill %baby [aeon .*(0 core) ~] ~ ~]

rust/ares/src/interpreter.rs

@@ -2,6 +2,7 @@ use self::NockWork::*;
 use crate::jets;
 use crate::mem::unifying_equality;
 use crate::mem::NockStack;
+use crate::newt::Newt;
 use crate::noun::{Atom, Cell, DirectAtom, IndirectAtom, Noun};
 use ares_macros::tas;
 use bitvec::prelude::{BitSlice, Lsb0};
@@ -64,7 +65,13 @@ fn noun_to_work(noun: Noun) -> NockWork {
     }
 }
 
-pub fn interpret(stack: &mut NockStack, mut subject: Noun, formula: Noun) -> Noun {
+/** Interpret nock */
+pub fn interpret(
+    stack: &mut NockStack,
+    newt: &mut Option<&mut Newt>, // For printing slogs; if None, print to stdout
+    mut subject: Noun,
+    formula: Noun,
+) -> Noun {
     let mut res = unsafe { DirectAtom::new_unchecked(0).as_atom().as_noun() };
     stack.push(1);
     unsafe {
@@ -314,30 +321,27 @@ pub fn interpret(stack: &mut NockStack, mut subject: Noun, formula: Noun) -> Nou
             Nock11ComputeHint => unsafe {
                 let hint = *stack.local_noun_pointer(1);
                 if let Ok(hint_cell) = hint.as_cell() {
-                    // match %sham hints, which are scaffolding until we have a real jet dashboard
-                    if hint_cell
-                        .head()
-                        .raw_equals(DirectAtom::new_unchecked(tas!(b"sham")).as_noun())
-                    {
-                        if let Ok(jet_formula) = hint_cell.tail().as_cell() {
-                            let jet_name = jet_formula.tail();
-                            if let Ok(jet) = jets::get_jet(jet_name) {
-                                res = jet(stack, subject);
-                                stack.pop(&mut res);
-                                continue;
-                            }
-                        }
+                    if let Ok(found) = match_pre_hint(stack, subject, hint_cell) {
+                        res = found;
+                        stack.pop(&mut res);
+                    } else {
+                        *(stack.local_noun_pointer(0)) = work_to_noun(Nock11ComputeResult);
+                        push_formula(stack, hint_cell.tail());
                     }
-                    *(stack.local_noun_pointer(0)) = work_to_noun(Nock11ComputeResult);
-                    push_formula(stack, hint_cell.tail());
                 } else {
                     panic!("IMPOSSIBLE: tried to compute a dynamic hint but hint is an atom");
                 }
             },
             Nock11ComputeResult => unsafe {
-                *(stack.local_noun_pointer(0)) = work_to_noun(Nock11Done);
-                let formula = *stack.local_noun_pointer(2);
-                push_formula(stack, formula);
+                let hint = *stack.local_noun_pointer(1);
+                if let Ok(found) = match_post_hint(stack, newt, subject, hint, res) {
+                    res = found;
+                    stack.pop(&mut res);
+                } else {
+                    *(stack.local_noun_pointer(0)) = work_to_noun(Nock11Done);
+                    let formula = *stack.local_noun_pointer(2);
+                    push_formula(stack, formula);
+                }
             },
             Nock11Done => {
                 stack.pop(&mut res);
@@ -517,7 +521,7 @@ fn push_formula(stack: &mut NockStack, formula: Noun) {
             }
         }
     } else {
-        panic!("Bad formula: atoms are not formulas");
+        panic!("Bad formula: atoms are not formulas: {:?}", formula);
     }
 }
 
@@ -619,3 +623,42 @@ fn inc(stack: &mut NockStack, atom: Atom) -> Atom {
         }
     }
 }
+
+/** Match hints which apply before the formula is evaluated */
+fn match_pre_hint(stack: &mut NockStack, subject: Noun, cell: Cell) -> Result<Noun, ()> {
+    let direct = cell.head().as_direct()?;
+    match direct.data() {
+        // %sham hints are scaffolding until we have a real jet dashboard
+        tas!(b"sham") => {
+            let jet_formula = cell.tail().as_cell()?;
+            let jet = jets::get_jet(jet_formula.tail())?;
+            return Ok(jet(stack, subject));
+        }
+        _ => Err(()),
+    }
+}
+
+/** Match static hints and dynamic hints after they're evaluated */
+fn match_post_hint(
+    stack: &mut NockStack,
+    newt: &mut Option<&mut Newt>,
+    _subject: Noun,
+    hint: Noun,
+    res: Noun,
+) -> Result<Noun, ()> {
+    let direct = hint.as_cell()?.head().as_direct()?;
+    match direct.data() {
+        tas!(b"slog") => {
+            let slog_cell = res.as_cell()?;
+            let pri = slog_cell.head().as_direct()?.data();
+            let tank = slog_cell.tail();
+            if let Some(not) = newt {
+                not.slog(stack, pri, tank);
+            } else {
+                println!("slog: {:?} {:?}", pri, tank);
+            }
+            Err(())
+        }
+        _ => Err(()),
+    }
+}

rust/ares/src/lib.rs

@@ -4,8 +4,11 @@ pub mod interpreter;
 pub mod jets;
 pub mod mem;
 pub mod mug;
+pub mod newt;
 pub mod noun;
+pub mod serf;
 pub mod serialization;
+pub mod snapshot;
 
 #[cfg(test)]
 mod tests {

rust/ares/src/main.rs

@@ -1,6 +1,7 @@
 use ares::interpreter::interpret;
 use ares::mem::NockStack;
 use ares::noun::IndirectAtom;
+use ares::serf::serf;
 use ares::serialization::{cue, jam};
 use memmap::Mmap;
 use memmap::MmapMut;
@@ -14,6 +15,10 @@ use std::ptr::write_bytes;
 
 fn main() -> io::Result<()> {
     let filename = env::args().nth(1).expect("Must provide input filename");
+    if filename == "serf" {
+        return serf();
+    }
+
     let output_filename = format!("{}.out", filename.clone());
     let f = File::open(filename)?;
     let in_len = f.metadata()?.len();
@@ -31,7 +36,7 @@ fn main() -> io::Result<()> {
     let input_cell = input
         .as_cell()
         .expect("Input must be jam of subject/formula pair");
-    let result = interpret(&mut stack, input_cell.head(), input_cell.tail());
+    let result = interpret(&mut stack, &mut None, input_cell.head(), input_cell.tail());
     if let Ok(atom) = result.as_atom() {
         println!("Result: {:?}", atom);
     }

rust/ares/src/mem.rs

@@ -661,12 +661,8 @@ pub unsafe fn unifying_equality(stack: &mut NockStack, a: *mut Noun, b: *mut Nou
         return true;
     };
     // If the nouns have cached mugs which are disequal we have nothing to do
-    if let (Ok(a_alloc), Ok(b_alloc)) = (
-        (*a).as_allocated(),
-        (*b).as_allocated(),
-    ) {
-        if let (Some(a_mug), Some(b_mug)) = (a_alloc.get_cached_mug(), b_alloc.get_cached_mug())
-        {
+    if let (Ok(a_alloc), Ok(b_alloc)) = ((*a).as_allocated(), (*b).as_allocated()) {
+        if let (Some(a_mug), Some(b_mug)) = (a_alloc.get_cached_mug(), b_alloc.get_cached_mug()) {
             if a_mug != b_mug {
                 return false;
             };

rust/ares/src/newt.rs

@@ -0,0 +1,219 @@
+/** Newt: IPC to the king
+ *
+ * This manages an IPC connection to the king over stdin and stdout.  The protocol is jammed nouns,
+ * with the following schema:
+ *
+ * |%
+ * ::  +writ: from king to serf
+ * ::
+ * +$  writ
+ *   $%  $:  %live
+ *           $%  [%cram eve=@]
+ *               [%exit cod=@]
+ *               [%save eve=@]
+ *               [%meld ~]
+ *               [%pack ~]
+ *       ==  ==
+ *       [%peek mil=@ sam=*]  :: gang (each path $%([%once @tas @tas path] [%beam @tas beam]))
+ *       [%play eve=@ lit=(list ?((pair @da ovum) *))]
+ *       [%work mil=@ job=(pair @da ovum)]
+ *   ==
+ * ::  +plea: from serf to king
+ * ::
+ * +$  plea
+ *   $%  [%live ~]
+ *       [%ripe [pro=%1 hon=@ nok=@] eve=@ mug=@]
+ *       [%slog pri=@ tank]
+ *       [%flog cord]
+ *       $:  %peek
+ *           $%  [%done dat=(unit (cask))]
+ *               [%bail dud=goof]
+ *       ==  ==
+ *       $:  %play
+ *           $%  [%done mug=@]
+ *               [%bail eve=@ mug=@ dud=goof]
+ *       ==  ==
+ *       $:  %work
+ *           $%  [%done eve=@ mug=@ fec=(list ovum)]
+ *               [%swap eve=@ mug=@ job=(pair @da ovum) fec=(list ovum)]
+ *               [%bail lud=(list goof)]
+ *       ==  ==
+ *   ==
+ * --
+ *
+ * NB: stdin and stdout are generally buffered, and there's no officially supported way to work
+ * around that: https://github.com/rust-lang/rust/issues/58326.
+ *
+ * We use stdin and stdout with File::from_raw_fd(0) and File::from_raw_fd(1), which seems to get
+ * around this.  We tested that using io::Stdout requires flushing while this doesn't, but we
+ * haven't tested the same for stdin.
+ *
+ * It's important to not use io::Stdin and io::Stdout directly.  All printfs should use stderr.
+ */
+use crate::mem::NockStack;
+use crate::noun::{IndirectAtom, Noun, D, T};
+use crate::serialization::{cue, jam};
+use ares_macros::tas;
+use either::Either;
+use std::io::{Read, Write};
+use std::os::unix::prelude::FromRawFd;
+use std::ptr::copy_nonoverlapping;
+
+pub struct Newt {
+    input: std::fs::File,
+    output: std::fs::File,
+}
+
+impl Newt {
+    pub fn new() -> Newt {
+        Newt {
+            input: unsafe { std::fs::File::from_raw_fd(0) },
+            output: unsafe { std::fs::File::from_raw_fd(1) },
+        }
+    }
+
+    /** Write a noun to the newt.
+     *
+     * NB: we write 64-bit words, while vere writes bytes.  The extra zero bytes shouldn't be a
+     * problem.
+     */
+    fn write_noun(&mut self, stack: &mut NockStack, noun: Noun) {
+        let atom = jam(stack, noun);
+        let size = atom.size() << 3;
+        let mut buf = vec![0 as u8; size + 5];
+        buf[1] = size as u8;
+        buf[2] = (size >> 8) as u8;
+        buf[3] = (size >> 16) as u8;
+        buf[4] = (size >> 24) as u8;
+        match atom.as_either() {
+            Either::Left(direct) => unsafe {
+                copy_nonoverlapping(
+                    &direct.data() as *const u64 as *const u8,
+                    buf.as_mut_ptr().add(5) as *mut u8,
+                    size,
+                );
+            },
+            Either::Right(indirect) => unsafe {
+                // REVIEW: is this safe/the right way to do this?
+                copy_nonoverlapping(
+                    indirect.data_pointer() as *const u8,
+                    buf.as_mut_ptr().add(5) as *mut u8,
+                    size,
+                );
+            },
+        };
+        self.output.write_all(&buf).unwrap();
+    }
+
+    /** Send %ripe, the first event. */
+    pub fn ripe(&mut self, stack: &mut NockStack, eve: u64, mug: u64) {
+        let version = T(
+            stack,
+            &[
+                D(1),   // newt protocol
+                D(139), // hoon kelvin
+                D(4),   // nock kelvin
+            ],
+        );
+        let ripe = T(stack, &[D(tas!(b"ripe")), version, D(eve), D(mug)]);
+        self.write_noun(stack, ripe);
+    }
+
+    /** Send %live, acknowledging. */
+    pub fn live(&mut self, stack: &mut NockStack) {
+        let live = T(stack, &[D(tas!(b"live")), D(0)]);
+        self.write_noun(stack, live);
+    }
+
+    /** Send %slog, pretty-printed debug output. */
+    pub fn slog(&mut self, stack: &mut NockStack, pri: u64, tank: Noun) {
+        let slog = T(stack, &[D(tas!(b"slog")), D(pri), tank]);
+        self.write_noun(stack, slog);
+    }
+
+    /** Send %flog, raw debug output. */
+    pub fn flog(&mut self, stack: &mut NockStack, cord: Noun) {
+        let flog = T(stack, &[D(tas!(b"flog")), cord]);
+        self.write_noun(stack, flog);
+    }
+
+    /** Send %peek %done, successfully scried. */
+    pub fn peek_done(&mut self, stack: &mut NockStack, dat: Noun) {
+        let peek = T(stack, &[D(tas!(b"peek")), D(tas!(b"done")), dat]);
+        self.write_noun(stack, peek);
+    }
+
+    /** Send %peek %bail, unsuccessfully scried. */
+    pub fn peek_bail(&mut self, stack: &mut NockStack, dud: Noun) {
+        let peek = T(stack, &[D(tas!(b"peek")), D(tas!(b"bail")), dud]);
+        self.write_noun(stack, peek);
+    }
+
+    /** Send %play %done, successfully replayed events. */
+    pub fn play_done(&mut self, stack: &mut NockStack, mug: u64) {
+        let play = T(stack, &[D(tas!(b"play")), D(tas!(b"done")), D(mug)]);
+        self.write_noun(stack, play);
+    }
+
+    /** Send %play %bail, failed to replay events. */
+    pub fn play_bail(&mut self, stack: &mut NockStack, eve: u64, mug: u64, dud: Noun) {
+        let play = T(
+            stack,
+            &[D(tas!(b"play")), D(tas!(b"bail")), D(eve), D(mug), dud],
+        );
+        self.write_noun(stack, play);
+    }
+
+    /** Send %work %done, successfully ran event. */
+    pub fn work_done(&mut self, stack: &mut NockStack, eve: u64, mug: u64, fec: Noun) {
+        let work = T(
+            stack,
+            &[D(tas!(b"work")), D(tas!(b"done")), D(eve), D(mug), fec],
+        );
+        self.write_noun(stack, work);
+    }
+
+    /** Send %work %swap, successfully replaced failed event. */
+    pub fn work_swap(&mut self, stack: &mut NockStack, eve: u64, mug: u64, job: Noun, fec: Noun) {
+        let work = T(
+            stack,
+            &[D(tas!(b"work")), D(tas!(b"swap")), D(eve), D(mug), job, fec],
+        );
+        self.write_noun(stack, work);
+    }
+
+    /** Send %work %bail, failed to run event. */
+    pub fn work_bail(&mut self, stack: &mut NockStack, lud: Noun) {
+        let work = T(stack, &[D(tas!(b"work")), D(tas!(b"bail")), lud]);
+        self.write_noun(stack, work);
+    }
+
+    /** Fetch next message. */
+    pub fn next(&mut self, stack: &mut NockStack) -> Option<Noun> {
+        let mut header: Vec<u8> = Vec::with_capacity(5);
+        header.resize(5, 0);
+        if let Err(err) = self.input.read_exact(&mut header) {
+            if err.kind() == std::io::ErrorKind::UnexpectedEof {
+                return None;
+            } else {
+                panic!("Error reading header: {}", err);
+            }
+        }
+
+        let byte_len = u32::from_le_bytes([header[1], header[2], header[3], header[4]]) as usize;
+
+        let atom = unsafe {
+            let (mut atom, dest) = IndirectAtom::new_raw_mut_bytes(stack, byte_len);
+            if let Err(err) = self.input.read_exact(dest) {
+                if err.kind() == std::io::ErrorKind::UnexpectedEof {
+                    return None;
+                } else {
+                    panic!("Error reading body: {}", err);
+                }
+            }
+            atom.normalize_as_atom()
+        };
+
+        Some(cue(stack, atom))
+    }
+}

rust/ares/src/noun.rs

@@ -131,7 +131,7 @@ impl DirectAtom {
         Atom { direct: self }
     }
 
-    pub fn as_noun(self) -> Noun {
+    pub const fn as_noun(self) -> Noun {
         Noun { direct: self }
     }
 
@@ -150,6 +150,16 @@ impl fmt::Debug for DirectAtom {
     }
 }
 
+#[allow(non_snake_case)]
+pub const fn D(n: u64) -> Noun {
+    DirectAtom::new_panic(n).as_noun()
+}
+
+#[allow(non_snake_case)]
+pub fn T(allocator: &mut dyn NounAllocator, tup: &[Noun]) -> Noun {
+    Cell::new_tuple(allocator, tup).as_noun()
+}
+
 /** An indirect atom.
  *
  * Indirect atoms represent atoms above DIRECT_MAX as a tagged pointer to a memory buffer
@@ -251,6 +261,20 @@ impl IndirectAtom {
         )
     }
 
+    /** Make an indirect atom that can be written into as a slice of bytes. The constraints of
+     * [new_raw_mut_zeroed] also apply here
+     *
+     * Note: size is bytes, not words
+     */
+    pub unsafe fn new_raw_mut_bytes<'a>(
+        allocator: &mut dyn NounAllocator,
+        size: usize,
+    ) -> (Self, &'a mut [u8]) {
+        let word_size = (size + 7) << 3;
+        let (noun, ptr) = Self::new_raw_mut_zeroed(allocator, word_size);
+        (noun, from_raw_parts_mut(ptr as *mut u8, size))
+    }
+
     /** Size of an indirect atom in 64-bit words */
     pub fn size(&self) -> usize {
         unsafe { *(self.to_raw_pointer().add(1)) as usize }
@@ -387,6 +411,19 @@ impl Cell {
         }
     }
 
+    pub fn new_tuple(allocator: &mut dyn NounAllocator, tup: &[Noun]) -> Cell {
+        if tup.len() < 2 {
+            panic!("Cannot create tuple with fewer than 2 elements");
+        }
+
+        let len = tup.len();
+        let mut cell = Cell::new(allocator, tup[len - 2], tup[len - 1]);
+        for i in (0..len - 2).rev() {
+            cell = Cell::new(allocator, tup[i], cell.as_noun());
+        }
+        cell
+    }
+
     pub unsafe fn new_raw_mut(allocator: &mut dyn NounAllocator) -> (Cell, *mut CellMemory) {
         let memory = allocator.alloc_cell();
         (*memory).metadata = 0;

rust/ares/src/serf.rs

@@ -0,0 +1,128 @@
+use crate::interpreter::{interpret, raw_slot};
+use crate::mem::NockStack;
+use crate::mug::mug_u32;
+use crate::newt::Newt;
+use crate::noun::{Noun, D, T};
+use crate::snapshot::{load, save};
+use ares_macros::tas;
+use std::fs::create_dir_all;
+use std::io;
+use std::path::PathBuf;
+
+#[allow(dead_code)]
+const LOAD_AXIS: u64 = 4;
+const PEEK_AXIS: u64 = 22;
+const POKE_AXIS: u64 = 23;
+#[allow(dead_code)]
+const WISH_AXIS: u64 = 10;
+
+/**
+ * This is suitable for talking to the king process.  To test, change the arg_c[0] line in
+ * u3_lord_init in vere to point at this binary and start vere like normal.
+ */
+pub fn serf() -> io::Result<()> {
+    let snap_path_string = std::env::args()
+        .nth(2)
+        .ok_or(io::Error::new(io::ErrorKind::Other, "no pier path"))?;
+    let mut snap_path = PathBuf::from(snap_path_string);
+    snap_path.push(".urb");
+    snap_path.push("chk");
+    create_dir_all(&snap_path)?;
+
+    let ref mut stack = NockStack::new(8 << 10 << 10, 0);
+    let ref mut newt = Newt::new();
+    let mut event_number;
+    let mut arvo;
+
+    (event_number, arvo) = load(stack, snap_path.clone()).unwrap_or((0, D(0)));
+    let mug = mug_u32(stack, arvo);
+
+    newt.ripe(stack, event_number, mug as u64);
+
+    // Can't use for loop because it borrows newt
+    loop {
+        let writ = if let Some(writ) = newt.next(stack) {
+            writ
+        } else {
+            break;
+        };
+
+        let tag = raw_slot(writ, 2).as_direct().unwrap();
+        match tag.data() {
+            tas!(b"live") => {
+                let inner = raw_slot(writ, 6);
+                match inner.as_direct().unwrap().data() {
+                    tas!(b"cram") => eprintln!("cram"),
+                    tas!(b"exit") => eprintln!("exit"),
+                    tas!(b"save") => {
+                        // XX what is eve for?
+                        eprintln!("save");
+                        save(stack, snap_path.clone(), event_number, arvo);
+                    }
+                    tas!(b"meld") => eprintln!("meld"),
+                    tas!(b"pack") => eprintln!("pack"),
+                    _ => eprintln!("unknown live"),
+                }
+                newt.live(stack);
+            }
+            tas!(b"peek") => {
+                let sam = raw_slot(writ, 7);
+                let res = slam(stack, newt, arvo, PEEK_AXIS, sam);
+                newt.peek_done(stack, res);
+            }
+            tas!(b"play") => {
+                let run = if event_number == 0 {
+                    // apply lifecycle to first batch
+                    let lit = raw_slot(writ, 7);
+                    let sub = T(stack, &[D(0), D(3)]);
+                    let lyf = T(stack, &[D(2), sub, D(0), D(2)]);
+                    let gat = interpret(stack, &mut Some(newt), lit, lyf);
+                    arvo = raw_slot(gat, 7);
+                    false
+                } else {
+                    true
+                };
+
+                // do we need to assert something here?
+                // event_number = raw_slot(writ, 6).as_direct().unwrap().data();
+
+                let mut lit = raw_slot(writ, 7);
+                loop {
+                    if let Ok(cell) = lit.as_cell() {
+                        if run {
+                            let ovo = cell.head();
+                            let res = slam(stack, newt, arvo, POKE_AXIS, ovo).as_cell().unwrap();
+                            arvo = res.tail();
+                        }
+                        event_number += 1;
+                        lit = cell.tail();
+                    } else {
+                        break;
+                    }
+                }
+                newt.play_done(stack, 0);
+            }
+            tas!(b"work") => {
+                let ovo = raw_slot(writ, 7);
+                let res = slam(stack, newt, arvo, POKE_AXIS, ovo).as_cell().unwrap();
+                let fec = res.head();
+                arvo = res.tail();
+
+                event_number += 1;
+
+                newt.work_done(stack, event_number, 0, fec);
+            }
+            _ => panic!("got message with unknown tag {:?}", tag),
+        };
+    }
+
+    Ok(())
+}
+
+pub fn slam(stack: &mut NockStack, newt: &mut Newt, core: Noun, axis: u64, ovo: Noun) -> Noun {
+    let pul = T(stack, &[D(9), D(axis), D(0), D(2)]);
+    let sam = T(stack, &[D(6), D(0), D(7)]);
+    let fol = T(stack, &[D(8), pul, D(9), D(2), D(10), sam, D(0), D(2)]);
+    let sub = T(stack, &[core, ovo]);
+    interpret(stack, &mut Some(newt), sub, fol)
+}

rust/ares/src/snapshot.rs

@@ -0,0 +1,154 @@
+/** Jam-based snapshotting
+ *
+ * This is a simple checkpoint system that should be safe but has (very) poor performance.  This is
+ * intended as a working placeholder until the real PMA is hooked up.
+ *
+ * This keeps two files, .urb/chk/snapshot-0.jam and .urbit/chk/snapshot-1.jam.  Each of these
+ * contains 64 bits for a mug checksum, then 64 bits for the event number, then a jam of the state.
+ * We alternate between writing these two files, so that at least one is always valid.
+ *
+ * When we start up, we read both files and pick the one with the higher event number.  If either
+ * is corrupted, we use the other.
+ */
+use crate::mem::NockStack;
+use crate::mug::mug_u32;
+use crate::noun::{IndirectAtom, Noun};
+use crate::serialization::{cue, jam};
+use either::Either;
+use memmap::Mmap;
+use memmap::MmapMut;
+use std::fs::{File, OpenOptions};
+use std::io;
+use std::mem;
+use std::path::PathBuf;
+use std::ptr::copy_nonoverlapping;
+use std::ptr::write_bytes;
+
+pub fn save(stack: &mut NockStack, mut snap_path: PathBuf, event_number: u64, arvo: Noun) {
+    // Find the latest valid snapshot, and write to the other file.
+    let prev_snap = if let Ok((prev_snap, _, _)) = latest_snapshot(stack, snap_path.clone()) {
+        prev_snap
+    } else {
+        0
+    };
+    let snap_number = if prev_snap == 0 { 1 } else { 0 };
+    snap_path.push(format!("snapshot-{}.snap", snap_number));
+
+    let jammed_arvo = jam(stack, arvo);
+    let state = unsafe {
+        let (mut state, dest) = IndirectAtom::new_raw_mut(stack, jammed_arvo.size() + 1);
+        dest.write(event_number);
+        match jammed_arvo.as_either() {
+            Either::Left(direct) => {
+                copy_nonoverlapping(&direct.data() as *const u64, dest.add(1), 1);
+            }
+            Either::Right(indirect) => {
+                copy_nonoverlapping(indirect.data_pointer(), dest.add(1), jammed_arvo.size());
+            }
+        };
+        state.normalize_as_atom()
+    };
+
+    let mugged = mug_u32(stack, state.as_noun());
+
+    let f = OpenOptions::new()
+        .read(true)
+        .write(true)
+        .create(true)
+        .open(snap_path)
+        .unwrap();
+
+    f.set_len(((state.size() + 1) << 3) as u64).unwrap();
+    unsafe {
+        let mut out_map = MmapMut::map_mut(&f).unwrap();
+        let out_ptr = out_map.as_mut_ptr();
+        out_ptr.add(0).write(mugged as u8);
+        out_ptr.add(1).write((mugged >> 8) as u8);
+        out_ptr.add(2).write((mugged >> 16) as u8);
+        out_ptr.add(3).write((mugged >> 24) as u8);
+        copy_nonoverlapping(
+            state.data_pointer() as *mut u8,
+            out_ptr.add(8),
+            state.size() << 3,
+        );
+        out_map.flush().unwrap();
+
+        // This appears to match c3/portable.h: fdatasync for linux, fcntl with F_FULLFSYNC for for
+        // macos, and fsync for some other platforms.
+        f.sync_data().unwrap();
+    };
+}
+
+pub fn load(stack: &mut NockStack, snap_path: PathBuf) -> io::Result<(u64, Noun)> {
+    let (_num, event_number, state) = latest_snapshot(stack, snap_path)?;
+
+    let jammed_arvo =
+        unsafe { IndirectAtom::new_raw(stack, state.size() - 1, state.data_pointer().add(1)) };
+
+    let arvo = cue(stack, jammed_arvo.as_atom());
+
+    Ok((event_number, arvo))
+}
+
+fn latest_snapshot(
+    stack: &mut NockStack,
+    snap_path: PathBuf,
+) -> io::Result<(u8, u64, IndirectAtom)> {
+    let res0 = load_snapshot(stack, snap_path.clone(), 0);
+    let res1 = load_snapshot(stack, snap_path.clone(), 1);
+
+    match (res0, res1) {
+        (Ok((event_number_0, state_0)), Ok((event_number_1, state_1))) => {
+            if event_number_0 > event_number_1 {
+                Ok((0, event_number_0, state_0))
+            } else {
+                Ok((1, event_number_1, state_1))
+            }
+        }
+        (Ok((event_number_0, state_0)), Err(_)) => Ok((0, event_number_0, state_0)),
+        (Err(_), Ok((event_number_1, state_1))) => Ok((1, event_number_1, state_1)),
+        (Err(_), Err(_)) => Err(io::Error::new(
+            io::ErrorKind::NotFound,
+            "no valid snapshot found",
+        )),
+    }
+}
+
+fn load_snapshot(
+    stack: &mut NockStack,
+    mut snap_path: PathBuf,
+    number: u8,
+) -> io::Result<(u64, IndirectAtom)> {
+    snap_path.push(format!("snapshot-{}.snap", number));
+
+    eprintln!("\rload: snapshot at {:?}", snap_path);
+
+    let f = File::open(snap_path)?;
+
+    let in_len = f.metadata().unwrap().len() - 8;
+    let word_len = (in_len + 7) >> 3;
+    let (event_number, state) = unsafe {
+        let in_map = Mmap::map(&f).unwrap();
+        let in_ptr = in_map.as_ptr();
+        let (mut state, dest) = IndirectAtom::new_raw_mut(stack, word_len as usize);
+        let mugged = (*in_ptr.add(0) as u32)
+            | ((*in_ptr.add(1) as u32) << 8)
+            | ((*in_ptr.add(2) as u32) << 16)
+            | ((*in_ptr.add(3) as u32) << 24);
+        write_bytes(dest.add(word_len as usize - 1), 0, 8);
+        copy_nonoverlapping(in_ptr.add(8), dest as *mut u8, in_len as usize);
+        mem::drop(in_map);
+        state.normalize(); // know it's not direct because first word is event number
+
+        if mug_u32(stack, state.as_noun()) != mugged {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                "snapshot checksum mismatch",
+            ));
+        }
+
+        (*state.data_pointer(), state)
+    };
+
+    Ok((event_number, state))
+}