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).

DeliveryMode::Pool, DeliveryMode::Signalled and EntryDelivery::new take windows-threadpool-sys's CallbackEnviron, so that crate's types are part of this crate's public API. A minor release of windows-threadpool-sys that changes those types is a minor release of this crate too, in the 0.x series: 0.3 of this crate pairs with 0.3 of windows-threadpool-sys.

Opening a file

A file target is made from a file this crate opened, and from nothing else (DI-D-60). How a handle was opened decides what its target can claim, and the only call that reports it is one this repository does not make, so the open mode is recorded by the type that opens the file instead of asked of a handle afterwards. OpenedFile<Cached> is a file opened for cached I/O: it claims BestEffort and needs no alignment. OpenedFile<Unbuffered> is opened with no intermediate buffering: it claims SharedDevice, needs a sector's alignment (4096 unless told less with with_sector_size), and offers per-write write-through. A consumer that needs the stronger warranty asks for OpenedFile<Unbuffered> in its own signature, and a cached file cannot be passed. A handle opened elsewhere cannot be given to a target in this version.

use win_durable_ioring::windows_backend::{
    Cached, FileOptions, OpenedFile, Unbuffered, Win32FileTarget,
};
use win_durable_ioring::{IoTarget, WarrantyClass};

let path = std::env::temp_dir().join(format!("durable-ioring-open-{}", std::process::id()));
std::fs::write(&path, [0u8; 8192])?;
let options = FileOptions::new().read(true).write(true);

let cached = Win32FileTarget::new(OpenedFile::<Cached>::open(&path, options)?);
assert_eq!(cached.warranty(), WarrantyClass::BestEffort);

let unbuffered = Win32FileTarget::new(OpenedFile::<Unbuffered>::open(&path, options)?);
assert_eq!(unbuffered.warranty(), WarrantyClass::SharedDevice);
assert!(unbuffered.capabilities().write_through());
# std::fs::remove_file(&path)?;
# Ok::<(), std::io::Error>(())

An unbuffered file is refused on a remote device, where the claim cannot hold.

A raw volume

A raw volume is a target too (DI-D-61): one handle, one commit per volume, and regions are offsets the consumer allocates inside it, not targets of their own. OpenedVolume opens it by a path that can only name a volume, a drive letter or a volume GUID, and finds what the backing must know for itself with documented device controls (win-volume-sys): the disks it sits on, which become its flush domains (disk_flush_domain names a disk, so a consumer can declare the same domain for a file on it), and the sector it is addressed in, which is the alignment of its operations. A volume that is remote, or not a volume, is refused. It claims SharedDevice when opened unbuffered and BestEffort when opened cached, and never Complete: a lock keeps other handles off the volume, not other writers off its disks. Opening one for writing needs an administrator, and VolumeOptions::lock takes the volume lock, which a write to a volume with a file system on it needs and which fails if anything else has the volume open.

use win_durable_ioring::windows_backend::{
    OpenedVolume, Unbuffered, VolumeOptions, VolumePath, Win32FileTarget,
};

let path = VolumePath::drive('E')?;
let volume = OpenedVolume::<Unbuffered>::open(&path, VolumeOptions::new().lock(true))?;
let target = Win32FileTarget::from_volume(volume);
# Ok::<(), Box<dyn std::error::Error>>(())

A whole disk

OpenedDisk opens a physical disk by its number, \\.\PhysicalDriveN, rather than a volume on it (DI-D-62). The disk is its own flush domain, named by disk_flush_domain as a volume on it would name it, and its operations are aligned to its sector, which a device control reports. It claims SharedDevice when opened unbuffered and BestEffort when opened cached, never Complete, for the reason a volume does not. Windows documents that it blocks a direct write to a sector a mounted volume covers, so a disk with volumes on it is written through them, or has them locked or is taken offline first; the tests here use a disk with no partition. Opening one for writing needs an administrator.

use win_durable_ioring::windows_backend::{DiskPath, OpenedDisk, Unbuffered, Win32FileTarget};

let disk = OpenedDisk::<Unbuffered>::open(DiskPath::number(3))?;
let target = Win32FileTarget::from_disk(disk);
# Ok::<(), std::io::Error>(())

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::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 = windows_backend::OpenedFile::<windows_backend::Cached>::open(
#     &path,
#     windows_backend::FileOptions::new().read(true).write(true),
# )?;
# 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(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::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 = windows_backend::OpenedFile::<windows_backend::Cached>::open(
#     &path,
#     windows_backend::FileOptions::new().read(true).write(true),
# )?;
# 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(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::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 = windows_backend::OpenedFile::<windows_backend::Cached>::open(
#     &path,
#     windows_backend::FileOptions::new().read(true).write(true),
# )?;
# 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(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>(())