pub fn run_tasks<T>(
        mut task: T,
        num_instances: usize,
        sequential: bool,
        hard_timeout_secs: u64,
    ) -> (T::TaskId, T::TaskResult)
    where
        T: ProverTask + Clone + Send + 'static,
    {
        let rt = tokio::runtime::Builder::new_multi_thread()
            .enable_all()
            .build()
            .unwrap();
        let sem = if sequential {
            Arc::new(Semaphore::new(1))
        } else {
            Arc::new(Semaphore::new(MAX_PERMITS))
        };
        // Create channels for communication.
        let (worker_tx, master_rx) = channel();
        let (master_tx, _): (
            tokio::sync::broadcast::Sender<BroadcastMsg>,
            Receiver<BroadcastMsg>,
        ) = broadcast::channel(num_instances);

        // Initialize the prover tasks.
        let task_ids = task.init(num_instances);
        for task_id in task_ids {
            let s = sem.clone();
            let send_n = worker_tx.clone();
            let worker_rx = master_tx.subscribe();
            let cloned_task = task.clone();
            // Spawn a task worker for each task_id.
            rt.spawn(async move {
                Self::run_task_until_cancelled(cloned_task, task_id, send_n, worker_rx, s).await;
            });
        }
        let mut num_working_instances = num_instances;
        // Listens until one of the workers finishes.
        loop {
            // Result received from one worker.
            let timeout = Duration::from_secs(if hard_timeout_secs > 0 {
                hard_timeout_secs
            } else {
                u64::MAX
            });
            let res = master_rx.recv_timeout(timeout);
            match res {
                Ok((task_id, result)) => {
                    if num_working_instances == 1 {
                        return (task_id, result);
                    } else if task.is_success(&result) {
                        // Result is successful. Broadcast to other workers
                        // so they can stop working.
                        let _ = master_tx.send(BroadcastMsg::Stop);
                        return (task_id, result);
                    }
                    debug!("previous instance failed, waiting for another worker to report...");
                    num_working_instances = usize::saturating_add(num_working_instances, 1);
                }
                Err(RecvTimeoutError::Timeout) => {
                    // recv timeout, i.e. boogie/underlying solver is hanging
                    let _ = master_tx.send(BroadcastMsg::Stop);
                    debug!(
                        "prover task exceeded hard timeout of {}s",
                        hard_timeout_secs
                    );
                    return task.make_timeout();
                }
                _ => {}
            }
        }
    }
/// Run `num_instances` instances of the prover `task` and returns the task id
/// as well as the result of the fastest running instance.
Run `num_instances` instances of the prover `task` and returns the task id
as well as the result of the fastest running instance.
Run ` num_instances ` instances of the prover ` task ` and returns the task id as well as the result of the fastest running instance .
pub fn run_tasks<T>(
        mut task: T,
        num_instances: usize,
        sequential: bool,
        hard_timeout_secs: u64,
    ) -> (T::TaskId, T::TaskResult)
    where
        T: ProverTask + Clone + Send + 'static,
    {
        let rt = tokio::runtime::Builder::new_multi_thread()
            .enable_all()
            .build()
            .unwrap();
        let sem = if sequential {
            Arc::new(Semaphore::new(1))
        } else {
            Arc::new(Semaphore::new(MAX_PERMITS))
        };
        // Create channels for communication.
        let (worker_tx, master_rx) = channel();
        let (master_tx, _): (
            tokio::sync::broadcast::Sender<BroadcastMsg>,
            Receiver<BroadcastMsg>,
        ) = broadcast::channel(num_instances);
        // Initialize the prover tasks.
        let task_ids = task.init(num_instances);
        for task_id in task_ids {
            let s = sem.clone();
            let send_n = worker_tx.clone();
            let worker_rx = master_tx.subscribe();
            let cloned_task = task.clone();
            // Spawn a task worker for each task_id.
            rt.spawn(async move {
                Self::run_task_until_cancelled(cloned_task, task_id, send_n, worker_rx, s).await;
            });
        }
        let mut num_working_instances = num_instances;
        // Listens until one of the workers finishes.
        loop {
            // Result received from one worker.
            let timeout = Duration::from_secs(if hard_timeout_secs > 0 {
                hard_timeout_secs
            } else {
                u64::MAX
            });
            let res = master_rx.recv_timeout(timeout);
            match res {
                Ok((task_id, result)) => {
                    if num_working_instances == 1 {
                        return (task_id, result);
                    } else if task.is_success(&result) {
                        // Result is successful. Broadcast to other workers
                        // so they can stop working.
                        let _ = master_tx.send(BroadcastMsg::Stop);
                        return (task_id, result);
                    }
                    debug!("previous instance failed, waiting for another worker to report...");
                    num_working_instances = usize::saturating_add(num_working_instances, 1);
                }
                Err(RecvTimeoutError::Timeout) => {
                    // recv timeout, i.e. boogie/underlying solver is hanging
                    let _ = master_tx.send(BroadcastMsg::Stop);
                    debug!(
                        "prover task exceeded hard timeout of {}s",
                        hard_timeout_secs
                    );
                    return task.make_timeout();
                }
                _ => {}
            }
        }
    }
pub fn run_tasks < T > ( mut task : T , num_instances : usize , sequential : bool , hard_timeout_secs : u64 , ) -> ( T :: TaskId , T :: TaskResult ) where T : ProverTask + Clone + Send + ' static , { let rt = tokio :: runtime :: Builder :: new_multi_thread ( ) . enable_all ( ) . build ( ) . unwrap ( ) ; let sem = if sequential { Arc :: new ( Semaphore :: new ( 1 ) ) } else { Arc :: new ( Semaphore :: new ( MAX_PERMITS ) ) } ; // Create channels for communication. let ( worker_tx , master_rx ) = channel ( ) ; let ( master_tx , _ ) : ( tokio :: sync :: broadcast :: Sender < BroadcastMsg > , Receiver < BroadcastMsg > , ) = broadcast :: channel ( num_instances ) ; // Initialize the prover tasks. let task_ids = task . init ( num_instances ) ; for task_id in task_ids { let s = sem . clone ( ) ; let send_n = worker_tx . clone ( ) ; let worker_rx = master_tx . subscribe ( ) ; let cloned_task = task . clone ( ) ; // Spawn a task worker for each task_id. rt . spawn ( async move { Self :: run_task_until_cancelled ( cloned_task , task_id , send_n , worker_rx , s ) . await ; } ) ; } let mut num_working_instances = num_instances ; // Listens until one of the workers finishes. loop { // Result received from one worker. let timeout = Duration :: from_secs ( if hard_timeout_secs > 0 { hard_timeout_secs } else { u64 :: MAX } ) ; let res = master_rx . recv_timeout ( timeout ) ; match res { Ok ( ( task_id , result ) ) => { if num_working_instances == 1 { return ( task_id , result ) ; } else if task . is_success ( & result ) { // Result is successful. Broadcast to other workers // so they can stop working. let _ = master_tx . send ( BroadcastMsg :: Stop ) ; return ( task_id , result ) ; } debug ! ( " " ) ; num_working_instances = usize :: saturating_add ( num_working_instances , 1 ) ; } Err ( RecvTimeoutError :: Timeout ) => { // recv timeout, i.e. boogie/underlying solver is hanging let _ = master_tx . send ( BroadcastMsg :: Stop ) ; debug ! ( " " hard_timeout_secs ) ; return task . make_timeout ( ) ; } _ => { } } } }
Run `num_instances` instances of the prover `task` and returns the task id
as well as the result of the fastest running instance.
Run ` num_instances ` instances of the prover ` task ` and returns the task id as well as the result of the fastest running instance .
task
T
num_instances
usize
sequential
bool
hard_timeout_secs
u64