Compare commits
8 Commits
master
...
io-uring-a
| Author | SHA1 | Date |
|---|---|---|
 Avril
|
2404721418
|
2 weeks ago |
|
Avril
|
9e5257507f
|
2 weeks ago |
|
Avril
|
216aa7abd1
|
2 weeks ago |
|
Avril
|
5293810b79
|
2 weeks ago |
|
Avril
|
0a7d530068
|
2 months ago |
|
Avril
|
066811444a
|
2 months ago |
|
Avril
|
c4fc2fde1d
|
2 months ago |
|
Avril
|
8b241a41fb
|
3 years ago |
@ -0,0 +1,688 @@
|
|||||||
|
//! Handles the chain load/save format
|
||||||
|
use super::*;
|
||||||
|
use std::{
|
||||||
|
io::{
|
||||||
|
self,
|
||||||
|
Read, Write, BufRead,
|
||||||
|
},
|
||||||
|
fmt,
|
||||||
|
};
|
||||||
|
use bytes::{
|
||||||
|
Buf, BufMut, Bytes,
|
||||||
|
};
|
||||||
|
use zstd::{
|
||||||
|
Encoder, Decoder,
|
||||||
|
};
|
||||||
|
|
||||||
|
/// The chain that can be saved / loaded.
|
||||||
|
pub type Chain<T = String> = crate::Chain<T>;
|
||||||
|
|
||||||
|
/// The version of the encoded format stream
|
||||||
|
#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord, Copy)]
|
||||||
|
#[repr(packed)]
|
||||||
|
pub struct Version(pub u8,pub u8,pub u8,pub u8);
|
||||||
|
|
||||||
|
impl fmt::Display for Version
|
||||||
|
{
|
||||||
|
#[inline]
|
||||||
|
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result
|
||||||
|
{
|
||||||
|
write!(f, "{}.{}.{}", self.0,self.1, self.2)?;
|
||||||
|
if self.3 != 0 {
|
||||||
|
write!(f, "r{}", self.3)
|
||||||
|
} else {
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
impl Version {
|
||||||
|
/// Current save version
|
||||||
|
pub const CURRENT: Self = Version(0,0,0,0);
|
||||||
|
|
||||||
|
/// Current value as a native integer
|
||||||
|
const CURRENT_VALUE: u32 = Self::CURRENT.as_native();
|
||||||
|
|
||||||
|
pub const fn as_native(&self) -> u32 {
|
||||||
|
u32::from_be_bytes([self.0, self.1, self.2, self.3])
|
||||||
|
}
|
||||||
|
|
||||||
|
#[inline]
|
||||||
|
pub const fn from_native(value: u32) -> Self {
|
||||||
|
let [a,b,c,d] = u32::to_be_bytes(value);
|
||||||
|
Self(a,b,c,d)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Default for Version
|
||||||
|
{
|
||||||
|
#[inline]
|
||||||
|
fn default() -> Self
|
||||||
|
{
|
||||||
|
Self::CURRENT
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub unsafe trait AutoBinaryFormat: Sized {
|
||||||
|
#[inline]
|
||||||
|
fn as_raw_for_encode(&self) -> *const [u8] {
|
||||||
|
let ptr = self as *const Self;
|
||||||
|
std::ptr::slice_from_raw_parts(ptr as *const u8, std::mem::size_of::<Self>())
|
||||||
|
}
|
||||||
|
#[inline]
|
||||||
|
fn as_raw_for_decode(&mut self) -> *mut [u8] {
|
||||||
|
let ptr = self as *mut Self;
|
||||||
|
std::ptr::slice_from_raw_parts_mut(ptr as *mut u8, std::mem::size_of::<Self>())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn raw_format_read_size(&mut self) -> usize {
|
||||||
|
std::mem::size_of::<Self>()
|
||||||
|
}
|
||||||
|
fn raw_format_write_size(&self) -> usize {
|
||||||
|
std::mem::size_of::<Self>()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
unsafe impl<T, const N: usize> AutoBinaryFormat for [T; N] {}
|
||||||
|
|
||||||
|
pub trait BinaryFormat {
|
||||||
|
fn read_from<S: ?Sized>(&mut self, stream: &mut S) -> io::Result<usize>
|
||||||
|
where S: io::Read;
|
||||||
|
|
||||||
|
fn write_to<S: ?Sized>(&self, stream: &mut S) -> io::Result<usize>
|
||||||
|
where S: io::Write;
|
||||||
|
|
||||||
|
fn binary_format_read_size(&mut self) -> Option<usize>;
|
||||||
|
fn binary_format_write_size(&self) -> usize;
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<T> BinaryFormat for T
|
||||||
|
where T: AutoBinaryFormat
|
||||||
|
{
|
||||||
|
#[inline(always)]
|
||||||
|
fn read_from<S: ?Sized>(&mut self, stream: &mut S) -> io::Result<usize>
|
||||||
|
where S: io::Read {
|
||||||
|
let ptr = self.as_raw_for_decode();
|
||||||
|
// SAFETY: The read data is guaranteed to be valid here.
|
||||||
|
Ok(unsafe {
|
||||||
|
stream.read_exact(&mut *ptr)?;
|
||||||
|
(*ptr).len()
|
||||||
|
})
|
||||||
|
}
|
||||||
|
#[inline(always)]
|
||||||
|
fn write_to<S: ?Sized>(&self, stream: &mut S) -> io::Result<usize>
|
||||||
|
where S: io::Write {
|
||||||
|
let ptr = self.as_raw_for_encode();
|
||||||
|
// SAFETY: The written data is guaranteed to be valid here.
|
||||||
|
Ok(unsafe {
|
||||||
|
stream.write_all(&*ptr)?;
|
||||||
|
(*ptr).len()
|
||||||
|
})
|
||||||
|
}
|
||||||
|
#[inline]
|
||||||
|
fn binary_format_read_size(&mut self) -> Option<usize> {
|
||||||
|
Some(self.raw_format_read_size())
|
||||||
|
}
|
||||||
|
#[inline]
|
||||||
|
fn binary_format_write_size(&self) -> usize {
|
||||||
|
self.raw_format_write_size()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<T> BinaryFormat for [T]
|
||||||
|
where T: BinaryFormat
|
||||||
|
{
|
||||||
|
#[inline]
|
||||||
|
fn read_from<S: ?Sized>(&mut self, stream: &mut S) -> io::Result<usize>
|
||||||
|
where S: io::Read {
|
||||||
|
let mut sz = 0;
|
||||||
|
for i in self.iter_mut() {
|
||||||
|
sz += i.read_from(stream)?;
|
||||||
|
}
|
||||||
|
Ok(sz)
|
||||||
|
}
|
||||||
|
#[inline]
|
||||||
|
fn write_to<S: ?Sized>(&self, stream: &mut S) -> io::Result<usize>
|
||||||
|
where S: io::Write {
|
||||||
|
let mut sz =0;
|
||||||
|
for i in self.iter() {
|
||||||
|
sz += i.write_to(stream)?;
|
||||||
|
}
|
||||||
|
Ok(sz)
|
||||||
|
}
|
||||||
|
#[inline]
|
||||||
|
fn binary_format_read_size(&mut self) -> Option<usize> {
|
||||||
|
self.iter_mut().map(|x| x.binary_format_read_size()).try_fold(0, |x, y| {
|
||||||
|
match (x, y) {
|
||||||
|
(x, Some(y)) => Some(x + y),
|
||||||
|
_ => None,
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
#[inline]
|
||||||
|
fn binary_format_write_size(&self) -> usize {
|
||||||
|
self.iter().map(|x| x.binary_format_write_size()).sum()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl BinaryFormat for Version {
|
||||||
|
#[inline]
|
||||||
|
fn read_from<S: ?Sized>(&mut self, stream: &mut S) -> io::Result<usize>
|
||||||
|
where S: io::Read {
|
||||||
|
let mut vi = [0u8; 4];
|
||||||
|
stream.read_exact(&mut vi[..])?;
|
||||||
|
Ok(4)
|
||||||
|
}
|
||||||
|
#[inline]
|
||||||
|
fn write_to<S: ?Sized>(&self, stream: &mut S) -> io::Result<usize>
|
||||||
|
where S: io::Write {
|
||||||
|
let vi = [self.0,self.1,self.2,self.3];
|
||||||
|
stream.write_all(&vi[..])?;
|
||||||
|
Ok(4)
|
||||||
|
|
||||||
|
}
|
||||||
|
#[inline]
|
||||||
|
fn binary_format_read_size(&mut self) -> Option<usize> {
|
||||||
|
Some(std::mem::size_of::<u8>() * 4)
|
||||||
|
}
|
||||||
|
#[inline]
|
||||||
|
fn binary_format_write_size(&self) -> usize {
|
||||||
|
std::mem::size_of::<u8>() * 4
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Clone, PartialEq, Eq, Hash, Copy, Default)]
|
||||||
|
#[repr(u32)]
|
||||||
|
pub enum Compressed {
|
||||||
|
#[default]
|
||||||
|
No = 0,
|
||||||
|
Zstd = 1,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Compressed {
|
||||||
|
|
||||||
|
#[inline]
|
||||||
|
const fn to_int(&self) -> u32 {
|
||||||
|
*self as u32
|
||||||
|
}
|
||||||
|
#[inline]
|
||||||
|
fn try_from_int(val: u32) -> Option<Self> {
|
||||||
|
match val {
|
||||||
|
// SAFETY: These variants are known
|
||||||
|
0..=1 => Some(unsafe {
|
||||||
|
std::mem::transmute(val)
|
||||||
|
}),
|
||||||
|
_ => None,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Clone, PartialEq, Eq, Hash, Default)]
|
||||||
|
pub struct FormatMetadata {
|
||||||
|
pub version: Version,
|
||||||
|
|
||||||
|
pub compressed: Compressed,
|
||||||
|
pub chain_size: usize, // NOTE: Unused
|
||||||
|
pub checksum: u64, // NOTE: Unused
|
||||||
|
}
|
||||||
|
|
||||||
|
impl FormatMetadata {
|
||||||
|
const MAGIC_NUM: &[u8; 8] = b"MARKOV\x00\xcf";
|
||||||
|
}
|
||||||
|
|
||||||
|
impl BinaryFormat for FormatMetadata
|
||||||
|
{
|
||||||
|
fn write_to<S: ?Sized>(&self, mut stream: &mut S) -> io::Result<usize>
|
||||||
|
where S: io::Write {
|
||||||
|
let sz = self.version.write_to(&mut stream)?;
|
||||||
|
|
||||||
|
let mut obuf = [0u8; std::mem::size_of::<u32>() +std::mem::size_of::<u64>() + std::mem::size_of::<u64>()];
|
||||||
|
{
|
||||||
|
let mut obuf = &mut obuf[..];
|
||||||
|
use std::convert::TryInto;
|
||||||
|
|
||||||
|
obuf.put_u32(self.compressed.to_int());
|
||||||
|
obuf.put_u64(self.chain_size.try_into().map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "Chain size attribute out-of-bounds for format size"))?);
|
||||||
|
obuf.put_u64(self.checksum);
|
||||||
|
}
|
||||||
|
stream.write_all(&obuf[..])?;
|
||||||
|
|
||||||
|
Ok(sz + obuf.len())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn read_from<S: ?Sized>(&mut self, mut stream: &mut S) -> io::Result<usize>
|
||||||
|
where S: io::Read {
|
||||||
|
let sz = self.version.read_from(&mut stream)?;
|
||||||
|
|
||||||
|
if self.version > Version::CURRENT {
|
||||||
|
return Err(io::Error::new(io::ErrorKind::Unsupported, format!("Unknown format version {}", self.version)));
|
||||||
|
}
|
||||||
|
|
||||||
|
let mut ibuf = [0u8; std::mem::size_of::<u32>() +std::mem::size_of::<u64>() + std::mem::size_of::<u64>()];
|
||||||
|
stream.read_exact(&mut ibuf[..])?;
|
||||||
|
{
|
||||||
|
let mut ibuf = &ibuf[..];
|
||||||
|
use std::convert::TryInto;
|
||||||
|
|
||||||
|
self.compressed = Compressed::try_from_int(ibuf.get_u32()).ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "Invalid compression attribute"))?;
|
||||||
|
self.chain_size = ibuf.get_u64().try_into().map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "Chain size attribute out-of-bounds for native size"))?;
|
||||||
|
self.checksum = ibuf.get_u64();
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(sz + ibuf.len())
|
||||||
|
}
|
||||||
|
|
||||||
|
#[inline]
|
||||||
|
fn binary_format_read_size(&mut self) -> Option<usize> {
|
||||||
|
let szm = self.version.binary_format_read_size()?;
|
||||||
|
Some(szm + std::mem::size_of::<u32>()
|
||||||
|
+ std::mem::size_of::<u64>()
|
||||||
|
+ std::mem::size_of::<u64>())
|
||||||
|
}
|
||||||
|
|
||||||
|
#[inline(always)]
|
||||||
|
fn binary_format_write_size(&self) -> usize {
|
||||||
|
self.version.binary_format_write_size()
|
||||||
|
+ std::mem::size_of::<u32>()
|
||||||
|
+ std::mem::size_of::<u64>()
|
||||||
|
+ std::mem::size_of::<u64>()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
/// Load a chain from a stream
|
||||||
|
#[inline]
|
||||||
|
pub fn load_chain_from_sync<S>(stream: &mut S) -> io::Result<Chain<String>>
|
||||||
|
where S: io::Read + ?Sized
|
||||||
|
{
|
||||||
|
let mut stream = io::BufReader::new(stream);
|
||||||
|
{
|
||||||
|
let mut magic = FormatMetadata::MAGIC_NUM.clone();
|
||||||
|
stream.read_exact(&mut magic[..])?;
|
||||||
|
|
||||||
|
if &magic != FormatMetadata::MAGIC_NUM {
|
||||||
|
return Err(io::Error::new(io::ErrorKind::InvalidData, "Invalid file header tag magic number"));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
let metadata = {
|
||||||
|
let mut metadata = FormatMetadata::default();
|
||||||
|
metadata.read_from(&mut stream)?;
|
||||||
|
metadata
|
||||||
|
};
|
||||||
|
|
||||||
|
match metadata.version {
|
||||||
|
Version::CURRENT => {
|
||||||
|
let read = |read: &mut (dyn io::Read)| serde_cbor::from_reader(read).expect("Failed to read chain from input stream"); // TODO: Error type
|
||||||
|
|
||||||
|
match metadata.compressed {
|
||||||
|
Compressed::No =>
|
||||||
|
Ok(read(&mut stream)),
|
||||||
|
Compressed::Zstd => {
|
||||||
|
let mut stream = zstd::Decoder::with_buffer(stream)?;
|
||||||
|
|
||||||
|
//#[cfg(feature="threads")]
|
||||||
|
//stream.multithread(num_cpus::get() as i32);
|
||||||
|
|
||||||
|
//NOTE: Not required here: //stream.finish()?;
|
||||||
|
Ok(read(&mut stream))
|
||||||
|
},
|
||||||
|
}
|
||||||
|
},
|
||||||
|
unsup => {
|
||||||
|
return Err(io::Error::new(io::ErrorKind::Unsupported, format!("Unsupported payload version {}", unsup)));
|
||||||
|
},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Save a chain to a stream with optional compression.
|
||||||
|
#[inline]
|
||||||
|
pub fn save_chain_to_sync<S>(stream: &mut S, chain: &Chain<String>, compress: bool) -> io::Result<()>
|
||||||
|
where S: io::Write + ?Sized
|
||||||
|
{
|
||||||
|
let mut stream = io::BufWriter::new(stream);
|
||||||
|
|
||||||
|
let metadata = FormatMetadata {
|
||||||
|
compressed: compress
|
||||||
|
.then_some(Compressed::Zstd)
|
||||||
|
.unwrap_or(Compressed::No),
|
||||||
|
..Default::default()
|
||||||
|
};
|
||||||
|
|
||||||
|
stream.write_all(FormatMetadata::MAGIC_NUM)?;
|
||||||
|
metadata.write_to(&mut stream)?;
|
||||||
|
|
||||||
|
let write = |stream: &mut (dyn io::Write)| serde_cbor::to_writer(stream, chain).expect("Failed to write chain to output stream"); // TODO: Error type
|
||||||
|
|
||||||
|
let mut stream = match metadata.compressed {
|
||||||
|
Compressed::No => {
|
||||||
|
write(&mut stream);
|
||||||
|
stream
|
||||||
|
},
|
||||||
|
Compressed::Zstd => {
|
||||||
|
let mut stream = zstd::Encoder::new(stream, 22)?;
|
||||||
|
|
||||||
|
#[cfg(feature="threads")]
|
||||||
|
stream.multithread(num_cpus::get() as u32)?;
|
||||||
|
|
||||||
|
write(&mut stream);
|
||||||
|
// XXX: Should we flush after write here..?
|
||||||
|
|
||||||
|
// NOTE: Required here.
|
||||||
|
stream.finish()?
|
||||||
|
},
|
||||||
|
};
|
||||||
|
stream.flush()
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(feature="io_uring")]
|
||||||
|
#[deprecated = "Runs through compression *twice* (one on the sync side, even) see below. Must be re-written to properly use `io_uring::*`"]
|
||||||
|
pub fn save_chain_to_file(output_file: std::fs::File, chain: &Chain<String>, compress: bool) -> io::Result<()>
|
||||||
|
{
|
||||||
|
compile_error!("TODO: Make `save_to_chain() that takes callback for write stream to create compression stream & write to it, since io_uring backing takes care of that, but we want the uncompressed metadata writing to be the same.");
|
||||||
|
// let output_file = fs::OpenOptions::new()
|
||||||
|
// .create_new(! force)
|
||||||
|
// .create(true)
|
||||||
|
// .truncate(force)
|
||||||
|
// .write(true)
|
||||||
|
// .open(path);
|
||||||
|
|
||||||
|
let (mut stream, bg) = io_uring::create_write_compressed(output_file)?;
|
||||||
|
todo!("^^^ Ehh... No, this isn't right... `stream` is the *encoder* stream, we still need to do the metadata thing done in `_sync`()... XXX: Can we factor that out to something that will take a callback maybe??? Idk... im tired...");
|
||||||
|
|
||||||
|
save_chain_to_sync(&mut stream, chain, compress)?;
|
||||||
|
stream.flush()?;
|
||||||
|
drop(stream); // NOTE: Drop the sending side so the recv side can know there is no more data. **MUST** be done before the `.join()`.
|
||||||
|
|
||||||
|
bg.join().expect("Fatal error in background I/O thread")?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(feature="io_uring")]
|
||||||
|
mod io_uring {
|
||||||
|
use super::*;
|
||||||
|
use std::{
|
||||||
|
path::Path,
|
||||||
|
};
|
||||||
|
use futures::{
|
||||||
|
prelude::*,
|
||||||
|
io::{
|
||||||
|
//AsyncRead,
|
||||||
|
//AsyncBufRead,
|
||||||
|
//AsyncWrite,
|
||||||
|
|
||||||
|
AllowStdIo,
|
||||||
|
},
|
||||||
|
};
|
||||||
|
use tokio::{
|
||||||
|
io::{
|
||||||
|
AsyncRead,
|
||||||
|
AsyncBufRead,
|
||||||
|
AsyncWrite,
|
||||||
|
|
||||||
|
ReadHalf, WriteHalf,
|
||||||
|
SimplexStream,
|
||||||
|
simplex,
|
||||||
|
|
||||||
|
AsyncReadExt,
|
||||||
|
AsyncWriteExt,
|
||||||
|
},
|
||||||
|
|
||||||
|
};
|
||||||
|
use tokio_uring::{
|
||||||
|
fs,
|
||||||
|
buf,
|
||||||
|
};
|
||||||
|
use async_compression::{
|
||||||
|
Level,
|
||||||
|
zstd::CParameter,
|
||||||
|
|
||||||
|
tokio::{
|
||||||
|
bufread::{
|
||||||
|
ZstdDecoder,
|
||||||
|
ZstdEncoder,
|
||||||
|
},
|
||||||
|
write as zstd_write,
|
||||||
|
},
|
||||||
|
};
|
||||||
|
// Add `tokio_uring` version of `save_chain_to_file()`/`load_chain_from_file()` that spawns the `tokio_uring` runtime internally to queue reads/writes from/to a file. (Maybe default-feature-gate this?)
|
||||||
|
|
||||||
|
/// Creates an future that, when executed, reads all data from `reader` via `io_uring`, into generic byte sink `writer`.
|
||||||
|
///
|
||||||
|
/// # Task dispatch
|
||||||
|
/// This function *creates* the async function, it should be executed via `spawn()` *inside* a `tokio_uring` context.
|
||||||
|
///
|
||||||
|
/// # Returns
|
||||||
|
/// The future polls to return an `io::Result<usize>` of the number of bytes read from `reader` and written to `writer`.
|
||||||
|
#[inline]
|
||||||
|
#[must_use]
|
||||||
|
fn create_read_pump<'f, B>(reader: &'f fs::File, writer: B) -> impl Future<Output = std::io::Result<usize>> + use<'f, B>
|
||||||
|
where B: AsyncWrite
|
||||||
|
{
|
||||||
|
async move {
|
||||||
|
let mut pos = 0;
|
||||||
|
let mut buf = Vec::with_capacity(4096);
|
||||||
|
|
||||||
|
futures::pin_mut!(writer);
|
||||||
|
|
||||||
|
loop {
|
||||||
|
let (bytes, nbuf) = reader.read_at(buf, pos as u64).await;
|
||||||
|
match bytes? {
|
||||||
|
0 => break,
|
||||||
|
bytes => {
|
||||||
|
// Push `&nbuf[..bytes]` to pump.
|
||||||
|
writer.write_all(&nbuf[..bytes]).await?;
|
||||||
|
|
||||||
|
pos += bytes;
|
||||||
|
buf = nbuf;
|
||||||
|
},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Ok::<_, std::io::Error>(pos)
|
||||||
|
}//).map(|_| ())
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
/// Creates an future that, when executed, reads all data from generic byte source `reader` and writes them into `writer` through `io_uring`.
|
||||||
|
///
|
||||||
|
/// # Task dispatch
|
||||||
|
/// This function *creates* the async function, it should be executed via `spawn()` *inside* a `tokio_uring` context.
|
||||||
|
///
|
||||||
|
/// # Returns
|
||||||
|
/// The future polls to return an `io::Result<usize>` of the number of bytes read from `reader` and written to `writer`.
|
||||||
|
fn create_write_pump<'f, B>(reader: B, writer: &'f fs::File) -> impl Future<Output = std::io::Result<usize>> + use<'f, B>
|
||||||
|
where B: AsyncRead
|
||||||
|
{
|
||||||
|
async move {
|
||||||
|
let mut pos =0;
|
||||||
|
let mut buf = vec![0u8; 4096];
|
||||||
|
|
||||||
|
futures::pin_mut!(reader);
|
||||||
|
|
||||||
|
loop {
|
||||||
|
let sz = reader.read(&mut buf[..]).await?;
|
||||||
|
if sz == 0 {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
buf.truncate(sz);
|
||||||
|
|
||||||
|
let (res, nbuf) = writer.write_all_at(buf, pos as u64).await;
|
||||||
|
res?;
|
||||||
|
pos += sz;
|
||||||
|
buf = nbuf;
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok::<usize, std::io::Error>(pos)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Spawns a task that reads all bytes from `reader` via `io_uring` and writes them into generic byte sink `writer`.
|
||||||
|
/// Returns a handle to the async task that completes when the task has completed.
|
||||||
|
///
|
||||||
|
/// # Runtime
|
||||||
|
/// The file is closed after the operation completes, ownership is transfered into the background task.
|
||||||
|
///
|
||||||
|
/// __NOTE__: **Must** be called from within an active `tokio_uring` runtime.
|
||||||
|
fn spawn_read_pump<B>(reader: fs::File, writer: B) -> impl Future<Output = std::io::Result<usize>> + Unpin + 'static
|
||||||
|
where B: AsyncWrite + Send + 'static
|
||||||
|
{
|
||||||
|
tokio_uring::spawn(async move {
|
||||||
|
let sz = create_read_pump(&reader, writer).await?;
|
||||||
|
let _ = reader.sync_data().await;
|
||||||
|
reader.close().await.map(move |_| sz)
|
||||||
|
}).map(|handle| handle.expect("Background reading task panic"))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Spawns a task that reads all bytes from generic byte source `reader` and writes them into `writer` via `io_uring`.
|
||||||
|
/// Returns a handle to the async task that completes when the task has completed.
|
||||||
|
///
|
||||||
|
/// # Runtime
|
||||||
|
/// The file is closed after the operation completes, ownership is transfered into the background task.
|
||||||
|
///
|
||||||
|
/// __NOTE__: **Must** be called from within an active `tokio_uring` runtime.
|
||||||
|
fn spawn_write_pump<B>(reader: B, writer: fs::File) -> impl Future<Output = std::io::Result<usize>> + Unpin + 'static
|
||||||
|
where B: AsyncRead + Send + 'static
|
||||||
|
{
|
||||||
|
tokio_uring::spawn(async move {
|
||||||
|
let sz = create_write_pump(reader, &writer).await?;
|
||||||
|
let _ = writer.sync_data().await;
|
||||||
|
writer.close().await.map(move |_| sz)
|
||||||
|
}).map(|handle| handle.expect("Background writing task panic"))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Create a wrapper over `output_stream` in which bytes written are compressed via `zstd`.
|
||||||
|
fn create_write_encoder<S>(output_stream: S) -> zstd_write::ZstdEncoder<S>
|
||||||
|
where S: AsyncWrite
|
||||||
|
{
|
||||||
|
use zstd_write::*;
|
||||||
|
if cfg!(feature="threads") {
|
||||||
|
ZstdEncoder::with_quality_and_params(output_stream, Level::Precise(22), &[CParameter::nb_workers(num_cpus::get() as u32)])
|
||||||
|
} else {
|
||||||
|
ZstdEncoder::with_quality(output_stream, Level::Precise(22))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Creates a wrapper over `output_stream` in which bytes written to are **de**compressed via `zstd`.
|
||||||
|
fn create_write_decoder<S>(output_stream: S) -> zstd_write::ZstdDecoder<S>
|
||||||
|
where S: AsyncWrite
|
||||||
|
{
|
||||||
|
use zstd_write::*;
|
||||||
|
ZstdDecoder::new(output_stream)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Create a wrapper over `output_stream` in which bytes read are compressed via `zstd`.
|
||||||
|
fn create_read_encoder<S>(output_stream: S) -> ZstdEncoder<S>
|
||||||
|
where S: AsyncBufRead
|
||||||
|
{
|
||||||
|
if cfg!(feature="threads") {
|
||||||
|
ZstdEncoder::with_quality_and_params(output_stream, Level::Precise(22), &[CParameter::nb_workers(num_cpus::get() as u32)])
|
||||||
|
} else {
|
||||||
|
ZstdEncoder::with_quality(output_stream, Level::Precise(22))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Creates a wrapper over `output_stream` in which bytes read from are **de**compressed via `zstd`.
|
||||||
|
fn create_read_decoder<S>(output_stream: S) -> ZstdDecoder<S>
|
||||||
|
where S: AsyncBufRead
|
||||||
|
{
|
||||||
|
ZstdDecoder::new(output_stream)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn pipe_async_uring() -> std::io::Result<(fs::File, fs::File)>
|
||||||
|
{
|
||||||
|
use std::os::unix::io::*;
|
||||||
|
|
||||||
|
let (rx, tx) = os_pipe::pipe()?;
|
||||||
|
|
||||||
|
let tx = unsafe {
|
||||||
|
std::fs::File::from_raw_fd(tx.into_raw_fd())
|
||||||
|
};
|
||||||
|
let rx = unsafe {
|
||||||
|
std::fs::File::from_raw_fd(rx.into_raw_fd())
|
||||||
|
};
|
||||||
|
|
||||||
|
Ok((fs::File::from_std(rx), fs::File::from_std(tx)))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Create a pipe *reader* that decompresses the bytes read from `input_file` on a backing task.
|
||||||
|
///
|
||||||
|
/// Must be called within a `tokio_uring` context.
|
||||||
|
///
|
||||||
|
/// # Returns
|
||||||
|
/// A tuple of:
|
||||||
|
/// - Backing task join-handle. It will complete when all bytes have been copied (__XXX__: and the returned read stream has been closed? Does it?)
|
||||||
|
/// - Read end of async pipe. When all data has been read, it should be closed and then the handle should be awaited to ensure the data is flushed.
|
||||||
|
fn pipe_to_decoder(input_file: fs::File) -> (impl Future<Output = std::io::Result<usize>> + Unpin + Send + 'static, ReadHalf<SimplexStream>)
|
||||||
|
{
|
||||||
|
let (rx, tx) = simplex(4096 << 1);
|
||||||
|
let tx = create_write_decoder(tx);
|
||||||
|
|
||||||
|
let pump = spawn_read_pump(input_file, tx);
|
||||||
|
|
||||||
|
(pump, rx)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Create a pipe *writer* that compresses the bytes written into `output_file` on a backing task.
|
||||||
|
///
|
||||||
|
/// Must be called within a `tokio_uring` context.
|
||||||
|
///
|
||||||
|
/// # Returns
|
||||||
|
/// A tuple of:
|
||||||
|
/// - Backing task join-handle. It will complete when all bytes have been copied and the returned write stream has been closed.
|
||||||
|
/// - Write end of async pipe. When all data has been written, it should be closed and then the handle should be awaited to ensure the data is flushed.
|
||||||
|
fn pipe_to_encoder(output_file: fs::File) -> (impl Future<Output = std::io::Result<usize>> + Unpin + Send + 'static, WriteHalf<SimplexStream>)
|
||||||
|
{
|
||||||
|
let (rx, tx) = simplex(4096 << 1);
|
||||||
|
let rx = create_read_encoder(tokio::io::BufReader::new(rx));
|
||||||
|
|
||||||
|
let pump = spawn_write_pump(rx, output_file);
|
||||||
|
|
||||||
|
(pump, tx)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Write `data` to file `output` using `io_uring` (overwriting if `force` is `true`.)
|
||||||
|
///
|
||||||
|
/// # Synchonicity
|
||||||
|
/// This runs on a background thread, the encoding of the object bytes into the stream can be done on the current thread, the compression and writing to the file is done on a backing thread.
|
||||||
|
///
|
||||||
|
/// # Returns
|
||||||
|
/// A synchonous writer to write the data to, and a `JoinHandle` to the backing thread that completes to the number of bytes written to the file.
|
||||||
|
pub fn create_write_compressed(output_file: std::fs::File) -> std::io::Result<(os_pipe::PipeWriter, std::thread::JoinHandle<std::io::Result<usize>>)>
|
||||||
|
{
|
||||||
|
let (rx, tx) = os_pipe::pipe()?;
|
||||||
|
|
||||||
|
let b = std::thread::spawn(move || {
|
||||||
|
tokio_uring::start(async move {
|
||||||
|
// let output_file = fs::OpenOptions::new()
|
||||||
|
// .create_new(! force)
|
||||||
|
// .create(true)
|
||||||
|
// .truncate(force)
|
||||||
|
// .write(true)
|
||||||
|
// .open(output).await?;
|
||||||
|
let output_file = fs::File::from_std(output_file);
|
||||||
|
|
||||||
|
let (bg, tx) = pipe_to_encoder(output_file);
|
||||||
|
let rx = fs::File::from_std(unsafe {
|
||||||
|
use std::os::unix::io::*;
|
||||||
|
std::fs::File::from_raw_fd(rx.into_raw_fd())
|
||||||
|
});
|
||||||
|
|
||||||
|
let bgt = spawn_read_pump(rx, tx);
|
||||||
|
// Await the two spawned tasks together.
|
||||||
|
let (sz, szf) = futures::future::try_join(bgt, bg).await?;
|
||||||
|
|
||||||
|
if sz != szf {
|
||||||
|
//XXX: Should we expect these to be the same?
|
||||||
|
panic!("Invalid size transfer! {} bytes sent to backing, {} bytes sent to file.", sz, szf);
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok::<_, std::io::Error>(szf)
|
||||||
|
})
|
||||||
|
});
|
||||||
|
Ok((tx, b))
|
||||||
|
}
|
||||||
|
|
||||||
|
//TODO: create_read_compressed()
|
||||||
|
}
|
||||||
Loading…
Reference in new issue