timerwheel 0.1.0

// Copyright © 2026-present The Timerwheel Authors. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use std::panic::{self, AssertUnwindSafe};
use std::sync::{Arc, Mutex, mpsc};

use crate::executor::metrics::ExecutorMetricsInner;
use crate::executor::panic_handler::PanicHandler;
use crate::executor::task::BoxTask;

pub(crate) enum Message {
    Run(BoxTask),
    Stop,
}

pub(crate) fn run(
    receiver: Arc<Mutex<mpsc::Receiver<Message>>>,
    metrics: Arc<ExecutorMetricsInner>,
    panic_handler: Arc<dyn PanicHandler>,
) {
    loop {
        let message = match receiver.lock() {
            Ok(receiver) => receiver.recv(),
            Err(_) => return,
        };

        match message {
            Ok(Message::Run(task)) => {
                // Panic recovery is a worker boundary concern. A panicking task
                // updates metrics but does not terminate the worker loop.
                let result = panic::catch_unwind(AssertUnwindSafe(|| {
                    task.run();
                }));
                if let Err(payload) = result {
                    panic_handler.handle_panic(payload.as_ref());
                    metrics.panicked();
                }
                metrics.completed();
            }
            Ok(Message::Stop) | Err(_) => return,
        }
    }
}