29787659139

Committed 20 Jul 2026 11:34PM UTC coverage: 51.578% (-1.1%) from 52.66%

Build # 29787659139

Build Type

Pull #99

github

Committed by

web-flow

Commit Message

Merge ee805adeb into 06bc84ec9

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

Coverage Stats

34832 of 86278 branches covered (40.37%)

Branch coverage included in aggregate %.

1034 of 1319 new or added lines in 30 files covered. (78.39%)

5193 existing lines in 197 files now uncovered.

35349 of 49791 relevant lines covered (70.99%)

9765.54 hits per line

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

87.63

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

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

llnl / dftracer-utils / 29787659139

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