28693295402

Committed 04 Jul 2026 03:17AM UTC coverage: 52.408% (+0.1%) from 52.278%

Build # 28693295402

Build Type

push

github

Committed by

hariharan-devarajan

Commit Message

feat: silence noisy warnings on aarch64

Coverage Stats

37318 of 92666 branches covered (40.27%)

Branch coverage included in aggregate %.

33462 of 42389 relevant lines covered (78.94%)

20557.64 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

67.86

/src/dftracer/utils/core/io/kqueue_thread_pool_backend.cpp

#if defined(__APPLE__) || defined(__FreeBSD__) || defined(__OpenBSD__) || \
    defined(__NetBSD__) || defined(__DragonFly__)

#include <dftracer/utils/core/common/logging.h>
#include <dftracer/utils/core/io/kqueue_thread_pool_backend.h>
#include <dftracer/utils/core/pipeline/executor.h>
#include <sys/event.h>
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>

#include <cerrno>
#include <cstring>

namespace dftracer::utils::io {

KqueueThreadPoolBackend::KqueueThreadPoolBackend(Executor& executor,
                                                 std::size_t pool_size,
                                                 unsigned batch_threshold)
    : ThreadPoolFileOps(executor, pool_size, batch_threshold) {}

KqueueThreadPoolBackend::~KqueueThreadPoolBackend() {
    // Ensure cleanup even if stop() was not called.
    if (kqueue_fd_ >= 0) {
        ::close(kqueue_fd_);
        kqueue_fd_ = -1;
    }
}

void KqueueThreadPoolBackend::start() {
    pool_.start();

    kqueue_fd_ = ::kqueue();
    if (kqueue_fd_ < 0) {
        DFTRACER_UTILS_LOG_ERROR("kqueue() failed: %s", std::strerror(errno));
        return;
    }

    // Register a user event (EVFILT_USER) for shutdown signaling.
    struct kevent ev{};
    EV_SET(&ev, SHUTDOWN_IDENT, EVFILT_USER, EV_ADD | EV_CLEAR, 0, 0, nullptr);
    if (::kevent(kqueue_fd_, &ev, 1, nullptr, 0, nullptr) < 0) {
        DFTRACER_UTILS_LOG_ERROR("kevent(register EVFILT_USER) failed: %s",
                                 std::strerror(errno));
    }

    DFTRACER_UTILS_LOG_DEBUG("kqueue+threadpool backend started (kqueue_fd=%d)",
                             kqueue_fd_);

    completion_thread_.start([this] { kqueue_loop(); });
}

void KqueueThreadPoolBackend::stop() {
    // Signal the completion thread to exit.
    completion_thread_.signal_stop();

    // Wake kevent() by triggering the user event.
    if (kqueue_fd_ >= 0) {
        struct kevent ev{};
        EV_SET(&ev, SHUTDOWN_IDENT, EVFILT_USER, 0, NOTE_TRIGGER, 0, nullptr);
        ::kevent(kqueue_fd_, &ev, 1, nullptr, 0, nullptr);
    }

    completion_thread_.join();
    pool_.stop();

    if (kqueue_fd_ >= 0) {
        ::close(kqueue_fd_);
        kqueue_fd_ = -1;
    }
}

void KqueueThreadPoolBackend::kqueue_loop() {
    constexpr int MAX_EVENTS = 64;
    struct kevent events[MAX_EVENTS];

    // 100ms timeout for periodic running() check.
    struct timespec timeout{};
    timeout.tv_sec = 0;
    timeout.tv_nsec = 100 * 1000 * 1000;  // 100ms

    while (completion_thread_.running()) {
        int n = ::kevent(kqueue_fd_, nullptr, 0, events, MAX_EVENTS, &timeout);
        if (n < 0) {
            if (errno == EINTR) continue;
            break;
        }

        for (int i = 0; i < n; ++i) {
            if (events[i].filter == EVFILT_USER &&
                events[i].ident == SHUTDOWN_IDENT) {
                // Shutdown signal -- process any other events first.
                continue;
            }

            // Future: dispatch socket I/O events here.
            // For now, only the user event is registered.
        }
    }
}

}  // namespace dftracer::utils::io

#endif  // kqueue platforms

1	#if defined(__APPLE__) \|\| defined(__FreeBSD__) \|\| defined(__OpenBSD__) \|\| \
2	defined(__NetBSD__) \|\| defined(__DragonFly__)
3
4	#include <dftracer/utils/core/common/logging.h>
5	#include <dftracer/utils/core/io/kqueue_thread_pool_backend.h>
6	#include <dftracer/utils/core/pipeline/executor.h>
7	#include <sys/event.h>
8	#include <sys/time.h>
9	#include <sys/types.h>
10	#include <unistd.h>
11
12	#include <cerrno>
13	#include <cstring>
14
15	namespace dftracer::utils::io {
16
17	KqueueThreadPoolBackend::KqueueThreadPoolBackend(Executor& executor,	1,602✔
18	std::size_t pool_size,
19	unsigned batch_threshold)
20	: ThreadPoolFileOps(executor, pool_size, batch_threshold) {}	1,602✔
21
22	KqueueThreadPoolBackend::~KqueueThreadPoolBackend() {	1,602✔
23	// Ensure cleanup even if stop() was not called.
24	if (kqueue_fd_ >= 0) {	534!
25	::close(kqueue_fd_);	×
26	kqueue_fd_ = -1;	×
27	}	×
28	}	1,602✔
29
30	void KqueueThreadPoolBackend::start() {	534✔
31	pool_.start();	534✔
32
33	kqueue_fd_ = ::kqueue();	534✔
34	if (kqueue_fd_ < 0) {	534!
35	DFTRACER_UTILS_LOG_ERROR("kqueue() failed: %s", std::strerror(errno));	×
36	return;	×
37	}
38
39	// Register a user event (EVFILT_USER) for shutdown signaling.
40	struct kevent ev{};	534✔
41	EV_SET(&ev, SHUTDOWN_IDENT, EVFILT_USER, EV_ADD \| EV_CLEAR, 0, 0, nullptr);	534✔
42	if (::kevent(kqueue_fd_, &ev, 1, nullptr, 0, nullptr) < 0) {	534!
43	DFTRACER_UTILS_LOG_ERROR("kevent(register EVFILT_USER) failed: %s",	×
44	std::strerror(errno));
45	}	×
46
47	DFTRACER_UTILS_LOG_DEBUG("kqueue+threadpool backend started (kqueue_fd=%d)",	534✔
48	kqueue_fd_);
49
50	completion_thread_.start([this] { kqueue_loop(); });	1,068!
51	}	534✔
52
53	void KqueueThreadPoolBackend::stop() {	534✔
54	// Signal the completion thread to exit.
55	completion_thread_.signal_stop();	534✔
56
57	// Wake kevent() by triggering the user event.
58	if (kqueue_fd_ >= 0) {	534!
59	struct kevent ev{};	534✔
60	EV_SET(&ev, SHUTDOWN_IDENT, EVFILT_USER, 0, NOTE_TRIGGER, 0, nullptr);	534✔
61	::kevent(kqueue_fd_, &ev, 1, nullptr, 0, nullptr);	534✔
62	}	534✔
63
64	completion_thread_.join();	534✔
65	pool_.stop();	534✔
66
67	if (kqueue_fd_ >= 0) {	534!
68	::close(kqueue_fd_);	534✔
69	kqueue_fd_ = -1;	534✔
70	}	534✔
71	}	534✔
72
73	void KqueueThreadPoolBackend::kqueue_loop() {	534✔
74	constexpr int MAX_EVENTS = 64;	534✔
75	struct kevent events[MAX_EVENTS];
76
77	// 100ms timeout for periodic running() check.
78	struct timespec timeout{};	534✔
79	timeout.tv_sec = 0;	534✔
80	timeout.tv_nsec = 100 * 1000 * 1000; // 100ms	534✔
81
82	while (completion_thread_.running()) {	1,814✔
83	int n = ::kevent(kqueue_fd_, nullptr, 0, events, MAX_EVENTS, &timeout);	1,280✔
84	if (n < 0) {	1,280!
85	if (errno == EINTR) continue;	×
86	break;	×
87	}
88
89	for (int i = 0; i < n; ++i) {	1,814✔
90	if (events[i].filter == EVFILT_USER &&	534!
91	events[i].ident == SHUTDOWN_IDENT) {	534✔
92	// Shutdown signal -- process any other events first.
93	continue;	534✔
94	}
95
96	// Future: dispatch socket I/O events here.
97	// For now, only the user event is registered.
98	}	×
99	}
100	}	534✔
101
102	} // namespace dftracer::utils::io
103
104	#endif // kqueue platforms

llnl / dftracer-utils / 28693295402

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous