fn spawn(mut self) -> Option<std::thread::JoinHandle<()>> {
        let thread = std::thread::spawn(move || {
            self.setup();
            let mut stats = ExecutorStats {
                id: self.id,
                ..Default::default()
            };
            let mut stats_event = SimpleEventTimer::default();
            let time_slice = get_time_slice();
            let backoff = LinearBackoff::new();
            log::debug!("executor {} is alive", self.id);
            let mut last_try_recv = Instant::now();
            'terminate: loop {
                // report to the system monitor, maybe look at fixing drift
                if stats_event.check() && self.monitor.send(MonitorMessage::ExecutorStats(stats)).is_err() {
                    log::debug!("failed to send exec stats to monitor");
                }
                if last_try_recv.elapsed() >= Duration::from_secs(2) {
                    last_try_recv += Duration::from_secs(2);
                    // processs any received commands, break out of loop if told to terminate
                    if self.try_recv(&stats) {
                        break 'terminate;
                    }
                }

                // move blocked senders along...
                if self.blocked_sender_count != 0 {
                    self.blocked_sender_count = self.try_completing_send(&mut stats);
                }
                // try to run a task
                let ran_task = if self.get_state() == ExecutorState::Running {
                    self.run_task(time_slice, &mut stats)
                } else {
                    false
                };

                // if no longer running and we don't have any blocked senders, we can leave
                if self.get_state() == ExecutorState::Parked {
                    tls_executor_data.with(|t| {
                        let tls = t.borrow();
                        self.blocked_sender_count = tls.blocked_senders.len();
                    });
                    if self.blocked_sender_count == 0 {
                        break;
                    }
                }
                // and after all that, we've got nothing left to do. Let's catch some zzzz's
                if self.blocked_sender_count == 0 && !ran_task {
                    if backoff.is_completed() {
                        log::trace!("executor {} is sleeping", self.id);
                        let start = std::time::Instant::now();
                        let park_duration = stats_event.remaining();
                        // sanity bailout
                        last_try_recv = start;
                        if self.try_recv(&stats) {
                            break 'terminate;
                        }
                        if RUN_QUEUE_LEN.load(Ordering::SeqCst) == 0 {
                            EXECUTORS_SNOOZING.fetch_add(1, Ordering::SeqCst);
                            thread::park_timeout(park_duration);
                            EXECUTORS_SNOOZING.fetch_sub(1, Ordering::SeqCst);
                            stats.sleep_count += 1;
                            stats.sleep_time += start.elapsed();
                        }
                        log::trace!("executor {} is awake", self.id);
                    } else {
                        backoff.snooze();
                    }
                } else if backoff.reset() {
                    stats.disturbed_nap += 1;
                }
            }
            log::debug!("executor {} is dead", self.id);
            log::debug!("{:#?}", stats);
            let remaining_tasks = self.work.len();
            if remaining_tasks > 0 {
                log::debug!(
                    "exec {} exiting with {} tasks in the worker q, will re-inject",
                    self.id,
                    remaining_tasks
                );
                // re-inject any remaining tasks
                while let Some(task) = self.work.pop() {
                    self.run_queue.push(task);
                }
            }
            tls_executor_data.with(|t| {
                let tls = t.borrow_mut();
                if !tls.blocked_senders.is_empty() {
                    log::error!(
                        "executor {} exited, but continues to have {} blocked senders",
                        self.id,
                        tls.blocked_senders.len()
                    );
                }
            });
            // as a last and final, tell the monitor that we're dead...
            if self.monitor.send(MonitorMessage::Terminated(self.id)).is_err() {
                log::warn!("executor {} exiting without informing system monitor", self.id);
            }
        });
        Some(thread)
    }
// This is the executor thread. It has a few responsibilites. It listens for,
// and acts upon events (via channel). It complete senders that are blocked.
// It gets tasks and runs them. It might be better to have a select
// for blocked senders, or aggregate them and have a separate thread responsible
// for them. In a well tuned system, we shouldn't see blocking of senders,
// which represents a failure in dataflow management.
This is the executor thread. It has a few responsibilites. It listens for,
and acts upon events (via channel). It complete senders that are blocked.
It gets tasks and runs them. It might be better to have a select
for blocked senders, or aggregate them and have a separate thread responsible
for them. In a well tuned system, we shouldn't see blocking of senders,
which represents a failure in dataflow management.
This is the executor thread . It has a few responsibilites . It listens for and acts upon events ( via channel ) . It complete senders that are blocked . It gets tasks and runs them . It might be better to have a select for blocked senders or aggregate them and have a separate thread responsible for them . In a well tuned system we shouldn ' t see blocking of senders which represents a failure in dataflow management .
fn spawn(mut self) -> Option<std::thread::JoinHandle<()>> {
        let thread = std::thread::spawn(move || {
            self.setup();
            let mut stats = ExecutorStats {
                id: self.id,
                ..Default::default()
            };
            let mut stats_event = SimpleEventTimer::default();
            let time_slice = get_time_slice();
            let backoff = LinearBackoff::new();
            log::debug!("executor {} is alive", self.id);
            let mut last_try_recv = Instant::now();
            'terminate: loop {
                // report to the system monitor, maybe look at fixing drift
                if stats_event.check() && self.monitor.send(MonitorMessage::ExecutorStats(stats)).is_err() {
                    log::debug!("failed to send exec stats to monitor");
                }
                if last_try_recv.elapsed() >= Duration::from_secs(2) {
                    last_try_recv += Duration::from_secs(2);
                    // processs any received commands, break out of loop if told to terminate
                    if self.try_recv(&stats) {
                        break 'terminate;
                    }
                }
                // move blocked senders along...
                if self.blocked_sender_count != 0 {
                    self.blocked_sender_count = self.try_completing_send(&mut stats);
                }
                // try to run a task
                let ran_task = if self.get_state() == ExecutorState::Running {
                    self.run_task(time_slice, &mut stats)
                } else {
                    false
                };
                // if no longer running and we don't have any blocked senders, we can leave
                if self.get_state() == ExecutorState::Parked {
                    tls_executor_data.with(|t| {
                        let tls = t.borrow();
                        self.blocked_sender_count = tls.blocked_senders.len();
                    });
                    if self.blocked_sender_count == 0 {
                        break;
                    }
                }
                // and after all that, we've got nothing left to do. Let's catch some zzzz's
                if self.blocked_sender_count == 0 && !ran_task {
                    if backoff.is_completed() {
                        log::trace!("executor {} is sleeping", self.id);
                        let start = std::time::Instant::now();
                        let park_duration = stats_event.remaining();
                        // sanity bailout
                        last_try_recv = start;
                        if self.try_recv(&stats) {
                            break 'terminate;
                        }
                        if RUN_QUEUE_LEN.load(Ordering::SeqCst) == 0 {
                            EXECUTORS_SNOOZING.fetch_add(1, Ordering::SeqCst);
                            thread::park_timeout(park_duration);
                            EXECUTORS_SNOOZING.fetch_sub(1, Ordering::SeqCst);
                            stats.sleep_count += 1;
                            stats.sleep_time += start.elapsed();
                        }
                        log::trace!("executor {} is awake", self.id);
                    } else {
                        backoff.snooze();
                    }
                } else if backoff.reset() {
                    stats.disturbed_nap += 1;
                }
            }
            log::debug!("executor {} is dead", self.id);
            log::debug!("{:#?}", stats);
            let remaining_tasks = self.work.len();
            if remaining_tasks > 0 {
                log::debug!(
                    "exec {} exiting with {} tasks in the worker q, will re-inject",
                    self.id,
                    remaining_tasks
                );
                // re-inject any remaining tasks
                while let Some(task) = self.work.pop() {
                    self.run_queue.push(task);
                }
            }
            tls_executor_data.with(|t| {
                let tls = t.borrow_mut();
                if !tls.blocked_senders.is_empty() {
                    log::error!(
                        "executor {} exited, but continues to have {} blocked senders",
                        self.id,
                        tls.blocked_senders.len()
                    );
                }
            });
            // as a last and final, tell the monitor that we're dead...
            if self.monitor.send(MonitorMessage::Terminated(self.id)).is_err() {
                log::warn!("executor {} exiting without informing system monitor", self.id);
            }
        });
        Some(thread)
    }
fn spawn ( mut self ) -> Option < std :: thread :: JoinHandle < ( ) > > { let thread = std :: thread :: spawn ( move | | { self . setup ( ) ; let mut stats = ExecutorStats { id : self . id , .. Default :: default ( ) } ; let mut stats_event = SimpleEventTimer :: default ( ) ; let time_slice = get_time_slice ( ) ; let backoff = LinearBackoff :: new ( ) ; log :: debug ! ( " " self id ) ; let mut last_try_recv = Instant :: now ( ) ; ' terminate : loop { // report to the system monitor, maybe look at fixing drift if stats_event . check ( ) && self . monitor . send ( MonitorMessage :: ExecutorStats ( stats ) ) . is_err ( ) { log :: debug ! ( " " ) ; } if last_try_recv . elapsed ( ) >= Duration :: from_secs ( 2 ) { last_try_recv += Duration :: from_secs ( 2 ) ; // processs any received commands, break out of loop if told to terminate if self . try_recv ( & stats ) { break ' terminate ; } } // move blocked senders along... if self . blocked_sender_count != 0 { self . blocked_sender_count = self . try_completing_send ( & mut stats ) ; } // try to run a task let ran_task = if self . get_state ( ) == ExecutorState :: Running { self . run_task ( time_slice , & mut stats ) } else { false } ; // if no longer running and we don't have any blocked senders, we can leave if self . get_state ( ) == ExecutorState :: Parked { tls_executor_data . with ( | t | { let tls = t . borrow ( ) ; self . blocked_sender_count = tls . blocked_senders . len ( ) ; } ) ; if self . blocked_sender_count == 0 { break ; } } // and after all that, we've got nothing left to do. Let's catch some zzzz's if self . blocked_sender_count == 0 && ! ran_task { if backoff . is_completed ( ) { log :: trace ! ( " " self id ) ; let start = std :: time :: Instant :: now ( ) ; let park_duration = stats_event . remaining ( ) ; // sanity bailout last_try_recv = start ; if self . try_recv ( & stats ) { break ' terminate ; } if RUN_QUEUE_LEN . load ( Ordering :: SeqCst ) == 0 { EXECUTORS_SNOOZING . fetch_add ( 1 , Ordering :: SeqCst ) ; thread :: park_timeout ( park_duration ) ; EXECUTORS_SNOOZING . fetch_sub ( 1 , Ordering :: SeqCst ) ; stats . sleep_count += 1 ; stats . sleep_time += start . elapsed ( ) ; } log :: trace ! ( " " self id ) ; } else { backoff . snooze ( ) ; } } else if backoff . reset ( ) { stats . disturbed_nap += 1 ; } } log :: debug ! ( " " self id ) ; log :: debug ! ( " " stats ) ; let remaining_tasks = self . work . len ( ) ; if remaining_tasks > 0 { log :: debug ! ( " " self id remaining_tasks ) ; // re-inject any remaining tasks while let Some ( task ) = self . work . pop ( ) { self . run_queue . push ( task ) ; } } tls_executor_data . with ( | t | { let tls = t . borrow_mut ( ) ; if ! tls . blocked_senders . is_empty ( ) { log :: error ! ( " " self id tls blocked_senders len ( ) ) ; } } ) ; // as a last and final, tell the monitor that we're dead... if self . monitor . send ( MonitorMessage :: Terminated ( self . id ) ) . is_err ( ) { log :: warn ! ( " " self id ) ; } } ) ; Some ( thread ) }
This is the executor thread.
This is the executor thread .
mut self