# win-durable-ioring
A ring for Windows file I/O with **durability scheduled into its work**: the Windows driver for
[durable-ioring](../durable-ioring/COMPONENT.md), built on
[windows-ioring-sys](../windows-ioring-sys/README.md)'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](../durable-ioring/DESIGN-NOTES.md#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](../durable-ioring/COMPONENT.md), 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](../durable-ioring/CONTRACT.md); its spelling as Rust types is
[API.md](../durable-ioring/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/](examples/blocking_push/main.rs), to
use as it stands or adapt ([DI-D-56](../durable-ioring/DESIGN-NOTES.md#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
```rust
# 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.
```rust
# 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.
```rust
# 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>(())
```