Function tpc_event_loop

nodedb-bridge/tests/cross_runtime.rs:30–86

Simulated TPC event loop using raw epoll + eventfd. This is what Glommio/monoio would do internally — poll an fd, wake up, process work, signal the other side.

(
    mut data: nodedb_bridge::async_bridge::DataHandle<u64, u64>,
    done: Arc<AtomicBool>,
)

/// This is what Glommio/monoio would do internally — poll an fd, wake up,
/// process work, signal the other side.
fn tpc_event_loop(
    mut data: nodedb_bridge::async_bridge::DataHandle<u64, u64>,
    done: Arc<AtomicBool>,
) {
    let epoll_fd = unsafe { libc::epoll_create1(0) };
    assert!(epoll_fd >= 0, "epoll_create1 failed");

    // Register the request-available eventfd with epoll (edge-triggered).
    let req_fd = data.request_wake_fd();
    let mut event = libc::epoll_event {
        events: (libc::EPOLLIN | libc::EPOLLET) as u32,
        u64: req_fd as u64,
    };
    let ret = unsafe { libc::epoll_ctl(epoll_fd, libc::EPOLL_CTL_ADD, req_fd, &mut event) };
    assert_eq!(ret, 0, "epoll_ctl failed");

    let mut processed = 0u64;
    let mut events = [libc::epoll_event { events: 0, u64: 0 }; 8];

    while processed < MESSAGE_COUNT {
        // epoll_wait with short timeout so we don't hang on edge-triggered misses.
        let nfds = unsafe { libc::epoll_wait(epoll_fd, events.as_mut_ptr(), 8, 10) };

        if nfds < 0 {
            let err = std::io::Error::last_os_error();
            if err.kind() == std::io::ErrorKind::Interrupted {
                continue;
            }
            panic!("epoll_wait failed: {err}");
        }

        // Consume the eventfd signal to re-arm edge trigger.
        if nfds > 0 {
            let _ = data.req_wake.consumer_wake.try_read();
        }

        // Drain all available requests.
        let mut batch = Vec::new();
        data.drain_requests(&mut batch, 512);

        for req in batch {
            loop {
                match data.try_send_response(req * 2) {
                    Ok(()) => break,
                    Err(nodedb_bridge::BridgeError::Full { .. }) => {
                        thread::yield_now();
                    }
                    Err(e) => panic!("TPC send error: {e}"),
                }
            }
            processed += 1;
        }
    }

    done.store(true, Ordering::Release);
    unsafe { libc::close(epoll_fd) };
}