Skip to content

fix(block): classify block I/O errors into retryable and fatal #1896

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
emersonbusson wants to merge 1 commit into from
Closed

fix(block): classify block I/O errors into retryable and fatal #1896

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 15 additions & 15 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

14 changes: 7 additions & 7 deletions crates/ramshared-block/src/isolated_origin.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -235,7 +235,7 @@ pub struct AuthoritativeOriginBackend<O, C> {
impl<O: OriginStorage, C: BestEffortCache> AuthoritativeOriginBackend<O, C> {
pub fn new(origin: O, cache: C, size: u64, block: u32) -> Result<Self, IoError> {
if size == 0 || block == 0 || !size.is_multiple_of(block as u64) {
return Err(IoError("invalid authoritative origin geometry".into()));
return Err(IoError::Fatal("invalid authoritative origin geometry".into()));
}
Ok(Self {
origin,
Expand Down Expand Up @@ -275,7 +275,7 @@ impl<O: OriginStorage, C: BestEffortCache> AuthoritativeOriginBackend<O, C> {
self.telemetry.releases = self.telemetry.releases.saturating_add(1);
Ok(0)
}
CacheMutation::Failed => Err(IoError(
CacheMutation::Failed => Err(IoError::Fatal(
"cache release acknowledgement was unavailable".into(),
)),
}
Expand Down Expand Up @@ -313,14 +313,14 @@ impl<O: OriginStorage, C: BestEffortCache> AuthoritativeOriginBackend<O, C> {
.checked_add(len as u64)
.filter(|end| *end <= self.size)
.map(|_| ())
.ok_or_else(|| IoError("authoritative origin I/O is out of range".into()))
.ok_or_else(|| IoError::Fatal("authoritative origin I/O is out of range".into()))
}

fn require_ready_origin(&self) -> Result<(), IoError> {
if self.origin_state == OriginState::Ready {
Ok(())
} else {
Err(IoError(
Err(IoError::Fatal(
"origin authority is unavailable pending three read+sync probes".into(),
))
}
Expand Down Expand Up @@ -618,7 +618,7 @@ mod tests {
impl OriginStorage for FaultOrigin {
fn read_at(&mut self, offset: u64, destination: &mut [u8]) -> Result<usize, IoError> {
if self.0.fail_read.get() {
return Err(IoError("fixture origin read failure".into()));
return Err(IoError::Fatal("fixture origin read failure".into()));
}
let start = offset as usize;
destination.copy_from_slice(&self.0.bytes.borrow()[start..start + destination.len()]);
Expand All @@ -627,7 +627,7 @@ mod tests {

fn write_at(&mut self, offset: u64, data: &[u8]) -> Result<usize, IoError> {
if self.0.fail_write.get() {
return Err(IoError("fixture origin write failure".into()));
return Err(IoError::Fatal("fixture origin write failure".into()));
}
let start = offset as usize;
self.0.bytes.borrow_mut()[start..start + data.len()].copy_from_slice(data);
Expand All @@ -636,7 +636,7 @@ mod tests {

fn sync_data(&mut self) -> Result<(), IoError> {
if self.0.fail_sync.get() {
return Err(IoError("fixture origin sync failure".into()));
return Err(IoError::Fatal("fixture origin sync failure".into()));
}
self.0.syncs.set(self.0.syncs.get().saturating_add(1));
Ok(())
Expand Down
3 changes: 3 additions & 0 deletions crates/ramshared-block/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,3 +56,6 @@ mod tests {
assert_eq!(DEFAULT_CHUNK_MIB, 128);
}
}

#[cfg(test)]
mod red_test;
42 changes: 21 additions & 21 deletions crates/ramshared-block/src/origin_cache.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,14 +29,14 @@ pub trait OriginStorage {
while !buf.is_empty() {
let read = self.read_at(off, buf)?;
if read == 0 {
return Err(IoError("origin read made no progress".into()));
return Err(IoError::Fatal("origin read made no progress".into()));
}
if read > buf.len() {
return Err(IoError("origin read exceeded requested length".into()));
return Err(IoError::Fatal("origin read exceeded requested length".into()));
}
off = off
.checked_add(read as u64)
.ok_or_else(|| IoError("origin read offset overflow".into()))?;
.ok_or_else(|| IoError::Fatal("origin read offset overflow".into()))?;
buf = &mut buf[read..];
}
Ok(())
Expand All @@ -46,14 +46,14 @@ pub trait OriginStorage {
while !data.is_empty() {
let written = self.write_at(off, data)?;
if written == 0 {
return Err(IoError("origin write made no progress".into()));
return Err(IoError::Fatal("origin write made no progress".into()));
}
if written > data.len() {
return Err(IoError("origin write exceeded requested length".into()));
return Err(IoError::Fatal("origin write exceeded requested length".into()));
}
off = off
.checked_add(written as u64)
.ok_or_else(|| IoError("origin write offset overflow".into()))?;
.ok_or_else(|| IoError::Fatal("origin write offset overflow".into()))?;
data = &data[written..];
}
Ok(())
Expand All @@ -79,18 +79,18 @@ impl OriginStorage for FileOrigin {
{
self.file
.read_at(buf, off)
.map_err(|error| IoError(error.to_string()))
.map_err(|error| IoError::Fatal(error.to_string()))
}
#[cfg(windows)]
{
self.file
.seek_read(buf, off)
.map_err(|error| IoError(error.to_string()))
.map_err(|error| IoError::Fatal(error.to_string()))
}
#[cfg(not(any(unix, windows)))]
{
let _ = (off, buf);
Err(IoError("unsupported platform for FileOrigin".into()))
Err(IoError::Fatal("unsupported platform for FileOrigin".into()))
}
}

Expand All @@ -99,25 +99,25 @@ impl OriginStorage for FileOrigin {
{
self.file
.write_at(data, off)
.map_err(|error| IoError(error.to_string()))
.map_err(|error| IoError::Fatal(error.to_string()))
}
#[cfg(windows)]
{
self.file
.seek_write(data, off)
.map_err(|error| IoError(error.to_string()))
.map_err(|error| IoError::Fatal(error.to_string()))
}
#[cfg(not(any(unix, windows)))]
{
let _ = (off, data);
Err(IoError("unsupported platform for FileOrigin".into()))
Err(IoError::Fatal("unsupported platform for FileOrigin".into()))
}
}

fn sync_data(&mut self) -> Result<(), IoError> {
self.file
.sync_data()
.map_err(|error| IoError(error.to_string()))
.map_err(|error| IoError::Fatal(error.to_string()))
}
}

Expand Down Expand Up @@ -251,7 +251,7 @@ impl<'p, P: VramProvider + 'p, O: OriginStorage> WriteThroughCacheBackend<'p, P,
|| !size.is_multiple_of(block as u64)
|| !chunk_bytes.is_multiple_of(block as u64)
{
return Err(IoError("invalid origin cache geometry".into()));
return Err(IoError::Fatal("invalid origin cache geometry".into()));
}
let chunk_count = size.div_ceil(chunk_bytes);
let mut chunks = Vec::with_capacity(chunk_count as usize);
Expand Down Expand Up @@ -417,7 +417,7 @@ impl<'p, P: VramProvider + 'p, O: OriginStorage> WriteThroughCacheBackend<'p, P,
off.checked_add(len as u64)
.filter(|end| *end <= self.size)
.map(|_| ())
.ok_or_else(|| IoError("origin cache I/O is out of range".into()))
.ok_or_else(|| IoError::Fatal("origin cache I/O is out of range".into()))
}

fn valid_block_count(&self) -> u64 {
Expand Down Expand Up @@ -522,7 +522,7 @@ impl<'p, P: VramProvider + 'p, O: OriginStorage> WriteThroughCacheBackend<'p, P,
if self.origin_state == OriginState::Ready {
Ok(())
} else {
Err(IoError(
Err(IoError::Fatal(
"origin authority is unavailable pending three read+sync probes".into(),
))
}
Expand Down Expand Up @@ -771,7 +771,7 @@ mod tests {
fn read_at(&mut self, off: u64, buf: &mut [u8]) -> Result<usize, IoError> {
self.events.borrow_mut().push("origin_read");
if self.fail_read.get() {
return Err(IoError("injected origin read failure".into()));
return Err(IoError::Fatal("injected origin read failure".into()));
}
let bytes = self.bytes.borrow();
let start = off as usize;
Expand All @@ -783,11 +783,11 @@ mod tests {
fn write_at(&mut self, off: u64, data: &[u8]) -> Result<usize, IoError> {
self.events.borrow_mut().push("origin_write");
if self.fail_write.get() {
return Err(IoError("injected origin write failure".into()));
return Err(IoError::Fatal("injected origin write failure".into()));
}
let writes_before_failure = self.writes_before_failure.get();
if writes_before_failure == 0 {
return Err(IoError("injected partial origin write failure".into()));
return Err(IoError::Fatal("injected partial origin write failure".into()));
}
if self.zero_write.get() {
return Ok(0);
Expand All @@ -803,7 +803,7 @@ mod tests {
fn sync_data(&mut self) -> Result<(), IoError> {
self.events.borrow_mut().push("origin_sync");
if self.fail_sync.get() {
Err(IoError("injected origin sync failure".into()))
Err(IoError::Fatal("injected origin sync failure".into()))
} else {
Ok(())
}
Expand Down Expand Up @@ -1111,7 +1111,7 @@ mod tests {

let error = backend.write_at(0, b"stop").unwrap_err();

assert!(error.0.contains("no progress"));
assert!(error.to_string().contains("no progress"));
assert_eq!(backend.telemetry().origin_written_bytes, 0);
assert_eq!(backend.telemetry().valid_blocks, 0);
}
Expand Down
1 change: 1 addition & 0 deletions crates/ramshared-block/src/protocol.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ pub const NBD_EIO: u32 = 5;
pub const NBD_EACCES: u32 = 13;
pub const NBD_EINVAL: u32 = 22;
pub const NBD_ERANGE: u32 = 34;
pub const NBD_ENOSPC: u32 = 28;

/// NBD commands (`type` field).
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
Expand Down
39 changes: 39 additions & 0 deletions crates/ramshared-block/src/red_test.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,39 @@
use crate::protocol::{Command, NBD_ENOSPC, Request};
use crate::request::{serve, BlockBackend, IoError};

struct ChaosBackend;

impl BlockBackend for ChaosBackend {
fn size_bytes(&self) -> u64 {
4096
}
fn block_size(&self) -> u32 {
4096
}
fn read_at(&mut self, _off: u64, _buf: &mut [u8]) -> Result<(), IoError> {
Err(IoError::Retryable("chaos read network glitch".to_string()))
}
fn write_at(&mut self, _off: u64, _data: &[u8]) -> Result<(), IoError> {
Err(IoError::Retryable("chaos write network glitch".to_string()))
}
fn flush(&mut self) -> Result<(), IoError> {
Err(IoError::Retryable("chaos flush network glitch".to_string()))
}
}

#[test]
fn retryable_error_returns_enospc() {
let mut b = ChaosBackend;
let req = Request {
flags: 0,
cmd: Command::Write,
handle: 0x42,
offset: 0,
len: 4096,
};
let w = serve(&req, &vec![0; 4096], &mut b);
assert_eq!(
u32::from_be_bytes([w.reply[4], w.reply[5], w.reply[6], w.reply[7]]),
NBD_ENOSPC
);
}
Loading
Loading