LemmyNet/pict-rs
2
0
Fork 0
mirror of https://git.asonix.dog/asonix/pict-rs synced 2024-12-22 19:31:35 +00:00
Code Releases Activity

Defensive tracing

Browse source
This commit is contained in:
Aode (lion) 2022-04-07 12:56:40 -05:00
parent c80d207a87
commit e493e90dd4
10 changed files with 111 additions and 74 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
src/backgrounded.rs
View file

@ -76,16 +76,20 @@ where
if let Some(identifier) = self.identifier.take() {
let repo = self.repo.clone();
actix_rt::spawn(async move {
let _ = crate::queue::cleanup_identifier(&repo, identifier).await;
tracing::trace_span!(parent: None, "Spawn task").in_scope(|| {
actix_rt::spawn(async move {
let _ = crate::queue::cleanup_identifier(&repo, identifier).await;
})
});
}
if let Some(upload_id) = self.upload_id {
let repo = self.repo.clone();
actix_rt::spawn(async move {
let _ = repo.claim(upload_id).await;
tracing::trace_span!(parent: None, "Spawn task").in_scope(|| {
actix_rt::spawn(async move {
let _ = repo.claim(upload_id).await;
})
});
}
}

4
src/concurrent_processor.rs
View file

@ -58,7 +58,9 @@ where
(None, span)
}
Entry::Occupied(mut occupied) => {
let (tx, rx) = tokio::sync::oneshot::channel();
let (tx, rx) = tracing::trace_span!(parent: None, "Create channel")
.in_scope(tokio::sync::oneshot::channel);
occupied.get_mut().push(tx);
let span = tracing::info_span!(
"Waiting for processed image",

16
src/file.rs
View file

@ -20,13 +20,17 @@ mod tokio_file {
impl File {
pub(crate) async fn open(path: impl AsRef<Path>) -> std::io::Result<Self> {
Ok(File {
inner: tokio::fs::File::open(path).await?,
inner: tracing::trace_span!(parent: None, "Open File")
.in_scope(|| tokio::fs::File::open(path))
.await?,
})
}
pub(crate) async fn create(path: impl AsRef<Path>) -> std::io::Result<Self> {
Ok(File {
inner: tokio::fs::File::create(path).await?,
inner: tracing::trace_span!(parent: None, "Create File")
.in_scope(|| tokio::fs::File::create(path))
.await?,
})
}
@ -125,7 +129,9 @@ mod io_uring {
tracing::info!("Opening io-uring file: {:?}", path.as_ref());
Ok(File {
path: path.as_ref().to_owned(),
inner: tokio_uring::fs::File::open(path).await?,
inner: tracing::trace_span!(parent: None, "Open File")
.in_scope(|| tokio_uring::fs::File::open(path))
.await?,
})
}
@ -133,7 +139,9 @@ mod io_uring {
tracing::info!("Creating io-uring file: {:?}", path.as_ref());
Ok(File {
path: path.as_ref().to_owned(),
inner: tokio_uring::fs::File::create(path).await?,
inner: tracing::trace_span!(parent: None, "Create File")
.in_scope(|| tokio_uring::fs::File::create(path))
.await?,
})
}

10
src/generate.rs
View file

@ -10,6 +10,7 @@ use crate::{
use actix_web::web::Bytes;
use std::path::PathBuf;
use tokio::io::AsyncReadExt;
use tracing::Instrument;
#[tracing::instrument(skip(hash))]
pub(crate) async fn generate<R: FullRepo, S: Store + 'static>(
@ -47,9 +48,7 @@ async fn process<R: FullRepo, S: Store + 'static>(
thumbnail_args: Vec<String>,
hash: R::Bytes,
) -> Result<(Details, Bytes), Error> {
let permit = tracing::trace_span!(parent: None, "Aquire semaphore")
.in_scope(|| crate::PROCESS_SEMAPHORE.acquire())
.await;
let permit = crate::PROCESS_SEMAPHORE.acquire().await;
let identifier = if let Some(identifier) = repo
.still_identifier_from_alias::<S::Identifier>(&alias)
@ -77,7 +76,10 @@ async fn process<R: FullRepo, S: Store + 'static>(
crate::magick::process_image_store_read(store.clone(), identifier, thumbnail_args, format)?;
let mut vec = Vec::new();
processed_reader.read_to_end(&mut vec).await?;
processed_reader
.read_to_end(&mut vec)
.instrument(tracing::info_span!("Reading processed image to vec"))
.await?;
let bytes = Bytes::from(vec);
drop(permit);

32
src/ingest.rs
View file

@ -62,9 +62,7 @@ where
R: FullRepo + 'static,
S: Store,
{
let permit = tracing::trace_span!(parent: None, "Aquire semaphore")
.in_scope(|| crate::PROCESS_SEMAPHORE.acquire())
.await;
let permit = crate::PROCESS_SEMAPHORE.acquire().await;
let bytes = aggregate(stream).await?;
@ -205,31 +203,37 @@ where
fn drop(&mut self) {
if let Some(hash) = self.hash.take() {
let repo = self.repo.clone();
actix_rt::spawn(async move {
let _ = crate::queue::cleanup_hash(&repo, hash.into()).await;
tracing::trace_span!(parent: None, "Spawn task").in_scope(|| {
actix_rt::spawn(async move {
let _ = crate::queue::cleanup_hash(&repo, hash.into()).await;
})
});
}
if let Some(alias) = self.alias.take() {
let repo = self.repo.clone();
actix_rt::spawn(async move {
if let Ok(token) = repo.delete_token(&alias).await {
let _ = crate::queue::cleanup_alias(&repo, alias, token).await;
} else {
let token = DeleteToken::generate();
if let Ok(Ok(())) = repo.relate_delete_token(&alias, &token).await {
tracing::trace_span!(parent: None, "Spawn task").in_scope(|| {
actix_rt::spawn(async move {
if let Ok(token) = repo.delete_token(&alias).await {
let _ = crate::queue::cleanup_alias(&repo, alias, token).await;
} else {
let token = DeleteToken::generate();
if let Ok(Ok(())) = repo.relate_delete_token(&alias, &token).await {
let _ = crate::queue::cleanup_alias(&repo, alias, token).await;
}
}
}
})
});
}
if let Some(identifier) = self.identifier.take() {
let repo = self.repo.clone();
actix_rt::spawn(async move {
let _ = crate::queue::cleanup_identifier(&repo, identifier).await;
tracing::trace_span!(parent: None, "Spawn task").in_scope(|| {
actix_rt::spawn(async move {
let _ = crate::queue::cleanup_identifier(&repo, identifier).await;
})
});
}
}

3
src/ingest/hasher.rs
View file

@ -70,7 +70,8 @@ mod test {
actix_rt::System::new().block_on(async move {
let arbiter = actix_rt::Arbiter::new();
let (tx, rx) = tokio::sync::oneshot::channel();
let (tx, rx) = tracing::trace_span!(parent: None, "Create channel")
.in_scope(|| tokio::sync::oneshot::channel());
arbiter.spawn(async move {
let handle = actix_rt::spawn($fut);

30
src/main.rs
View file

@ -75,8 +75,10 @@ static DO_CONFIG: Lazy<(Configuration, Operation)> =
Lazy::new(|| config::configure().expect("Failed to configure"));
static CONFIG: Lazy<Configuration> = Lazy::new(|| DO_CONFIG.0.clone());
static OPERATION: Lazy<Operation> = Lazy::new(|| DO_CONFIG.1.clone());
static PROCESS_SEMAPHORE: Lazy<Semaphore> =
Lazy::new(|| Semaphore::new(num_cpus::get().saturating_sub(1).max(1)));
static PROCESS_SEMAPHORE: Lazy<Semaphore> = Lazy::new(|| {
tracing::trace_span!(parent: None, "Initialize semaphore")
.in_scope(|| Semaphore::new(num_cpus::get().saturating_sub(1).max(1)))
});
/// Handle responding to succesful uploads
#[instrument(name = "Uploaded files", skip(value))]
@ -790,16 +792,20 @@ async fn launch<R: FullRepo + Clone + 'static, S: Store + Clone + 'static>(
let store = store.clone();
let repo = repo.clone();
actix_rt::spawn(queue::process_cleanup(
repo.clone(),
store.clone(),
next_worker_id(),
));
actix_rt::spawn(queue::process_images(
repo.clone(),
store.clone(),
next_worker_id(),
));
tracing::trace_span!(parent: None, "Spawn task").in_scope(|| {
actix_rt::spawn(queue::process_cleanup(
repo.clone(),
store.clone(),
next_worker_id(),
))
});
tracing::trace_span!(parent: None, "Spawn task").in_scope(|| {
actix_rt::spawn(queue::process_images(
repo.clone(),
store.clone(),
next_worker_id(),
))
});
App::new()
.wrap(TracingLogger::default())

72
src/process.rs
View file

@ -43,14 +43,17 @@ pin_project_lite::pin_project! {
impl Process {
#[tracing::instrument]
pub(crate) fn run(command: &str, args: &[&str]) -> std::io::Result<Self> {
Self::spawn(Command::new(command).args(args))
tracing::trace_span!(parent: None, "Create command")
.in_scope(|| Self::spawn(Command::new(command).args(args)))
}
#[tracing::instrument]
pub(crate) fn spawn(cmd: &mut Command) -> std::io::Result<Self> {
let cmd = cmd.stdin(Stdio::piped()).stdout(Stdio::piped());
tracing::trace_span!(parent: None, "Spawn command").in_scope(|| {
let cmd = cmd.stdin(Stdio::piped()).stdout(Stdio::piped());
cmd.spawn().map(|child| Process { child })
cmd.spawn().map(|child| Process { child })
})
}
#[tracing::instrument]
@ -67,10 +70,11 @@ impl Process {
let mut stdin = self.child.stdin.take().expect("stdin exists");
let stdout = self.child.stdout.take().expect("stdout exists");
let (tx, rx) = channel::<std::io::Error>();
let (tx, rx) = tracing::trace_span!(parent: None, "Create channel")
.in_scope(channel::<std::io::Error>);
let mut child = self.child;
let handle = tracing::trace_span!(parent: None, "Spawn").in_scope(|| {
let handle = tracing::trace_span!(parent: None, "Spawn task").in_scope(|| {
actix_rt::spawn(async move {
if let Err(e) = stdin.write_all_buf(&mut input).await {
let _ = tx.send(e);
@ -104,21 +108,23 @@ impl Process {
pub(crate) fn read(mut self) -> impl AsyncRead + Unpin {
let stdout = self.child.stdout.take().expect("stdout exists");
let (tx, rx) = channel();
let (tx, rx) = tracing::trace_span!(parent: None, "Create channel").in_scope(channel);
let mut child = self.child;
let handle = actix_rt::spawn(async move {
match child.wait().await {
Ok(status) => {
if !status.success() {
let _ =
tx.send(std::io::Error::new(std::io::ErrorKind::Other, &StatusError));
let handle = tracing::trace_span!(parent: None, "Spawn task").in_scope(|| {
actix_rt::spawn(async move {
match child.wait().await {
Ok(status) => {
if !status.success() {
let _ = tx
.send(std::io::Error::new(std::io::ErrorKind::Other, &StatusError));
}
}
Err(e) => {
let _ = tx.send(e);
}
}
Err(e) => {
let _ = tx.send(e);
}
}
})
});
ProcessRead {
@ -138,27 +144,29 @@ impl Process {
let mut stdin = self.child.stdin.take().expect("stdin exists");
let stdout = self.child.stdout.take().expect("stdout exists");
let (tx, rx) = channel();
let (tx, rx) = tracing::trace_span!(parent: None, "Create channel").in_scope(channel);
let mut child = self.child;
let handle = actix_rt::spawn(async move {
if let Err(e) = store.read_into(&identifier, &mut stdin).await {
let _ = tx.send(e);
return;
}
drop(stdin);
let handle = tracing::trace_span!(parent: None, "Spawn task").in_scope(|| {
actix_rt::spawn(async move {
if let Err(e) = store.read_into(&identifier, &mut stdin).await {
let _ = tx.send(e);
return;
}
drop(stdin);
match child.wait().await {
Ok(status) => {
if !status.success() {
let _ =
tx.send(std::io::Error::new(std::io::ErrorKind::Other, &StatusError));
match child.wait().await {
Ok(status) => {
if !status.success() {
let _ = tx
.send(std::io::Error::new(std::io::ErrorKind::Other, &StatusError));
}
}
Err(e) => {
let _ = tx.send(e);
}
}
Err(e) => {
let _ = tx.send(e);
}
}
})
});
ProcessRead {

3
src/stream.rs
View file

@ -187,7 +187,8 @@ where
match std::mem::replace(&mut this.state, IterStreamState::Pending) {
IterStreamState::New { iterator, buffer } => {
let (sender, receiver) = tokio::sync::mpsc::channel(buffer);
let (sender, receiver) = tracing::trace_span!(parent: None, "Create channel")
.in_scope(|| tokio::sync::mpsc::channel(buffer));
let mut handle = actix_rt::task::spawn_blocking(move || {
let iterator = iterator.into_iter();

3
src/tmp_file.rs
View file

@ -13,7 +13,8 @@ struct TmpFile(PathBuf);
impl Drop for TmpFile {
fn drop(&mut self) {
actix_rt::spawn(tokio::fs::remove_file(self.0.clone()));
tracing::trace_span!(parent: None, "Spawn task")
.in_scope(|| actix_rt::spawn(tokio::fs::remove_file(self.0.clone())));
}
}