Repository: bevyengine/bevy
File: crates/bevy_asset/src/processor/mod.rs (Rust)

Code before:
            IoTaskPool::get()
                .spawn(async move {
                    while let Ok(event) = receiver.recv().await {
                        let Ok(source) = processor.get_source(source_id.clone()) else {
                            return;
                        };
                        processor
                            .handle_asset_source_event(source, event, &sender)
                            .await;
                    }
                })
                .detach();
        }
    }

    /// Executes all tasks that come through `receiver`, and updates the processor's overall state
    /// based on task starts and ends.
    async fn execute_processing_tasks(
        &self,
        sender: &async_channel::Sender<ProcessorTaskEvent>,
        receiver: &async_channel::Receiver<ProcessorTaskEvent>,
    ) {
        // If there aren't any tasks in the channel the first time around, we should immediately go
        // to the finished state (otherwise we'd be sitting around stuck in the `Initialized`
        // state).
        if receiver.is_empty() {
            self.set_state(ProcessorState::Finished).await;
        }
        // Note: only this method sends `ProcessorTaskEvent::Finished` (to itself). We could have
        // two channels, one for finished messages, and one for starting tasks, but this requires
        // selecting across them, which the macro is feature-gated behind `std`.
        // TODO: This task currently holds both the sender and receiver, meaning the channel never
        // "naturally" closes, so this task never terminates. It would be nice if this task
        // naturally terminated when the asset source event receivers close.
        let mut pending_tasks = 0;
        while let Ok(event) = receiver.recv().await {
            match event {
                ProcessorTaskEvent::Start(source_id, path) => {
                    let processor = self.clone();
                    let sender = sender.clone();
                    pending_tasks += 1;
                    IoTaskPool::get()
                        .spawn(async move {
                            let Ok(source) = processor.get_source(source_id) else {
                                return;
                            };
                            processor.process_asset(source, path, sender.clone()).await;
                            // If the channel gets closed, that's ok. Just ignore it.
                            let _ = sender.send(ProcessorTaskEvent::Finished).await;
                        })
                        .detach();

Review comment (suggestion):
I spent quite a bit of time trying to come up with suggestions to improve this, but the thing I missed is that it stays alive after the initial processing not due to the TODO, but because it needs to stay around to handle `AssetSourceEvent`s. That should be documented on the function, otherwise it's very natural to wonder why it doesn't exit when the receiver is empty the first time.

Code after:
            IoTaskPool::get()
                .spawn(async move {
                    while let Ok(event) = receiver.recv().await {
                        let Ok(source) = processor.get_source(source_id.clone()) else {
                            return;
                        };
                        processor
                            .handle_asset_source_event(source, event, &sender)
                            .await;
                    }
                })
                .detach();
        }
    }

    /// Executes all tasks that come through `receiver`, and updates the processor's overall state
    /// based on task starts and ends.
    ///
    /// This future does not terminate until the channel is closed (not when the channel is empty).
    /// This means that in [`AssetProcessor::start`], this execution will continue even after all
    /// the initial tasks are processed.
    async fn execute_processing_tasks(
        &self,
        new_task_sender: async_channel::Sender<(AssetSourceId<'static>, PathBuf)>,
        new_task_receiver: async_channel::Receiver<(AssetSourceId<'static>, PathBuf)>,
    ) {
        // Convert the Sender into a WeakSender so that once all task producers terminate (and drop
        // their sender), this task doesn't keep itself alive. We still however need a way to get
        // the sender since processing tasks can start the tasks of dependent assets.
        let new_task_sender = {
            let weak_sender = new_task_sender.downgrade();
            drop(new_task_sender);
            weak_sender
        };

        // If there aren't any tasks in the channel the first time around, we should immediately go
        // to the finished state (otherwise we'd be sitting around stuck in the `Initialized`
        // state).
        if new_task_receiver.is_empty() {
            self.set_state(ProcessorState::Finished).await;
        }
        enum ProcessorTaskEvent {
            Start(AssetSourceId<'static>, PathBuf),
            Finished,
        }
        let (task_finished_sender, task_finished_receiver) = async_channel::unbounded::<()>();

        let mut pending_tasks = 0;
        while let Ok(event) = {
            // It's ok to use `select_biased` since we prefer to start task rather than finish tasks
            // anyway - since otherwise we might mark the processor as finished before all queued