Repository: qdrant/qdrant
File: lib/collection/src/update_handler.rs (Rust)

Code before:
                _ = &mut stop_receiver => true,
            };

            if stop {
                log::debug!("Stopping flush worker for shard {}", shard_path.display());
                return;
            }

            let segments_clone = segments.clone();
            let wal_clone = wal.clone();
            let wal_keep_from_clone = wal_keep_from.clone();
            let clocks_clone = clocks.clone();
            let shard_path_clone = shard_path.clone();

            tokio::task::spawn_blocking(move || {
                Self::flush_worker_internal(
                    segments_clone,
                    wal_clone,
                    wal_keep_from_clone,
                    clocks_clone,
                    shard_path_clone,
                )
            });
        }
    }

    /// Returns confirmed version after flush of all segments
    ///
    /// # Errors
    /// Returns an error on flush failure
    fn flush_segments(segments: LockedSegmentHolder) -> OperationResult<SeqNumberType> {
        let read_segments = segments.read();
        let flushed_version = read_segments.flush_all(false, false)?;
        Ok(match read_segments.failed_operation.iter().cloned().min() {
            None => flushed_version,
            Some(failed_operation) => min(failed_operation, flushed_version),
        })
    }
}

/// Trigger optimizers when CPU budget is available
fn trigger_optimizers_on_resource_budget(
    optimizer_resource_budget: ResourceBudget,
    desired_cpus: usize,
    desired_io: usize,
    sender: Sender<OptimizerSignal>,
) -> JoinHandle<()> {
    task::spawn(async move {
        log::trace!("Skipping optimization checks, waiting for CPU budget to be available");
        optimizer_resource_budget
            .notify_on_budget_available(desired_cpus, desired_io)

Review comment (suggestion):
this is the main change, not time is async, but we spawn thread on wach timeout. (actually, thread keep-alive is higher than that, only one thread is spawned per multiple collections)

Code after:
                    log::debug!("Stopping flush worker for shard {}", shard_path.display());
                    return;
                },
                // Flush at the configured flush interval
                _ = tokio::time::sleep(Duration::from_secs(flush_interval_sec)) => {},
            };

            let segments_clone = segments.clone();
            let wal_clone = wal.clone();
            let wal_keep_from_clone = wal_keep_from.clone();
            let clocks_clone = clocks.clone();
            let shard_path_clone = shard_path.clone();

            tokio::task::spawn_blocking(move || {
                Self::flush_worker_internal(
                    segments_clone,
                    wal_clone,
                    wal_keep_from_clone,
                    clocks_clone,
                    shard_path_clone,
                )
            })
            .await
            .unwrap_or_else(|error| {
                log::error!("Flush worker failed: {error}",);
            });
        }
    }

    /// Returns confirmed version after flush of all segments
    ///
    /// # Errors
    /// Returns an error on flush failure
    fn flush_segments(segments: LockedSegmentHolder) -> OperationResult<SeqNumberType> {
        let read_segments = segments.read();
        let flushed_version = read_segments.flush_all(false, false)?;
        Ok(match read_segments.failed_operation.iter().cloned().min() {
            None => flushed_version,
            Some(failed_operation) => min(failed_operation, flushed_version),
        })
    }
}

/// Trigger optimizers when CPU budget is available
fn trigger_optimizers_on_resource_budget(
    optimizer_resource_budget: ResourceBudget,
    desired_cpus: usize,
    desired_io: usize,
    sender: Sender<OptimizerSignal>,
) -> JoinHandle<()> {
    task::spawn(async move {