Method start_thread

rate_limiter/src/group.rs:219–284

Start a worker thread to broadcast an event to each RateLimiterGroupHandle when the RateLimiter becomes unblocked.

(&mut self, exit_evt: EventFd)

    /// Start a worker thread to broadcast an event to each RateLimiterGroupHandle
    /// when the RateLimiter becomes unblocked.
    pub fn start_thread(&mut self, exit_evt: EventFd) -> result::Result<(), Error> {
        let inner = self.inner.clone();
        let epoll_fd = self.epoll_file.as_raw_fd();
        thread::Builder::new()
            .name(format!("rate-limit-group-{}", inner.id))
            .spawn(move || {
                let res = std::panic::catch_unwind(AssertUnwindSafe(move || {
                    const EPOLL_EVENTS_LEN: usize = 2;

                    let mut events =
                        [epoll::Event::new(epoll::Events::empty(), 0); EPOLL_EVENTS_LEN];

                    loop {
                        let num_events = match epoll::wait(epoll_fd, -1, &mut events[..]) {
                            Ok(res) => res,
                            Err(e) => {
                                if e.kind() == io::ErrorKind::Interrupted {
                                    continue;
                                }
                                return Err(Error::Epoll(e));
                            }
                        };

                        for event in events.iter().take(num_events) {
                            let dispatch_event: EpollDispatch = event.data.into();
                            match dispatch_event {
                                EpollDispatch::Unknown => {
                                    let event = event.data;
                                    warn!("Unknown rate-limiter loop event: {event}");
                                }
                                EpollDispatch::Unblocked => {
                                    inner.rate_limiter.event_handler().unwrap();
                                    let handles = inner.handles.lock().unwrap();
                                    for handle in handles.iter() {
                                        handle.write(1).map_err(Error::EventFdWrite)?;
                                    }
                                }
                                EpollDispatch::Kill => {
                                    info!(
                                        "KILL_EVENT received, stopping rate-limit-group epoll loop"
                                    );
                                    return Ok(());
                                }
                            }
                        }
                    }
                }));

                match res {
                    Ok(res) => {
                        if let Err(e) = res {
                            error!("Error running rate-limit-group worker: {e:?}");
                            exit_evt.write(1).unwrap();
                        }
                    }
                    Err(_) => {
                        error!("rate-limit-group worker panicked");
                        exit_evt.write(1).unwrap();