[pbs-devel] [PATCH v2 0/5] Fix PBS blockdriver for non-VM settings

[pbs-devel] [PATCH v2 0/5] Fix PBS blockdriver for non-VM settings

[Stefan Reiter]

When using the PBS blockdriver with qemu-nbd (for example), it can happen that
enough read requests are issued to saturate the tokio thread pool. Not an issue
in general, but as me and Wolfgang painstakenly discovered a while back, it does
break block_on, which is used in BufferedFixedReader. This means that reading
larger amounts of data would hang the QEMU process [0].

Fix this by replacing the BufferedFixedReader with an AsyncIndexReader,
implementing AsyncSeek for it in the process. This makes the entire API async,
requiring no block_on anymore.

Incidentally, this also gave me my best benchmark results yet, coming in at
above 1.6 Gb/s read speed via NBD on my local machine.

Additionally I discovered a seperate bug (fixed by patch 5), wherein read
requests that we're not aligned to the chunk size would return bogus data. This
too only seems to happen in non-VM connections (e.g. nbd, etc...).

v2:
* Remove BufferedFixedReader entirely, use AsyncIndexReader instead
* Implement AsyncSeek for AsyncIndexReader
* Fix the second bug in Rust instead of QEMU C


[0] ...and since the NBD kernel driver appears to be horribly broken, this often
also crashes most of the system, but that's a different story. If you ever get
in this situation, 'nbd-client -d /dev/nbdX' works (sometimes) to force
disconnect the device ('qemu-nbd -d' intelligently issues a read before
disconnecting, thus hanging before getting anything done...)


backup: Stefan Reiter (3):
  add and implement chunk_from_offset for IndexFile
  implement AsyncSeek for AsyncIndexReader
  remove BufferedFixedReader interface

 src/backup/async_index_reader.rs | 116 ++++++++++++++++++----
 src/backup/dynamic_index.rs      |  18 ++++
 src/backup/fixed_index.rs        | 165 +++----------------------------
 src/backup/index.rs              |   4 +
 4 files changed, 129 insertions(+), 174 deletions(-)

backup-qemu: Stefan Reiter (2):
  use AsyncIndexReader for read_image_at
  read_image_at: iterate until buffer is filled

 current-api.h  |  4 ++--
 src/lib.rs     |  4 ++--
 src/restore.rs | 35 +++++++++++++++++++++++------------
 3 files changed, 27 insertions(+), 16 deletions(-)

-- 
2.20.1

[pbs-devel] [PATCH v2 backup 1/5] add and implement chunk_from_offset for IndexFile

[Stefan Reiter]

Necessary for byte-wise seeking through chunks in an index.

Signed-off-by: Stefan Reiter <s.reiter@proxmox.com>
---
 src/backup/dynamic_index.rs | 18 ++++++++++++++++++
 src/backup/fixed_index.rs   | 11 +++++++++++
 src/backup/index.rs         |  3 +++
 3 files changed, 32 insertions(+)

diff --git a/src/backup/dynamic_index.rs b/src/backup/dynamic_index.rs
index 4907fe1f..887b7cf3 100644
--- a/src/backup/dynamic_index.rs
+++ b/src/backup/dynamic_index.rs
@@ -216,6 +216,24 @@ impl IndexFile for DynamicIndexReader {
             digest: self.index[pos].digest.clone(),
         })
     }
+
+    fn chunk_from_offset(&self, offset: u64) -> Option<(usize, u64)> {
+        let end_idx = self.index.len() - 1;
+        let end = self.chunk_end(end_idx);
+        let found_idx = self.binary_search(0, 0, end_idx, end, offset);
+        let found_idx = match found_idx {
+            Ok(i) => i,
+            Err(_) => return None
+        };
+
+        let found_start = if found_idx == 0 {
+            0
+        } else {
+            self.chunk_end(found_idx - 1)
+        };
+
+        Some((found_idx, offset - found_start))
+    }
 }
 
 struct CachedChunk {
diff --git a/src/backup/fixed_index.rs b/src/backup/fixed_index.rs
index 73d0dad0..b7e785d6 100644
--- a/src/backup/fixed_index.rs
+++ b/src/backup/fixed_index.rs
@@ -219,6 +219,17 @@ impl IndexFile for FixedIndexReader {
 
         (csum, chunk_end)
     }
+
+    fn chunk_from_offset(&self, offset: u64) -> Option<(usize, u64)> {
+        if offset >= self.size {
+            return None;
+        }
+
+        Some((
+            (offset / self.chunk_size as u64) as usize,
+            offset % self.chunk_size as u64
+        ))
+    }
 }
 
 pub struct FixedIndexWriter {
diff --git a/src/backup/index.rs b/src/backup/index.rs
index efdf3b54..2eab8524 100644
--- a/src/backup/index.rs
+++ b/src/backup/index.rs
@@ -22,6 +22,9 @@ pub trait IndexFile {
     fn index_bytes(&self) -> u64;
     fn chunk_info(&self, pos: usize) -> Option<ChunkReadInfo>;
 
+    /// Get the chunk index and the relative offset within it for a byte offset
+    fn chunk_from_offset(&self, offset: u64) -> Option<(usize, u64)>;
+
     /// Compute index checksum and size
     fn compute_csum(&self) -> ([u8; 32], u64);
 
-- 
2.20.1

[pbs-devel] [PATCH v2 backup 2/5] implement AsyncSeek for AsyncIndexReader

[Stefan Reiter]

Requires updating the AsyncRead implementation to cope with byte-wise
seeks to intra-chunk positions.

Uses chunk_from_offset to get locations within chunks, but tries to
avoid it for sequential read to not reduce performance from before.

AsyncSeek needs to use the temporary seek_to_pos to avoid changing the
position in case an invalid seek is given and it needs to error in
poll_complete.

Signed-off-by: Stefan Reiter <s.reiter@proxmox.com>
---
 src/backup/async_index_reader.rs | 116 +++++++++++++++++++++++++------
 src/backup/index.rs              |   1 +
 2 files changed, 97 insertions(+), 20 deletions(-)

diff --git a/src/backup/async_index_reader.rs b/src/backup/async_index_reader.rs
index 0911375e..98372aa1 100644
--- a/src/backup/async_index_reader.rs
+++ b/src/backup/async_index_reader.rs
@@ -1,30 +1,35 @@
 use std::future::Future;
 use std::task::{Poll, Context};
 use std::pin::Pin;
+use std::io::SeekFrom;
 
 use anyhow::Error;
 use futures::future::FutureExt;
 use futures::ready;
-use tokio::io::AsyncRead;
+use tokio::io::{AsyncRead, AsyncSeek};
 
 use proxmox::sys::error::io_err_other;
 use proxmox::io_format_err;
 
 use super::IndexFile;
 use super::read_chunk::AsyncReadChunk;
+use super::index::ChunkReadInfo;
 
 enum AsyncIndexReaderState<S> {
     NoData,
     WaitForData(Pin<Box<dyn Future<Output = Result<(S, Vec<u8>), Error>> + Send + 'static>>),
-    HaveData(usize),
+    HaveData,
 }
 
 pub struct AsyncIndexReader<S, I: IndexFile> {
     store: Option<S>,
     index: I,
     read_buffer: Vec<u8>,
+    current_chunk_offset: u64,
     current_chunk_idx: usize,
-    current_chunk_digest: [u8; 32],
+    current_chunk_info: Option<ChunkReadInfo>,
+    position: u64,
+    seek_to_pos: i64,
     state: AsyncIndexReaderState<S>,
 }
 
@@ -37,8 +42,11 @@ impl<S: AsyncReadChunk, I: IndexFile> AsyncIndexReader<S, I> {
             store: Some(store),
             index,
             read_buffer: Vec::with_capacity(1024 * 1024),
+            current_chunk_offset: 0,
             current_chunk_idx: 0,
-            current_chunk_digest: [0u8; 32],
+            current_chunk_info: None,
+            position: 0,
+            seek_to_pos: 0,
             state: AsyncIndexReaderState::NoData,
         }
     }
@@ -58,23 +66,41 @@ where
         loop {
             match &mut this.state {
                 AsyncIndexReaderState::NoData => {
-                    if this.current_chunk_idx >= this.index.index_count() {
+                    let (idx, offset) = if this.current_chunk_info.is_some() &&
+                        this.position == this.current_chunk_info.as_ref().unwrap().range.end
+                    {
+                        // optimization for sequential chunk read
+                        let next_idx = this.current_chunk_idx + 1;
+                        (next_idx, 0)
+                    } else {
+                        match this.index.chunk_from_offset(this.position) {
+                            Some(res) => res,
+                            None => return Poll::Ready(Ok(0))
+                        }
+                    };
+
+                    if idx >= this.index.index_count() {
                         return Poll::Ready(Ok(0));
                     }
 
-                    let digest = this
+                    let info = this
                         .index
-                        .index_digest(this.current_chunk_idx)
-                        .ok_or(io_format_err!("could not get digest"))?
-                        .clone();
+                        .chunk_info(idx)
+                        .ok_or(io_format_err!("could not get digest"))?;
 
-                    if digest == this.current_chunk_digest {
-                        this.state = AsyncIndexReaderState::HaveData(0);
-                        continue;
+                    this.current_chunk_offset = offset;
+                    this.current_chunk_idx = idx;
+                    let old_info = this.current_chunk_info.replace(info.clone());
+
+                    if let Some(old_info) = old_info {
+                        if old_info.digest == info.digest {
+                            // hit, chunk is currently in cache
+                            this.state = AsyncIndexReaderState::HaveData;
+                            continue;
+                        }
                     }
 
-                    this.current_chunk_digest = digest;
-
+                    // miss, need to download new chunk
                     let store = match this.store.take() {
                         Some(store) => store,
                         None => {
@@ -83,7 +109,7 @@ where
                     };
 
                     let future = async move {
-                        store.read_chunk(&digest)
+                        store.read_chunk(&info.digest)
                             .await
                             .map(move |x| (store, x))
                     };
@@ -95,7 +121,7 @@ where
                         Ok((store, mut chunk_data)) => {
                             this.read_buffer.clear();
                             this.read_buffer.append(&mut chunk_data);
-                            this.state = AsyncIndexReaderState::HaveData(0);
+                            this.state = AsyncIndexReaderState::HaveData;
                             this.store = Some(store);
                         }
                         Err(err) => {
@@ -103,8 +129,8 @@ where
                         }
                     };
                 }
-                AsyncIndexReaderState::HaveData(offset) => {
-                    let offset = *offset;
+                AsyncIndexReaderState::HaveData => {
+                    let offset = this.current_chunk_offset as usize;
                     let len = this.read_buffer.len();
                     let n = if len - offset < buf.len() {
                         len - offset
@@ -113,11 +139,13 @@ where
                     };
 
                     buf[0..n].copy_from_slice(&this.read_buffer[offset..(offset + n)]);
+                    this.position += n as u64;
+
                     if offset + n == len {
                         this.state = AsyncIndexReaderState::NoData;
-                        this.current_chunk_idx += 1;
                     } else {
-                        this.state = AsyncIndexReaderState::HaveData(offset + n);
+                        this.current_chunk_offset += n as u64;
+                        this.state = AsyncIndexReaderState::HaveData;
                     }
 
                     return Poll::Ready(Ok(n));
@@ -126,3 +154,51 @@ where
         }
     }
 }
+
+impl<S, I> AsyncSeek for AsyncIndexReader<S, I>
+where
+    S: AsyncReadChunk + Unpin + Sync + 'static,
+    I: IndexFile + Unpin,
+{
+    fn start_seek(
+        self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+        pos: SeekFrom,
+    ) -> Poll<tokio::io::Result<()>> {
+        let this = Pin::get_mut(self);
+        this.seek_to_pos = match pos {
+            SeekFrom::Start(offset) => {
+                offset as i64
+            },
+            SeekFrom::End(offset) => {
+                this.index.index_bytes() as i64 + offset
+            },
+            SeekFrom::Current(offset) => {
+                this.position as i64 + offset
+            }
+        };
+        Poll::Ready(Ok(()))
+    }
+
+    fn poll_complete(
+        self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+    ) -> Poll<tokio::io::Result<u64>> {
+        let this = Pin::get_mut(self);
+
+        let index_bytes = this.index.index_bytes();
+        if this.seek_to_pos < 0 {
+            return Poll::Ready(Err(io_format_err!("cannot seek to negative values")));
+        } else if this.seek_to_pos > index_bytes as i64 {
+            this.position = index_bytes;
+        } else {
+            this.position = this.seek_to_pos as u64;
+        }
+
+        // even if seeking within one chunk, we need to go to NoData to
+        // recalculate the current_chunk_offset (data is cached anyway)
+        this.state = AsyncIndexReaderState::NoData;
+
+        Poll::Ready(Ok(this.position))
+    }
+}
diff --git a/src/backup/index.rs b/src/backup/index.rs
index 2eab8524..c6bab56e 100644
--- a/src/backup/index.rs
+++ b/src/backup/index.rs
@@ -1,6 +1,7 @@
 use std::collections::HashMap;
 use std::ops::Range;
 
+#[derive(Clone)]
 pub struct ChunkReadInfo {
     pub range: Range<u64>,
     pub digest: [u8; 32],
-- 
2.20.1

[pbs-devel] [PATCH v2 backup 3/5] remove BufferedFixedReader interface

[Stefan Reiter]

replaced by AsyncIndexReader

Signed-off-by: Stefan Reiter <s.reiter@proxmox.com>
---
 src/backup/fixed_index.rs | 154 --------------------------------------
 1 file changed, 154 deletions(-)

diff --git a/src/backup/fixed_index.rs b/src/backup/fixed_index.rs
index b7e785d6..59f63d2e 100644
--- a/src/backup/fixed_index.rs
+++ b/src/backup/fixed_index.rs
@@ -13,7 +13,6 @@ use std::os::unix::io::AsRawFd;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 
-use super::read_chunk::*;
 use super::ChunkInfo;
 
 use proxmox::tools::io::ReadExt;
@@ -146,20 +145,6 @@ impl FixedIndexReader {
         Ok(())
     }
 
-    #[inline]
-    fn chunk_end(&self, pos: usize) -> u64 {
-        if pos >= self.index_length {
-            panic!("chunk index out of range");
-        }
-
-        let end = ((pos + 1) * self.chunk_size) as u64;
-        if end > self.size {
-            self.size
-        } else {
-            end
-        }
-    }
-
     pub fn print_info(&self) {
         println!("Size: {}", self.size);
         println!("ChunkSize: {}", self.chunk_size);
@@ -476,142 +461,3 @@ impl FixedIndexWriter {
         Ok(())
     }
 }
-
-pub struct BufferedFixedReader<S> {
-    store: S,
-    index: FixedIndexReader,
-    archive_size: u64,
-    read_buffer: Vec<u8>,
-    buffered_chunk_idx: usize,
-    buffered_chunk_start: u64,
-    read_offset: u64,
-}
-
-impl<S: ReadChunk> BufferedFixedReader<S> {
-    pub fn new(index: FixedIndexReader, store: S) -> Self {
-        let archive_size = index.size;
-        Self {
-            store,
-            index,
-            archive_size,
-            read_buffer: Vec::with_capacity(1024 * 1024),
-            buffered_chunk_idx: 0,
-            buffered_chunk_start: 0,
-            read_offset: 0,
-        }
-    }
-
-    pub fn archive_size(&self) -> u64 {
-        self.archive_size
-    }
-
-    fn buffer_chunk(&mut self, idx: usize) -> Result<(), Error> {
-        let index = &self.index;
-        let info = match index.chunk_info(idx) {
-            Some(info) => info,
-            None => bail!("chunk index out of range"),
-        };
-
-        // fixme: avoid copy
-
-        let data = self.store.read_chunk(&info.digest)?;
-        let size = info.range.end - info.range.start;
-        if size != data.len() as u64 {
-            bail!("read chunk with wrong size ({} != {}", size, data.len());
-        }
-
-        self.read_buffer.clear();
-        self.read_buffer.extend_from_slice(&data);
-
-        self.buffered_chunk_idx = idx;
-
-        self.buffered_chunk_start = info.range.start as u64;
-        Ok(())
-    }
-}
-
-impl<S: ReadChunk> crate::tools::BufferedRead for BufferedFixedReader<S> {
-    fn buffered_read(&mut self, offset: u64) -> Result<&[u8], Error> {
-        if offset == self.archive_size {
-            return Ok(&self.read_buffer[0..0]);
-        }
-
-        let buffer_len = self.read_buffer.len();
-        let index = &self.index;
-
-        // optimization for sequential read
-        if buffer_len > 0
-            && ((self.buffered_chunk_idx + 1) < index.index_length)
-            && (offset >= (self.buffered_chunk_start + (self.read_buffer.len() as u64)))
-        {
-            let next_idx = self.buffered_chunk_idx + 1;
-            let next_end = index.chunk_end(next_idx);
-            if offset < next_end {
-                self.buffer_chunk(next_idx)?;
-                let buffer_offset = (offset - self.buffered_chunk_start) as usize;
-                return Ok(&self.read_buffer[buffer_offset..]);
-            }
-        }
-
-        if (buffer_len == 0)
-            || (offset < self.buffered_chunk_start)
-            || (offset >= (self.buffered_chunk_start + (self.read_buffer.len() as u64)))
-        {
-            let idx = (offset / index.chunk_size as u64) as usize;
-            self.buffer_chunk(idx)?;
-        }
-
-        let buffer_offset = (offset - self.buffered_chunk_start) as usize;
-        Ok(&self.read_buffer[buffer_offset..])
-    }
-}
-
-impl<S: ReadChunk> std::io::Read for BufferedFixedReader<S> {
-    fn read(&mut self, buf: &mut [u8]) -> Result<usize, std::io::Error> {
-        use crate::tools::BufferedRead;
-        use std::io::{Error, ErrorKind};
-
-        let data = match self.buffered_read(self.read_offset) {
-            Ok(v) => v,
-            Err(err) => return Err(Error::new(ErrorKind::Other, err.to_string())),
-        };
-
-        let n = if data.len() > buf.len() {
-            buf.len()
-        } else {
-            data.len()
-        };
-
-        unsafe {
-            std::ptr::copy_nonoverlapping(data.as_ptr(), buf.as_mut_ptr(), n);
-        }
-
-        self.read_offset += n as u64;
-
-        Ok(n)
-    }
-}
-
-impl<S: ReadChunk> Seek for BufferedFixedReader<S> {
-    fn seek(&mut self, pos: SeekFrom) -> Result<u64, std::io::Error> {
-        let new_offset = match pos {
-            SeekFrom::Start(start_offset) => start_offset as i64,
-            SeekFrom::End(end_offset) => (self.archive_size as i64) + end_offset,
-            SeekFrom::Current(offset) => (self.read_offset as i64) + offset,
-        };
-
-        use std::io::{Error, ErrorKind};
-        if (new_offset < 0) || (new_offset > (self.archive_size as i64)) {
-            return Err(Error::new(
-                ErrorKind::Other,
-                format!(
-                    "seek is out of range {} ([0..{}])",
-                    new_offset, self.archive_size
-                ),
-            ));
-        }
-        self.read_offset = new_offset as u64;
-
-        Ok(self.read_offset)
-    }
-}
-- 
2.20.1

[pbs-devel] [PATCH v2 backup-qemu 4/5] use AsyncIndexReader for read_image_at

[Stefan Reiter]

Replaces the removed BufferedFixedReader and makes the API fully async
(not block_in_place or block_on which could break with many requests at
once).

A tokio::sync::Mutex needs to be used for exclusive access, since a
regular one would not work with async/await calls.

Signed-off-by: Stefan Reiter <s.reiter@proxmox.com>
---
 src/restore.rs | 21 ++++++++++-----------
 1 file changed, 10 insertions(+), 11 deletions(-)

diff --git a/src/restore.rs b/src/restore.rs
index 3e37066..2be0295 100644
--- a/src/restore.rs
+++ b/src/restore.rs
@@ -1,13 +1,14 @@
 use std::sync::{Arc, Mutex};
 use std::collections::HashMap;
-use std::io::{Read, Seek, SeekFrom};
+use std::io::SeekFrom;
 use std::convert::TryInto;
 
 use anyhow::{format_err, bail, Error};
 use once_cell::sync::OnceCell;
 use tokio::runtime::Runtime;
+use tokio::prelude::*;
 
-use proxmox_backup::tools::runtime::{get_runtime_with_builder, block_in_place};
+use proxmox_backup::tools::runtime::get_runtime_with_builder;
 use proxmox_backup::backup::*;
 use proxmox_backup::client::{HttpClient, HttpClientOptions, BackupReader, RemoteChunkReader};
 
@@ -16,7 +17,7 @@ use crate::registry::Registry;
 use crate::capi_types::DataPointer;
 
 struct ImageAccessInfo {
-    reader: Arc<Mutex<BufferedFixedReader<RemoteChunkReader>>>,
+    reader: Arc<tokio::sync::Mutex<AsyncIndexReader<RemoteChunkReader, FixedIndexReader>>>,
     _archive_name: String,
     archive_size: u64,
 }
@@ -231,12 +232,12 @@ impl RestoreTask {
         let index = client.download_fixed_index(&manifest, &archive_name).await?;
         let archive_size = index.index_bytes();
 
-        let reader = BufferedFixedReader::new(index, chunk_reader);
+        let reader = AsyncIndexReader::new(index, chunk_reader);
 
         let info = ImageAccessInfo {
             archive_size,
             _archive_name: archive_name, /// useful to debug
-            reader: Arc::new(Mutex::new(reader)),
+            reader: Arc::new(tokio::sync::Mutex::new(reader)),
         };
 
         (*self.image_registry.lock().unwrap()).register(info)
@@ -260,12 +261,10 @@ impl RestoreTask {
             bail!("read index {} out of bounds {}", offset, image_size);
         }
 
-        let bytes = block_in_place(|| {
-            let mut reader = reader.lock().unwrap();
-            reader.seek(SeekFrom::Start(offset))?;
-            let buf: &mut [u8] = unsafe { std::slice::from_raw_parts_mut(data.0 as *mut u8, size as usize)};
-            reader.read(buf)
-        })?;
+        let mut reader = reader.lock().await;
+        reader.seek(SeekFrom::Start(offset)).await?;
+        let buf: &mut [u8] = unsafe { std::slice::from_raw_parts_mut(data.0 as *mut u8, size as usize)};
+        let bytes = reader.read(buf).await?;
 
         Ok(bytes.try_into()?)
     }
-- 
2.20.1

[pbs-devel] [PATCH v2 backup-qemu 5/5] read_image_at: iterate until buffer is filled

[Stefan Reiter]

QEMU will always assume EOF when less bytes than requested are returned
by a block drivers 'read' interface, so we need to fill the buffer up to
'size' if possible.

Signed-off-by: Stefan Reiter <s.reiter@proxmox.com>
---

v2 note: this was previously a QEMU patch, but honestly that's stupid so let's
do it in Rust instead.

 current-api.h  |  4 ++--
 src/lib.rs     |  4 ++--
 src/restore.rs | 20 ++++++++++++++++----
 3 files changed, 20 insertions(+), 8 deletions(-)

diff --git a/current-api.h b/current-api.h
index 15bb275..d77eff6 100644
--- a/current-api.h
+++ b/current-api.h
@@ -364,8 +364,8 @@ int proxmox_restore_read_image_at(ProxmoxRestoreHandle *handle,
  * Note: The data pointer needs to be valid until the async
  * opteration is finished.
  *
- * Note: It is not an error for a successful call to transfer fewer
- * bytes than requested.
+ * Note: The call will only ever transfer less than 'size' bytes if
+ * the end of the file has been reached.
  */
 void proxmox_restore_read_image_at_async(ProxmoxRestoreHandle *handle,
                                          uint8_t aid,
diff --git a/src/lib.rs b/src/lib.rs
index d4b9370..3346be8 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -934,8 +934,8 @@ pub extern "C" fn proxmox_restore_read_image_at(
 /// Note: The data pointer needs to be valid until the async
 /// opteration is finished.
 ///
-/// Note: It is not an error for a successful call to transfer fewer
-/// bytes than requested.
+/// Note: The call will only ever transfer less than 'size' bytes if
+/// the end of the file has been reached.
 #[no_mangle]
 #[allow(clippy::not_unsafe_ptr_arg_deref)]
 pub extern "C" fn proxmox_restore_read_image_at_async(
diff --git a/src/restore.rs b/src/restore.rs
index 2be0295..e43d040 100644
--- a/src/restore.rs
+++ b/src/restore.rs
@@ -262,10 +262,22 @@ impl RestoreTask {
         }
 
         let mut reader = reader.lock().await;
-        reader.seek(SeekFrom::Start(offset)).await?;
-        let buf: &mut [u8] = unsafe { std::slice::from_raw_parts_mut(data.0 as *mut u8, size as usize)};
-        let bytes = reader.read(buf).await?;
 
-        Ok(bytes.try_into()?)
+        let buf: &mut [u8] = unsafe { std::slice::from_raw_parts_mut(data.0 as *mut u8, size as usize)};
+        let mut read = 0;
+
+        while read < size {
+            reader.seek(SeekFrom::Start(offset + read)).await?;
+            let bytes = reader.read(&mut buf[read as usize..]).await?;
+
+            if bytes == 0 {
+                // EOF
+                break;
+            }
+
+            read += bytes as u64;
+        }
+
+        Ok(read.try_into()?)
     }
 }
-- 
2.20.1

Re: [pbs-devel] [PATCH v2 backup 1/5] add and implement chunk_from_offset for IndexFile

[Thomas Lamprecht]

On 22.07.20 15:56, Stefan Reiter wrote:
> Necessary for byte-wise seeking through chunks in an index.
> 
> Signed-off-by: Stefan Reiter <s.reiter@proxmox.com>
> ---
>  src/backup/dynamic_index.rs | 18 ++++++++++++++++++
>  src/backup/fixed_index.rs   | 11 +++++++++++
>  src/backup/index.rs         |  3 +++
>  3 files changed, 32 insertions(+)
> 
> diff --git a/src/backup/dynamic_index.rs b/src/backup/dynamic_index.rs
> index 4907fe1f..887b7cf3 100644
> --- a/src/backup/dynamic_index.rs
> +++ b/src/backup/dynamic_index.rs
> @@ -216,6 +216,24 @@ impl IndexFile for DynamicIndexReader {
>              digest: self.index[pos].digest.clone(),
>          })
>      }
> +
> +    fn chunk_from_offset(&self, offset: u64) -> Option<(usize, u64)> {
> +        let end_idx = self.index.len() - 1;
> +        let end = self.chunk_end(end_idx);
> +        let found_idx = self.binary_search(0, 0, end_idx, end, offset);
> +        let found_idx = match found_idx {
> +            Ok(i) => i,
> +            Err(_) => return None
> +        };
> +
> +        let found_start = if found_idx == 0 {
> +            0
> +        } else {
> +            self.chunk_end(found_idx - 1)
> +        };
> +
> +        Some((found_idx, offset - found_start))
> +    }
>  }
>  
>  struct CachedChunk {
> diff --git a/src/backup/fixed_index.rs b/src/backup/fixed_index.rs
> index 73d0dad0..b7e785d6 100644
> --- a/src/backup/fixed_index.rs
> +++ b/src/backup/fixed_index.rs
> @@ -219,6 +219,17 @@ impl IndexFile for FixedIndexReader {
>  
>          (csum, chunk_end)
>      }
> +
> +    fn chunk_from_offset(&self, offset: u64) -> Option<(usize, u64)> {
> +        if offset >= self.size {
> +            return None;
> +        }
> +
> +        Some((
> +            (offset / self.chunk_size as u64) as usize,
> +            offset % self.chunk_size as u64

modulo is really slow, but isn't chunk_size always a 2^x and thus we can
do the same here as we do in other places:

offset & (self.chunk_size - 1)

> +        ))
> +    }
>  }
>  
>  pub struct FixedIndexWriter {
> diff --git a/src/backup/index.rs b/src/backup/index.rs
> index efdf3b54..2eab8524 100644
> --- a/src/backup/index.rs
> +++ b/src/backup/index.rs
> @@ -22,6 +22,9 @@ pub trait IndexFile {
>      fn index_bytes(&self) -> u64;
>      fn chunk_info(&self, pos: usize) -> Option<ChunkReadInfo>;
>  
> +    /// Get the chunk index and the relative offset within it for a byte offset
> +    fn chunk_from_offset(&self, offset: u64) -> Option<(usize, u64)>;
> +
>      /// Compute index checksum and size
>      fn compute_csum(&self) -> ([u8; 32], u64);
>  
> 

Re: [pbs-devel] [PATCH v2 backup 1/5] add and implement chunk_from_offset for IndexFile

[Stefan Reiter]

On 7/22/20 4:16 PM, Thomas Lamprecht wrote:
> On 22.07.20 15:56, Stefan Reiter wrote:
>> Necessary for byte-wise seeking through chunks in an index.
>>
>> Signed-off-by: Stefan Reiter <s.reiter@proxmox.com>
>> ---
>>   src/backup/dynamic_index.rs | 18 ++++++++++++++++++
>>   src/backup/fixed_index.rs   | 11 +++++++++++
>>   src/backup/index.rs         |  3 +++
>>   3 files changed, 32 insertions(+)
>>
>> diff --git a/src/backup/dynamic_index.rs b/src/backup/dynamic_index.rs
>> index 4907fe1f..887b7cf3 100644
>> --- a/src/backup/dynamic_index.rs
>> +++ b/src/backup/dynamic_index.rs
>> @@ -216,6 +216,24 @@ impl IndexFile for DynamicIndexReader {
>>               digest: self.index[pos].digest.clone(),
>>           })
>>       }
>> +
>> +    fn chunk_from_offset(&self, offset: u64) -> Option<(usize, u64)> {
>> +        let end_idx = self.index.len() - 1;
>> +        let end = self.chunk_end(end_idx);
>> +        let found_idx = self.binary_search(0, 0, end_idx, end, offset);
>> +        let found_idx = match found_idx {
>> +            Ok(i) => i,
>> +            Err(_) => return None
>> +        };
>> +
>> +        let found_start = if found_idx == 0 {
>> +            0
>> +        } else {
>> +            self.chunk_end(found_idx - 1)
>> +        };
>> +
>> +        Some((found_idx, offset - found_start))
>> +    }
>>   }
>>   
>>   struct CachedChunk {
>> diff --git a/src/backup/fixed_index.rs b/src/backup/fixed_index.rs
>> index 73d0dad0..b7e785d6 100644
>> --- a/src/backup/fixed_index.rs
>> +++ b/src/backup/fixed_index.rs
>> @@ -219,6 +219,17 @@ impl IndexFile for FixedIndexReader {
>>   
>>           (csum, chunk_end)
>>       }
>> +
>> +    fn chunk_from_offset(&self, offset: u64) -> Option<(usize, u64)> {
>> +        if offset >= self.size {
>> +            return None;
>> +        }
>> +
>> +        Some((
>> +            (offset / self.chunk_size as u64) as usize,
>> +            offset % self.chunk_size as u64
> 
> modulo is really slow, but isn't chunk_size always a 2^x and thus we can
> do the same here as we do in other places:
> 
> offset & (self.chunk_size - 1)
> 

I found it more readable this way and I don't think it's hot-path enough 
to make a real difference in performance.

But I don't mind, could even replace the div as well. Maybe an 
assert!(chunk_size.is_power_of_two()) might be good somewhere though.

>> +        ))
>> +    }
>>   }
>>   
>>   pub struct FixedIndexWriter {
>> diff --git a/src/backup/index.rs b/src/backup/index.rs
>> index efdf3b54..2eab8524 100644
>> --- a/src/backup/index.rs
>> +++ b/src/backup/index.rs
>> @@ -22,6 +22,9 @@ pub trait IndexFile {
>>       fn index_bytes(&self) -> u64;
>>       fn chunk_info(&self, pos: usize) -> Option<ChunkReadInfo>;
>>   
>> +    /// Get the chunk index and the relative offset within it for a byte offset
>> +    fn chunk_from_offset(&self, offset: u64) -> Option<(usize, u64)>;
>> +
>>       /// Compute index checksum and size
>>       fn compute_csum(&self) -> ([u8; 32], u64);
>>   
>>
> 

Re: [pbs-devel] [PATCH v2 backup 1/5] add and implement chunk_from_offset for IndexFile

[Thomas Lamprecht]

On 22.07.20 16:24, Stefan Reiter wrote:
> On 7/22/20 4:16 PM, Thomas Lamprecht wrote:
>> On 22.07.20 15:56, Stefan Reiter wrote:
>>> Necessary for byte-wise seeking through chunks in an index.
>>>
>>> Signed-off-by: Stefan Reiter <s.reiter@proxmox.com>
>>> ---
>>>   src/backup/dynamic_index.rs | 18 ++++++++++++++++++
>>>   src/backup/fixed_index.rs   | 11 +++++++++++
>>>   src/backup/index.rs         |  3 +++
>>>   3 files changed, 32 insertions(+)
>>>
>>> diff --git a/src/backup/dynamic_index.rs b/src/backup/dynamic_index.rs
>>> index 4907fe1f..887b7cf3 100644
>>> --- a/src/backup/dynamic_index.rs
>>> +++ b/src/backup/dynamic_index.rs
>>> @@ -216,6 +216,24 @@ impl IndexFile for DynamicIndexReader {
>>>               digest: self.index[pos].digest.clone(),
>>>           })
>>>       }
>>> +
>>> +    fn chunk_from_offset(&self, offset: u64) -> Option<(usize, u64)> {
>>> +        let end_idx = self.index.len() - 1;
>>> +        let end = self.chunk_end(end_idx);
>>> +        let found_idx = self.binary_search(0, 0, end_idx, end, offset);
>>> +        let found_idx = match found_idx {
>>> +            Ok(i) => i,
>>> +            Err(_) => return None
>>> +        };
>>> +
>>> +        let found_start = if found_idx == 0 {
>>> +            0
>>> +        } else {
>>> +            self.chunk_end(found_idx - 1)
>>> +        };
>>> +
>>> +        Some((found_idx, offset - found_start))
>>> +    }
>>>   }
>>>     struct CachedChunk {
>>> diff --git a/src/backup/fixed_index.rs b/src/backup/fixed_index.rs
>>> index 73d0dad0..b7e785d6 100644
>>> --- a/src/backup/fixed_index.rs
>>> +++ b/src/backup/fixed_index.rs
>>> @@ -219,6 +219,17 @@ impl IndexFile for FixedIndexReader {
>>>             (csum, chunk_end)
>>>       }
>>> +
>>> +    fn chunk_from_offset(&self, offset: u64) -> Option<(usize, u64)> {
>>> +        if offset >= self.size {
>>> +            return None;
>>> +        }
>>> +
>>> +        Some((
>>> +            (offset / self.chunk_size as u64) as usize,
>>> +            offset % self.chunk_size as u64
>>
>> modulo is really slow, but isn't chunk_size always a 2^x and thus we can
>> do the same here as we do in other places:
>>
>> offset & (self.chunk_size - 1)
>>
> 
> I found it more readable this way and I don't think it's hot-path enough to make a real difference in performance.

then why not add a comment instead? This is some orders of magnitude slower,
not just 2 or 3% ...

> 
> But I don't mind, could even replace the div as well. Maybe an assert!(chunk_size.is_power_of_two()) might be good somewhere though.

Yeah, somewhere in an initial entry point to such code an assert wouldn't hurt.

> 
>>> +        ))
>>> +    }
>>>   }
>>>     pub struct FixedIndexWriter {
>>> diff --git a/src/backup/index.rs b/src/backup/index.rs
>>> index efdf3b54..2eab8524 100644
>>> --- a/src/backup/index.rs
>>> +++ b/src/backup/index.rs
>>> @@ -22,6 +22,9 @@ pub trait IndexFile {
>>>       fn index_bytes(&self) -> u64;
>>>       fn chunk_info(&self, pos: usize) -> Option<ChunkReadInfo>;
>>>   +    /// Get the chunk index and the relative offset within it for a byte offset
>>> +    fn chunk_from_offset(&self, offset: u64) -> Option<(usize, u64)>;
>>> +
>>>       /// Compute index checksum and size
>>>       fn compute_csum(&self) -> ([u8; 32], u64);
>>>  
>>

[pbs-devel] applied-series: [PATCH v2 0/5] Fix PBS blockdriver for non-VM settings

[Thomas Lamprecht]

On 22.07.20 15:56, Stefan Reiter wrote:
> When using the PBS blockdriver with qemu-nbd (for example), it can happen that
> enough read requests are issued to saturate the tokio thread pool. Not an issue
> in general, but as me and Wolfgang painstakenly discovered a while back, it does
> break block_on, which is used in BufferedFixedReader. This means that reading
> larger amounts of data would hang the QEMU process [0].
> 
> Fix this by replacing the BufferedFixedReader with an AsyncIndexReader,
> implementing AsyncSeek for it in the process. This makes the entire API async,
> requiring no block_on anymore.
> 
> Incidentally, this also gave me my best benchmark results yet, coming in at
> above 1.6 Gb/s read speed via NBD on my local machine.
> 
> Additionally I discovered a seperate bug (fixed by patch 5), wherein read
> requests that we're not aligned to the chunk size would return bogus data. This
> too only seems to happen in non-VM connections (e.g. nbd, etc...).
> 
> v2:
> * Remove BufferedFixedReader entirely, use AsyncIndexReader instead
> * Implement AsyncSeek for AsyncIndexReader
> * Fix the second bug in Rust instead of QEMU C
> 
> 
> [0] ...and since the NBD kernel driver appears to be horribly broken, this often
> also crashes most of the system, but that's a different story. If you ever get
> in this situation, 'nbd-client -d /dev/nbdX' works (sometimes) to force
> disconnect the device ('qemu-nbd -d' intelligently issues a read before
> disconnecting, thus hanging before getting anything done...)
> 
> 
> backup: Stefan Reiter (3):
>   add and implement chunk_from_offset for IndexFile
>   implement AsyncSeek for AsyncIndexReader
>   remove BufferedFixedReader interface
> 
>  src/backup/async_index_reader.rs | 116 ++++++++++++++++++----
>  src/backup/dynamic_index.rs      |  18 ++++
>  src/backup/fixed_index.rs        | 165 +++----------------------------
>  src/backup/index.rs              |   4 +
>  4 files changed, 129 insertions(+), 174 deletions(-)
> 
> backup-qemu: Stefan Reiter (2):
>   use AsyncIndexReader for read_image_at
>   read_image_at: iterate until buffer is filled
> 
>  current-api.h  |  4 ++--
>  src/lib.rs     |  4 ++--
>  src/restore.rs | 35 +++++++++++++++++++++++------------
>  3 files changed, 27 insertions(+), 16 deletions(-)
> 

applied series, thanks!

using the noload mount option is really the key here, for ext4 at least ^^
# mount -o noload /dev/nbd0p1 /mnt/foo