parent
9c8707fc44
commit
9f4a0faa68
GPG Key ID:
284488987C31F630
@ -1,2 +1,3 @@
|
|||||||
/target
|
/target
|
||||||
*~
|
*~
|
||||||
|
*.dump
|
||||||
|
|||||||
@ -0,0 +1,39 @@
|
|||||||
|
//! Mechanism to defer dropping of large objects to background threads
|
||||||
|
use futures::{
|
||||||
|
prelude::*,
|
||||||
|
future::OptionFuture,
|
||||||
|
};
|
||||||
|
|
||||||
|
pub const DEFER_DROP_VEC_SIZE_FLOOR: usize = 1024 * 1024; // 1 MB
|
||||||
|
|
||||||
|
/// Drop a `Vec<T>` that is `Send` and `'static`.
|
||||||
|
///
|
||||||
|
/// This will move the object to a background task if it is deemed nessisary.
|
||||||
|
/// # Note
|
||||||
|
/// This *must* be `await`ed to work properly. If you are not in an async context, use `drop_vec_sync`.
|
||||||
|
pub fn drop_vec<T>(vec: Vec<T>) -> impl Future<Output = ()> + 'static
|
||||||
|
where T: Send + 'static
|
||||||
|
{
|
||||||
|
let len_bytes = vec.len() * std::mem::size_of::<T>();
|
||||||
|
OptionFuture::from(if len_bytes > DEFER_DROP_VEC_SIZE_FLOOR {
|
||||||
|
eprintln!("Size of vector ({} bytes, {} elements of {:?}) exceeds defer drop size floor {}. Moving vector to a seperate thread for de-allocation", len_bytes, vec.len(), std::any::type_name::<T>(), DEFER_DROP_VEC_SIZE_FLOOR);
|
||||||
|
Some(async move {
|
||||||
|
tokio::task::spawn_blocking(move || drop(vec)).await.expect("Child panic while dropping vector");
|
||||||
|
})
|
||||||
|
} else {
|
||||||
|
None
|
||||||
|
}).map(|_| ())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Drop a `Vec<T>` that is `Send` and `'static`.
|
||||||
|
///
|
||||||
|
/// This will move the object to a background task if it is deemed nessisary.
|
||||||
|
pub fn drop_vec_sync<T>(vec: Vec<T>)
|
||||||
|
where T: Send + 'static
|
||||||
|
{
|
||||||
|
let len_bytes = vec.len() * std::mem::size_of::<T>();
|
||||||
|
if len_bytes > DEFER_DROP_VEC_SIZE_FLOOR {
|
||||||
|
eprintln!("Size of vector ({} bytes, {} elements of {:?}) exceeds defer drop size floor {}. Moving vector to a seperate thread for de-allocation", len_bytes, vec.len(), std::any::type_name::<T>(), DEFER_DROP_VEC_SIZE_FLOOR);
|
||||||
|
tokio::task::spawn_blocking(move || drop(vec));
|
||||||
|
}
|
||||||
|
}
|
||||||
@ -0,0 +1,59 @@
|
|||||||
|
//! For serializing
|
||||||
|
use super::*;
|
||||||
|
use tokio::prelude::*;
|
||||||
|
|
||||||
|
use async_compression::tokio_02::write::{
|
||||||
|
BzEncoder,
|
||||||
|
BzDecoder,
|
||||||
|
};
|
||||||
|
|
||||||
|
type Compressor<T> = BzEncoder<T>;
|
||||||
|
type Decompressor<T> = BzDecoder<T>;
|
||||||
|
|
||||||
|
const DEFER_DROP_SIZE_FLOOR: usize = 1024 * 1024; // 1 MB
|
||||||
|
|
||||||
|
/// Serialise this object asynchronously
|
||||||
|
pub async fn write_async<T: Serialize, W>(mut to: W, item: &T) -> eyre::Result<()>
|
||||||
|
where W: AsyncWrite + Unpin
|
||||||
|
{
|
||||||
|
let sect_type_name = || std::any::type_name::<T>().header("Type trying to serialise was");
|
||||||
|
let sect_stream_type_name = || std::any::type_name::<W>().header("Stream type was");
|
||||||
|
|
||||||
|
let vec = tokio::task::block_in_place(|| serde_cbor::to_vec(item))
|
||||||
|
.wrap_err(eyre!("Failed to serialise item"))
|
||||||
|
.with_section(sect_stream_type_name.clone())
|
||||||
|
.with_section(sect_type_name.clone())?;
|
||||||
|
|
||||||
|
{
|
||||||
|
let mut stream = Compressor::new(&mut to);
|
||||||
|
|
||||||
|
eprintln!("Writing {} bytes of type {:?} to stream of type {:?}", vec.len(), std::any::type_name::<T>(), std::any::type_name::<W>());
|
||||||
|
|
||||||
|
stream.write_all(&vec[..])
|
||||||
|
.await
|
||||||
|
.wrap_err(eyre!("Failed to write serialised memory to stream"))
|
||||||
|
.with_section(|| vec.len().to_string().header("Size of the serialised object was"))
|
||||||
|
.with_section(sect_stream_type_name.clone())
|
||||||
|
.with_section(sect_type_name.clone())?;
|
||||||
|
|
||||||
|
stream.flush().await.wrap_err(eyre!("Failed to flush output compression stream"))?;
|
||||||
|
stream.shutdown().await.wrap_err(eyre!("Failed to shutdown output compression stream"))?;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Extremely overcomplicated concurrent flush+drop.
|
||||||
|
use futures::FutureExt;
|
||||||
|
let flush_fut = async {
|
||||||
|
to.flush().await.wrap_err(eyre!("Failed to flush output backing stream"))?;
|
||||||
|
to.shutdown().await.wrap_err(eyre!("Failed to shutdown output backing stream"))?;
|
||||||
|
Ok::<(), eyre::Report>(())
|
||||||
|
}.fuse();
|
||||||
|
|
||||||
|
tokio::pin!(flush_fut);
|
||||||
|
tokio::select!{
|
||||||
|
res = &mut flush_fut => {
|
||||||
|
return res;
|
||||||
|
}
|
||||||
|
_ = async move { drop!(async vec vec); } => {}
|
||||||
|
}
|
||||||
|
flush_fut.await
|
||||||
|
}
|
||||||
Loading…
Reference in new issue