diff --git a/crates/service/job/inference_job/src/job/job_types/workflow/video_style_transfer/util/preview_processor/bucket_uploader.rs b/crates/service/job/inference_job/src/job/job_types/workflow/video_style_transfer/util/preview_processor/bucket_uploader.rs index 97dc04575a..a14b19a48b 100644 --- a/crates/service/job/inference_job/src/job/job_types/workflow/video_style_transfer/util/preview_processor/bucket_uploader.rs +++ b/crates/service/job/inference_job/src/job/job_types/workflow/video_style_transfer/util/preview_processor/bucket_uploader.rs @@ -90,7 +90,7 @@ impl BucketUploadActor { } BucketUploadMessage::UploadMultipleFrames { requests, bucket_client, result } => { - let max_concurrent_uploads = 50; + let max_concurrent_uploads = 20; let mut join_set:JoinSet<(PathBuf,BucketUploadResult)> = JoinSet::new(); let mut results = vec![]; for request in requests { @@ -130,7 +130,7 @@ pub struct BucketUploadActorHandle { impl BucketUploadActorHandle { pub fn new() -> Self { - let (tx, rx) = mpsc::channel(500); + let (tx, rx) = mpsc::channel(20); let actor = BucketUploadActor::new(rx); tokio::spawn(async move { run_bucket_upload_actor(actor).await; diff --git a/crates/service/job/inference_job/src/job/job_types/workflow/video_style_transfer/util/process_preview_updates.rs b/crates/service/job/inference_job/src/job/job_types/workflow/video_style_transfer/util/process_preview_updates.rs index 12e08ea1e9..edb3aac2a7 100644 --- a/crates/service/job/inference_job/src/job/job_types/workflow/video_style_transfer/util/process_preview_updates.rs +++ b/crates/service/job/inference_job/src/job/job_types/workflow/video_style_transfer/util/process_preview_updates.rs @@ -24,7 +24,7 @@ use tokens::tokens::generic_inference_jobs::InferenceJobToken; use tokio::sync::oneshot::Sender; use tokio::sync::{mpsc, oneshot}; use tokio::task::JoinSet; - +use tokio::time::MissedTickBehavior; static ALLOWED_TYPES_FRAMES : Lazy> = Lazy::new(|| { HashSet::from([ "jpg", @@ -39,6 +39,17 @@ enum PreviewStage { SecondPass, FinalPass, } + +impl PreviewStage { + fn iter() -> Vec { + vec![ + PreviewStage::FirstPass, + PreviewStage::SecondPass, + PreviewStage::FinalPass, + ] + } +} + impl PreviewStage { fn expected_directory_name(&self) -> &'static str { match self { @@ -379,8 +390,8 @@ impl StageDirectoryState { async fn upload_multiple_frames_from_disk(&self, bucket_client: &BucketClient, frames: Vec<(u32, PathBuf)>) -> (Vec) { - let mut bucket_upload_requests = Vec::new(); - let mut updates = Vec::new(); + let mut bucket_upload_requests = Vec::with_capacity(frames.len()); + let mut updates = Vec::with_capacity(frames.len()); for (frame_number, disk_path) in frames.iter() { let local_file_extension = match disk_path.extension().and_then(|s| s.to_str()) { @@ -630,8 +641,9 @@ impl PreviewProcessor { pub async fn start_processing(&mut self, bucket_client: &BucketClient) -> () { let (frames_tx, mut frames_rx) = mpsc::channel(100); - let walker = DirWalker::new_with_batch_size(self.base_directory.clone(), frames_tx.clone(), 50); + let walker = DirWalker::new_with_batch_size(self.base_directory.clone(), frames_tx.clone(), 30); let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(5)); + interval.set_missed_tick_behavior(MissedTickBehavior::Skip); loop { let walker = walker.clone(); log::debug!("Processing preview frames loop ticked at {:?}", Utc::now()); @@ -755,18 +767,4 @@ impl PreviewProcessor { let result = stage_state.upload_frame_from_disk_with_retry(bucket_client, frame_number.clone(), disk_path).await; result.state } - - - async fn process_frame_from_disk(&self, stage: PreviewStage, frame_number: u32, bucket_client: &BucketClient, disk_path: PathBuf) -> () { - let result = self.upload_frame_from_disk(stage, frame_number, bucket_client, disk_path).await; - - if result == PreviewFrameUploadResult::UploadComplete { - self.stages.with_mut(&stage, |stage_state| { - stage_state.mark_frame_uploaded(frame_number); - if stage_state.all_frames_uploaded() { - stage_state.state = PreviewStageState::UploadComplete; - } - }); - } - } } diff --git a/service_gpu.Dockerfile b/service_gpu.Dockerfile index 054e042059..25a3cdfb04 100644 --- a/service_gpu.Dockerfile +++ b/service_gpu.Dockerfile @@ -19,7 +19,7 @@ RUN apt-get update \ curl \ fontconfig \ libfontconfig1-dev \ - pkg-config + pkg-config musl musl-dev musl-tools # NB: Fix for fontconfig (servo-fontconfig-sys): https://github.com/alacritty/alacritty/issues/4423#issuecomment-727277235 # TODO(bt, 2023-02-23): This has not been verified to work yet. @@ -31,8 +31,7 @@ RUN curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs \ | sh -s -- --default-toolchain $RUST_TOOLCHAIN -y # Install correct Rust version -#RUN $HOME/.cargo/bin/rustup install $RUST_VERSION -#RUN $HOME/.cargo/bin/rustup default $RUST_VERSION +#RUN $HOME/.cargo/bin/rustup default stable-x86_64-unknown-linux-gnu # Report Rust version for build debugging RUN $HOME/.cargo/bin/rustup show @@ -115,6 +114,15 @@ RUN RUSTFLAGS="-C target-feature=+crt-static" SQLX_OFFLINE=true \ --release --target=x86_64-unknown-linux-gnu \ --bin inference-job +RUN ls -lR $HOME/.cargo/bin +RUN $HOME/.cargo/bin/rustup target add x86_64-unknown-linux-musl + + +RUN RUSTFLAGS="-C target-feature=+crt-static" SQLX_OFFLINE=true \ + LD_LIBRARY_PATH=/usr/lib:${LD_LIBRARY_PATH} \ + $HOME/.cargo/bin/cargo build \ + --release --target=x86_64-unknown-linux-musl \ + --bin inference-job # Print a report on disk space RUN echo "Disk usage at current directory (after all builds):" RUN pwd @@ -155,6 +163,7 @@ RUN echo -n ${GIT_SHA} > GIT_SHA COPY --from=builder /tmp/target/x86_64-unknown-linux-gnu/release/dummy-service / COPY --from=builder /tmp/target/x86_64-unknown-linux-gnu/release/download-job / COPY --from=builder /tmp/target/x86_64-unknown-linux-gnu/release/inference-job / +COPY --from=builder /tmp/target/x86_64-unknown-linux-musl/release/inference-job /inference-job-musl # Container includes COPY includes/ /includes