win-durable-ioring

A ring for Windows file I/O with durability scheduled into its work: the Windows driver for durable-ioring, built on windows-ioring-sys's IoRing. A consumer submits reads and writes as it would to the ring, tags each write with an epoch, asks for durability with make_durable_through(n), and learns -- in the same completion stream as its operations -- when everything through n has become durable, or that it has failed and which writes are suspect as a result.

What is here, and what is not

Here (DI-D-50): Dioring, the threaded driver that implements the contract over the Windows backend; the backend and its file target (windows_backend), the thread-pool and pumped and signalled delivery of its completions, the readiness Event and the harness check of it (readiness), the interrupt-time clock the driver stamps with, and a way to read an HRESULT out of an io::Error (error_code_of). It forbids unsafe code: a Win32 call no lower crate wraps safely is added to that crate.

Not here: the contract, the core that decides when writes are durable, the target and backend traits, the in-memory backend and the conformance oracle. Those are durable-ioring, which has no Windows dependency, so another platform's driver can be built on it as this one is. The specification is that crate's CONTRACT.md; its spelling as Rust types is API.md. This crate re-exports the portable crate's API, so a consumer needs only this one.

A blocking push is not an API of either crate: a refused push names its cause, and the loop that waits it out is a maintained example, examples/blocking_push/, to use as it stands or adapt (DI-D-56).

Worked examples

Each example is dioring running the trace in its comments, compiled and run as a test, so a change that breaks one breaks the build. In the comments, -> is a call the consumer makes and <- an entry it pops. Every write is in the default lineage, so the lineage is not shown; construction returns a handle to it, and tag(n) is its epoch n as a write or a seal takes it, epoch(n) as a query or an entry names it. A commit failure cannot be provoked on a healthy machine through a Win32 file, so each example arms one with dioring's fault seam, fail_next_flush, which makes the next flush of a file report the error given whatever the device answered. Setting up the instance, and a pop_until that pops until an entry it is looking for, are hidden.

Healing a failure

# use std::os::windows::io::OwnedHandle;
# use std::time::{Duration, Instant};
# use win_durable_ioring::*;
# type Ring = Dioring<Vec<u8>>;
# const FILE: IoTargetKey = IoTargetKey(1);
# fn pop_until(ring: &mut Ring, stop: impl Fn(&EntryOf<Ring>) -> bool) -> EntryOf<Ring> {
#     let started = Instant::now();
#     loop {
#         match ring.pop().expect("pop") {
#             Some(entry) if stop(&entry) => return entry,
#             Some(_) => {}
#             None => {
#                 assert!(started.elapsed() < Duration::from_secs(10), "nothing arrived");
#                 std::thread::sleep(Duration::from_millis(1));
#             }
#         }
#     }
# }
# let path = std::env::temp_dir().join(format!("durable-ioring-heal-{}", std::process::id()));
# std::fs::write(&path, [0u8; 64])?;
# let file = std::fs::OpenOptions::new().read(true).write(true).open(&path)?;
# let (mut ring, lineage) = Ring::new(Setup {
#     submission_queue_size: 64,
#     completion_queue_size: 64,
#     files: vec![TargetSetup {
#         key: FILE,
#         target: windows_backend::Win32FileTarget::new(SharedHandle::new(OwnedHandle::from(file))),
#     }],
#     buffers: Vec::new(),
#     provider: None,
#     diagnostics: None,
#     delivery: DeliveryMode::Pool(None),
# })
# .expect("an instance");
let tag = |n: u64| lineage.at(n); // what a write or a seal takes
let epoch = |n: u64| Epoch::new(lineage.lineage(), n); // what a query or an entry names

// -> write(tag 41, A) ; write(tag 41, B) ; write(tag 41, C)
for offset in [0, 8, 16] {
    ring.write(FILE, offset, vec![1; 8], tag(41), ()).expect("a write");
}
ring.fail_next_flush(FILE, 1117); // ERROR_IO_DEVICE, for the flush the seal issues
// -> make_durable_through(41)                 // seals everything <= 41
ring.make_durable_through(tag(41)).expect("a seal");
// <- WriteDone(A) ; WriteDone(B) ; WriteDone(C)
// <- Failed { failure: F1, cause: Commit, suspect: [A@41, B@41, C@41] }
let Entry::Failed(f1) = pop_until(&mut ring, |e| matches!(e, Entry::Failed(_))) else {
    unreachable!()
};
assert!(matches!(f1.cause, FailureCause::Commit { .. }));
assert_eq!(f1.suspect.writes().len(), 3);
assert_eq!(ring.durable_through(lineage.lineage()), Ok(None)); // nothing is durable yet
// -> write(tag 41, A')                        // refused: 41 is sealed (guarantee 6)
let refused = ring.write(FILE, 0, vec![2; 8], tag(41), ()).expect_err("41 is sealed");
assert!(matches!(refused.reason, PushRefusal::Sealed { .. }));
// -> write(tag 42, A') ; write(tag 42, B') ; write(tag 42, C')
for offset in [0, 8, 16] {
    ring.write(FILE, offset, vec![2; 8], tag(42), ()).expect("a write");
}
// -> heal(F1)                                 // "I have re-issued what I need"
ring.resolve(vec![(f1.token, Resolution::Heal)]).expect("heal");
// -> make_durable_through(42)                 // the first after the heal reaching the re-issue
ring.make_durable_through(tag(42)).expect("a seal");
// <- WriteDone(A') ; WriteDone(B') ; WriteDone(C')
// <- Durable { through: 42 }                  // F1 healed; 41 and 42 durable
pop_until(&mut ring, |e| matches!(e, Entry::Durable { through } if *through == epoch(42)));
assert_eq!(ring.epoch_state(epoch(41)), Ok(EpochState::Durable));
assert!(ring.failures().is_empty());
# drop(ring);
# std::fs::remove_file(&path)?;
# Ok::<(), std::io::Error>(())

Without the heal, the last entry is Blocked { through: 42, by: F1 }: 42's own commit succeeded, but the high-water mark cannot pass the unresolved failure at 41.

Failures and epochs are many-to-many

P is pushed before the seal, rather than after it as a consumer might, so that it is certainly pushed before the failure is observed: the seal's commit is issued as soon as A completes.

# use std::os::windows::io::OwnedHandle;
# use std::time::{Duration, Instant};
# use win_durable_ioring::*;
# type Ring = Dioring<Vec<u8>>;
# const FILE: IoTargetKey = IoTargetKey(1);
# fn pop_until(ring: &mut Ring, stop: impl Fn(&EntryOf<Ring>) -> bool) -> EntryOf<Ring> {
#     let started = Instant::now();
#     loop {
#         match ring.pop().expect("pop") {
#             Some(entry) if stop(&entry) => return entry,
#             Some(_) => {}
#             None => {
#                 assert!(started.elapsed() < Duration::from_secs(10), "nothing arrived");
#                 std::thread::sleep(Duration::from_millis(1));
#             }
#         }
#     }
# }
# let path = std::env::temp_dir().join(format!("durable-ioring-many-{}", std::process::id()));
# std::fs::write(&path, [0u8; 64])?;
# let file = std::fs::OpenOptions::new().read(true).write(true).open(&path)?;
# let (mut ring, lineage) = Ring::new(Setup {
#     submission_queue_size: 64,
#     completion_queue_size: 64,
#     files: vec![TargetSetup {
#         key: FILE,
#         target: windows_backend::Win32FileTarget::new(SharedHandle::new(OwnedHandle::from(file))),
#     }],
#     buffers: Vec::new(),
#     provider: None,
#     diagnostics: None,
#     delivery: DeliveryMode::Pool(None),
# })
# .expect("an instance");
let tag = |n: u64| lineage.at(n); // what a write or a seal takes
let epoch = |n: u64| Epoch::new(lineage.lineage(), n); // what a query or an entry names
let ops = |set: &SuspectSet<WindowsIds<u64>>| set.writes().iter().map(|w| w.op).collect::<Vec<_>>();

// -> write(tag 41, A) ; write(tag 43, P)      // 43 is open
let a = ring.write(FILE, 0, vec![1; 8], tag(41), ()).expect("a write");
let p = ring.write(FILE, 8, vec![1; 8], tag(43), ()).expect("a write");
ring.fail_next_flush(FILE, 1117);
// -> make_durable_through(41)                 // seals 41
ring.make_durable_through(tag(41)).expect("a seal");
// <- WriteDone(A) ; WriteDone(P)
// <- Failed { failure: F1, cause: Commit, suspect: [A@41, P@43] }
let Entry::Failed(f1) = pop_until(&mut ring, |e| matches!(e, Entry::Failed(_))) else {
    unreachable!()
};
assert_eq!(ops(&f1.suspect), [a, p]); // one failure, two epochs: sealed 41 and open 43
// -> write(tag 43, R)                         // still allowed: 43 is open
let r = ring.write(FILE, 16, vec![1; 8], tag(43), ()).expect("a write");
// <- WriteDone(R)
pop_until(&mut ring, |e| matches!(e, Entry::Op(done) if done.id == r));
// -> import_failure()                         // the consumer learned of an outside failure
let f2_id = ring.import_failure(ImportScope::All).expect("the whole instance").failure;
// <- Failed { failure: F2, cause: Imported, suspect: [A@41, P@43, R@43] }
let Entry::Failed(f2) = pop_until(&mut ring, |e| matches!(e, Entry::Failed(_))) else {
    unreachable!()
};
assert_eq!(f2.id, f2_id);
assert_eq!(ops(&f2.suspect), [a, p, r]); // epoch 43 now sits in F1 and F2

// The first ending: heal both. Re-issue A, P and R under tag 44, resolve both in one atomic
// call, and seal 44.
for offset in [0, 8, 16] {
    ring.write(FILE, offset, vec![2; 8], tag(44), ()).expect("a write");
}
ring.resolve(vec![(f1.token, Resolution::Heal), (f2.token, Resolution::Heal)])
    .expect("heal both");
ring.make_durable_through(tag(44)).expect("a seal");
// <- Durable { through: 44 }                  // 41 and 43 passed once both failures were resolved
pop_until(&mut ring, |e| matches!(e, Entry::Durable { through } if *through == epoch(44)));
assert_eq!(ring.epoch_state(epoch(41)), Ok(EpochState::Durable));
assert_eq!(ring.epoch_state(epoch(43)), Ok(EpochState::Durable));
# drop(ring);
# std::fs::remove_file(&path)?;
# Ok::<(), std::io::Error>(())

The second ending: heal F1, abandon F2. Epochs 41 and 43 both sit in F2, so both become abandoned -- one abandonment is enough -- even though F1 was healed. The high-water mark moves past them once every failure containing them is resolved: F2 at once, and F1 when its heal takes effect, at the first seal made after it that reaches the re-issue. Any operation gated on 41 or 43 fails as never issued.

# use std::os::windows::io::OwnedHandle;
# use std::time::{Duration, Instant};
# use win_durable_ioring::*;
# type Ring = Dioring<Vec<u8>>;
# const FILE: IoTargetKey = IoTargetKey(1);
# fn pop_until(ring: &mut Ring, stop: impl Fn(&EntryOf<Ring>) -> bool) -> EntryOf<Ring> {
#     let started = Instant::now();
#     loop {
#         match ring.pop().expect("pop") {
#             Some(entry) if stop(&entry) => return entry,
#             Some(_) => {}
#             None => {
#                 assert!(started.elapsed() < Duration::from_secs(10), "nothing arrived");
#                 std::thread::sleep(Duration::from_millis(1));
#             }
#         }
#     }
# }
# let path = std::env::temp_dir().join(format!("durable-ioring-abandon-{}", std::process::id()));
# std::fs::write(&path, [0u8; 64])?;
# let file = std::fs::OpenOptions::new().read(true).write(true).open(&path)?;
# let (mut ring, lineage) = Ring::new(Setup {
#     submission_queue_size: 64,
#     completion_queue_size: 64,
#     files: vec![TargetSetup {
#         key: FILE,
#         target: windows_backend::Win32FileTarget::new(SharedHandle::new(OwnedHandle::from(file))),
#     }],
#     buffers: Vec::new(),
#     provider: None,
#     diagnostics: None,
#     delivery: DeliveryMode::Pool(None),
# })
# .expect("an instance");
# let tag = |n: u64| lineage.at(n);
# let epoch = |n: u64| Epoch::new(lineage.lineage(), n);
# ring.write(FILE, 0, vec![1; 8], tag(41), ()).expect("a write");
# ring.write(FILE, 8, vec![1; 8], tag(43), ()).expect("a write");
# ring.fail_next_flush(FILE, 1117);
# ring.make_durable_through(tag(41)).expect("a seal");
# let Entry::Failed(f1) = pop_until(&mut ring, |e| matches!(e, Entry::Failed(_))) else {
#     unreachable!()
# };
# let r = ring.write(FILE, 16, vec![1; 8], tag(43), ()).expect("a write");
# pop_until(&mut ring, |e| matches!(e, Entry::Op(done) if done.id == r));
# ring.import_failure(ImportScope::All).expect("the whole instance");
# let Entry::Failed(f2) = pop_until(&mut ring, |e| matches!(e, Entry::Failed(_))) else {
#     unreachable!()
# };
// The trace above, to F2's arrival, is hidden. Then:
ring.resolve(vec![(f1.token, Resolution::Heal), (f2.token, Resolution::Abandon)])
    .expect("heal F1, abandon F2");
// <- Abandoned { failure: F2, suspect: [A@41, P@43, R@43] }
pop_until(&mut ring, |e| matches!(e, Entry::Abandoned { .. }));
assert_eq!(ring.epoch_state(epoch(41)), Ok(EpochState::Abandoned));
assert_eq!(ring.epoch_state(epoch(43)), Ok(EpochState::Abandoned));
// -> make_durable_through(44)                 // F1's heal takes effect here
ring.make_durable_through(tag(44)).expect("a seal");
// <- Durable { through: 44 }                  // "durable through 44" means durable or abandoned
pop_until(&mut ring, |e| matches!(e, Entry::Durable { through } if *through == epoch(44)));
assert_eq!(ring.durable_through(lineage.lineage()), Ok(Some(44)));
assert_eq!(ring.epoch_state(epoch(43)), Ok(EpochState::Abandoned));
# drop(ring);
# std::fs::remove_file(&path)?;
# Ok::<(), std::io::Error>(())