mirror of
https://git.asonix.dog/asonix/pict-rs
synced 2024-12-22 03:11:24 +00:00
Timeout all operations while holding semaphore
This ensures that the semaphore is always guaranteed to be released, even if a processing operation stalls
This commit is contained in:
parent
8a551fa65d
commit
ac9a3773d1
2 changed files with 79 additions and 30 deletions
|
@ -2,6 +2,7 @@ mod ffmpeg;
|
||||||
mod magick;
|
mod magick;
|
||||||
|
|
||||||
use crate::{
|
use crate::{
|
||||||
|
bytes_stream::BytesStream,
|
||||||
details::Details,
|
details::Details,
|
||||||
error::{Error, UploadError},
|
error::{Error, UploadError},
|
||||||
formats::{ImageFormat, InputProcessableFormat, InternalVideoFormat, ProcessableFormat},
|
formats::{ImageFormat, InputProcessableFormat, InternalVideoFormat, ProcessableFormat},
|
||||||
|
@ -196,33 +197,16 @@ async fn process<S: Store + 'static>(
|
||||||
let guard = MetricsGuard::guard();
|
let guard = MetricsGuard::guard();
|
||||||
let permit = crate::process_semaphore().acquire().await?;
|
let permit = crate::process_semaphore().acquire().await?;
|
||||||
|
|
||||||
let identifier = input_identifier(state, output_format, hash.clone(), original_details).await?;
|
let bytes = do_process(
|
||||||
|
state,
|
||||||
let input_details = crate::ensure_details_identifier(state, &identifier).await?;
|
output_format,
|
||||||
|
variant_args,
|
||||||
let input_format = input_details
|
hash.clone(),
|
||||||
.internal_format()
|
original_details,
|
||||||
.processable_format()
|
)
|
||||||
.expect("Already verified format is processable");
|
.with_timeout(Duration::from_secs(state.config.media.process_timeout * 4))
|
||||||
|
.await
|
||||||
let format = input_format.process_to(output_format);
|
.map_err(|_| UploadError::ProcessTimeout)??;
|
||||||
|
|
||||||
let quality = match format {
|
|
||||||
ProcessableFormat::Image(format) => state.config.media.image.quality_for(format),
|
|
||||||
ProcessableFormat::Animation(format) => state.config.media.animation.quality_for(format),
|
|
||||||
};
|
|
||||||
|
|
||||||
let stream = state.store.to_stream(&identifier, None, None).await?;
|
|
||||||
|
|
||||||
let bytes =
|
|
||||||
crate::magick::process_image_command(state, variant_args, input_format, format, quality)
|
|
||||||
.await?
|
|
||||||
.drive_with_stream(stream)
|
|
||||||
.into_bytes_stream()
|
|
||||||
.instrument(tracing::info_span!(
|
|
||||||
"Reading processed image to BytesStream"
|
|
||||||
))
|
|
||||||
.await?;
|
|
||||||
|
|
||||||
drop(permit);
|
drop(permit);
|
||||||
|
|
||||||
|
@ -258,6 +242,47 @@ async fn process<S: Store + 'static>(
|
||||||
Ok((details, identifier)) as Result<(Details, Arc<str>), Error>
|
Ok((details, identifier)) as Result<(Details, Arc<str>), Error>
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn do_process<S>(
|
||||||
|
state: &State<S>,
|
||||||
|
output_format: InputProcessableFormat,
|
||||||
|
variant_args: Vec<String>,
|
||||||
|
hash: Hash,
|
||||||
|
original_details: &Details,
|
||||||
|
) -> Result<BytesStream, Error>
|
||||||
|
where
|
||||||
|
S: Store + 'static,
|
||||||
|
{
|
||||||
|
let identifier = input_identifier(state, output_format, hash.clone(), original_details).await?;
|
||||||
|
|
||||||
|
let input_details = crate::ensure_details_identifier(state, &identifier).await?;
|
||||||
|
|
||||||
|
let input_format = input_details
|
||||||
|
.internal_format()
|
||||||
|
.processable_format()
|
||||||
|
.expect("Already verified format is processable");
|
||||||
|
|
||||||
|
let format = input_format.process_to(output_format);
|
||||||
|
|
||||||
|
let quality = match format {
|
||||||
|
ProcessableFormat::Image(format) => state.config.media.image.quality_for(format),
|
||||||
|
ProcessableFormat::Animation(format) => state.config.media.animation.quality_for(format),
|
||||||
|
};
|
||||||
|
|
||||||
|
let stream = state.store.to_stream(&identifier, None, None).await?;
|
||||||
|
|
||||||
|
let bytes =
|
||||||
|
crate::magick::process_image_command(state, variant_args, input_format, format, quality)
|
||||||
|
.await?
|
||||||
|
.drive_with_stream(stream)
|
||||||
|
.into_bytes_stream()
|
||||||
|
.instrument(tracing::info_span!(
|
||||||
|
"Reading processed image to BytesStream"
|
||||||
|
))
|
||||||
|
.await?;
|
||||||
|
|
||||||
|
Ok(bytes)
|
||||||
|
}
|
||||||
|
|
||||||
pub(crate) async fn ensure_motion_identifier<S>(
|
pub(crate) async fn ensure_motion_identifier<S>(
|
||||||
state: &State<S>,
|
state: &State<S>,
|
||||||
hash: Hash,
|
hash: Hash,
|
||||||
|
|
|
@ -8,7 +8,7 @@ use crate::{
|
||||||
details::Details,
|
details::Details,
|
||||||
error::{Error, UploadError},
|
error::{Error, UploadError},
|
||||||
formats::InternalFormat,
|
formats::InternalFormat,
|
||||||
future::{WithMetrics, WithPollTimer},
|
future::{WithMetrics, WithPollTimer, WithTimeout},
|
||||||
repo::{Alias, ArcRepo, DeleteToken, Hash},
|
repo::{Alias, ArcRepo, DeleteToken, Hash},
|
||||||
state::State,
|
state::State,
|
||||||
store::Store,
|
store::Store,
|
||||||
|
@ -58,6 +58,32 @@ where
|
||||||
|
|
||||||
let permit = crate::process_semaphore().acquire().await?;
|
let permit = crate::process_semaphore().acquire().await?;
|
||||||
|
|
||||||
|
let res = process_ingest_bytestream(state, bytes, upload_query)
|
||||||
|
.with_timeout(Duration::from_secs(state.config.media.process_timeout * 4))
|
||||||
|
.await
|
||||||
|
.map_err(|_| UploadError::ProcessTimeout)?;
|
||||||
|
|
||||||
|
drop(permit);
|
||||||
|
|
||||||
|
res
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn process_ingest_bytestream<S>(
|
||||||
|
state: &State<S>,
|
||||||
|
bytes: BytesStream,
|
||||||
|
upload_query: &UploadQuery,
|
||||||
|
) -> Result<
|
||||||
|
(
|
||||||
|
InternalFormat,
|
||||||
|
Arc<str>,
|
||||||
|
Details,
|
||||||
|
Rc<RefCell<hasher::State>>,
|
||||||
|
),
|
||||||
|
Error,
|
||||||
|
>
|
||||||
|
where
|
||||||
|
S: Store,
|
||||||
|
{
|
||||||
tracing::trace!("Validating bytes");
|
tracing::trace!("Validating bytes");
|
||||||
let (input_type, process_read) =
|
let (input_type, process_read) =
|
||||||
validate::validate_bytes_stream(state, bytes, &upload_query.limits)
|
validate::validate_bytes_stream(state, bytes, &upload_query.limits)
|
||||||
|
@ -120,8 +146,6 @@ where
|
||||||
.with_poll_timer("details-from-bytes-stream")
|
.with_poll_timer("details-from-bytes-stream")
|
||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
drop(permit);
|
|
||||||
|
|
||||||
Ok((input_type, identifier, details, hash_state))
|
Ok((input_type, identifier, details, hash_state))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
Loading…
Reference in a new issue