29863677266

Committed 21 Jul 2026 07:57PM UTC coverage: 52.767% (+0.1%) from 52.66%

Build # 29863677266

Build Type

Pull #99

github

Committed by

web-flow

Commit Message

Merge 7169a8c39 into 06bc84ec9

Pull Request Pull Request #99: Support CM time_metric (NS/MS/SEC/US) across reader and viz

Coverage Stats

41662 of 102203 branches covered (40.76%)

Branch coverage included in aggregate %.

1724 of 2125 new or added lines in 32 files covered. (81.13%)

92 existing lines in 6 files now uncovered.

36790 of 46473 relevant lines covered (79.16%)

60195.88 hits per line

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

86.36

/src/dftracer/utils/server/cancel.cpp

#include <dftracer/utils/server/cancel.h>
#include <dftracer/utils/server/signal_handler.h>
#include <sys/socket.h>

#include <cerrno>

namespace dftracer::utils::server {

namespace {

// Non-blocking peek: true if the peer has closed or reset the connection.
bool peer_disconnected(int fd) {
    char b;
    ssize_t n = ::recv(fd, &b, 1, MSG_PEEK | MSG_DONTWAIT);
    if (n == 0) return true;                     // orderly shutdown (FIN)
    if (n < 0) {
        int e = errno;
        return e != EAGAIN && e != EWOULDBLOCK;  // reset/error vs. no data yet
    }
    return false;  // data buffered (e.g. a pipelined request): still connected
}

}  // namespace

bool CancelToken::cancelled() const {
    // Shutdown cancels everything, including work started without a token, so
    // Ctrl-C does not wait out a long scan.
    if (g_shutdown_requested.load(std::memory_order_acquire)) return true;
    if (!state_) return false;
    if (state_->flag.load(std::memory_order_relaxed)) return true;
    if (state_->peer_fd < 0) return false;
    // Probe the socket only every 16th check to keep the hot path cheap.
    if ((state_->probe_tick.fetch_add(1, std::memory_order_relaxed) & 0xF) != 0)
        return false;
    if (peer_disconnected(state_->peer_fd)) {
        state_->flag.store(true, std::memory_order_relaxed);
        return true;
    }
    return false;
}

CancelRegistry& CancelRegistry::instance() {
    static CancelRegistry registry;
    return registry;
}

CancelToken CancelRegistry::create(const std::string& id, int peer_fd) {
    auto state = std::make_shared<CancelState>();
    state->peer_fd = peer_fd;
    if (!id.empty()) {
        std::lock_guard<std::mutex> lock(mu_);
        tokens_[id] = state;
    }
    return CancelToken(state);
}

void CancelRegistry::remove(const std::string& id) {
    if (id.empty()) return;
    std::lock_guard<std::mutex> lock(mu_);
    tokens_.erase(id);
}

bool CancelRegistry::cancel(const std::string& id) {
    if (id.empty()) return false;
    std::shared_ptr<CancelState> state;
    {
        std::lock_guard<std::mutex> lock(mu_);
        auto it = tokens_.find(id);
        if (it == tokens_.end()) return false;
        state = it->second;
    }
    state->flag.store(true, std::memory_order_relaxed);
    return true;
}

std::size_t CancelRegistry::size() const {
    std::lock_guard<std::mutex> lock(mu_);
    return tokens_.size();
}

}  // namespace dftracer::utils::server

1	#include <dftracer/utils/server/cancel.h>
2	#include <dftracer/utils/server/signal_handler.h>
3	#include <sys/socket.h>
4
5	#include <cerrno>
6
7	namespace dftracer::utils::server {
8
9	namespace {
10
11	// Non-blocking peek: true if the peer has closed or reset the connection.
12	bool peer_disconnected(int fd) {	40✔
13	char b;
14	ssize_t n = ::recv(fd, &b, 1, MSG_PEEK \| MSG_DONTWAIT);	40!
15	if (n == 0) return true; // orderly shutdown (FIN)	40✔
16	if (n < 0) {	38✔
17	int e = errno;	38✔
18	return e != EAGAIN && e != EWOULDBLOCK; // reset/error vs. no data yet	38!
19	}
NEW 20	return false; // data buffered (e.g. a pipelined request): still connected	×
21	}	20✔
22
23	} // namespace
24
25	bool CancelToken::cancelled() const {	338✔
26	// Shutdown cancels everything, including work started without a token, so
27	// Ctrl-C does not wait out a long scan.
28	if (g_shutdown_requested.load(std::memory_order_acquire)) return true;	338!
29	if (!state_) return false;	338✔
30	if (state_->flag.load(std::memory_order_relaxed)) return true;	278✔
31	if (state_->peer_fd < 0) return false;	268✔
32	// Probe the socket only every 16th check to keep the hot path cheap.
33	if ((state_->probe_tick.fetch_add(1, std::memory_order_relaxed) & 0xF) != 0)	399✔
34	return false;	226✔
35	if (peer_disconnected(state_->peer_fd)) {	40✔
36	state_->flag.store(true, std::memory_order_relaxed);	2✔
37	return true;	2✔
38	}
39	return false;	38✔
40	}	169✔
41
42	CancelRegistry& CancelRegistry::instance() {	258✔
43	static CancelRegistry registry;	258✔
44	return registry;	258✔
45	}
46
47	CancelToken CancelRegistry::create(const std::string& id, int peer_fd) {	144✔
48	auto state = std::make_shared<CancelState>();	144!
49	state->peer_fd = peer_fd;	144✔
50	if (!id.empty()) {	144✔
51	std::lock_guard<std::mutex> lock(mu_);	16!
52	tokens_[id] = state;	16!
53	}	16✔
54	return CancelToken(state);	216!
55	}	144✔
56
57	void CancelRegistry::remove(const std::string& id) {	130✔
58	if (id.empty()) return;	130✔
59	std::lock_guard<std::mutex> lock(mu_);	4!
60	tokens_.erase(id);	4!
61	}	67✔
62
63	bool CancelRegistry::cancel(const std::string& id) {	14✔
64	if (id.empty()) return false;	14✔
65	std::shared_ptr<CancelState> state;	12✔
66	{
67	std::lock_guard<std::mutex> lock(mu_);	12!
68	auto it = tokens_.find(id);	12!
69	if (it == tokens_.end()) return false;	12✔
70	state = it->second;	6!
71	}	12✔
72	state->flag.store(true, std::memory_order_relaxed);	6✔
73	return true;	6✔
74	}	13✔
75
76	std::size_t CancelRegistry::size() const {	6✔
77	std::lock_guard<std::mutex> lock(mu_);	6!
78	return tokens_.size();	9✔
79	}	6✔
80
81	} // namespace dftracer::utils::server

llnl / dftracer-utils / 29863677266

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