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

48.51

/src/dftracer/utils/server/http_connection.cpp

#include <dftracer/utils/core/common/logging.h>
#include <dftracer/utils/core/io/ops.h>
#include <dftracer/utils/server/http_connection.h>
#include <dftracer/utils/server/http_request.h>
#include <dftracer/utils/server/http_response.h>
#include <dftracer/utils/server/router.h>
#include <sys/uio.h>

#include <cstring>
#include <vector>

#ifdef __linux__
#include <malloc.h>  // malloc_trim
#endif

namespace dftracer::utils::server {

coro::CoroTask<void> handle_connection(int client_fd,
                                       struct sockaddr_in /*addr*/,
                                       Router& router) {
    // 8 KiB receive buffer. For HTTP/1.1 GET requests this is plenty.
    // Requests larger than this are rejected as "too large".
    constexpr std::size_t BUF_SIZE = 8192;
    char buf[BUF_SIZE];
    std::size_t buf_used = 0;

    while (true) {
        // Read data from socket.
        ssize_t n = co_await io::recv(client_fd, buf + buf_used,
                                      BUF_SIZE - buf_used, 0);
        if (n <= 0) break;  // Connection closed or error
        buf_used += static_cast<std::size_t>(n);

        // Try to parse a complete request.
        HttpRequest req;
        int parsed = req.parse(buf, buf_used);
        if (parsed == -2) {
            // Incomplete — need more data.
            if (buf_used >= BUF_SIZE) {
                // Buffer full but still no complete request.
                auto resp = HttpResponse::bad_request("Request too large");
                auto out = resp.serialize();
                co_await io::send(client_fd, out.data(), out.size(), 0);
                break;
            }
            continue;
        }
        if (parsed < 0) {
            auto resp = HttpResponse::bad_request("Malformed HTTP request");
            auto out = resp.serialize();
            co_await io::send(client_fd, out.data(), out.size(), 0);
            break;
        }

        // Route and handle. Register a cancellation token keyed by the
        // client-supplied request id so a separate cancel request can reach it.
        std::string req_id(req.header("x-request-id"));
        CancelToken token =
            CancelRegistry::instance().create(req_id, client_fd);
        req.cancel_token = token;

        HttpResponse resp;
        try {
            resp = co_await router.handle(req);
        } catch (const std::exception& e) {
            DFTRACER_UTILS_LOG_ERROR("Handler exception: %s", e.what());
            resp = HttpResponse::internal_error(e.what());
        } catch (...) {
            DFTRACER_UTILS_LOG_ERROR("Handler threw unknown exception");
            resp = HttpResponse::internal_error("Internal server error");
        }
        if (resp.is_streaming()) {
            auto hdrs = resp.serialize_headers();
            auto hdr_rc =
                co_await io::send(client_fd, hdrs.data(), hdrs.size(), 0);
            if (hdr_rc < 0) goto stream_done;

            {
                static constexpr char newline_ch = '\n';
                static constexpr char crlf[] = "\r\n";
                char hex[24];
                std::vector<struct iovec> iovs;

                while (auto chunk = co_await resp.stream->next()) {
                    if (chunk->views.empty()) continue;

                    std::size_t payload_size = 0;
                    for (const auto& sv : chunk->views) {
                        payload_size += sv.size() + 1;
                    }

                    int hex_len = std::snprintf(hex, sizeof(hex), "%zx\r\n",
                                                payload_size);

                    iovs.clear();
                    iovs.reserve(chunk->views.size() * 2 + 2);
                    iovs.push_back({hex, static_cast<std::size_t>(hex_len)});
                    for (const auto& sv : chunk->views) {
                        iovs.push_back(
                            {const_cast<char*>(sv.data()), sv.size()});
                        iovs.push_back({const_cast<char*>(&newline_ch), 1});
                    }
                    iovs.push_back({const_cast<char*>(crlf), 2});

                    auto rc = co_await io::writev_all(
                        client_fd, iovs.data(), static_cast<int>(iovs.size()));
                    if (rc < 0) {
                        DFTRACER_UTILS_LOG_ERROR(
                            "Streaming write failed; closing connection");
                        token.cancel();
                        goto stream_done;
                    }
                }
            }
            co_await io::send(client_fd, "0\r\n\r\n", 5, 0);
        stream_done:;
        } else {
            auto out = resp.serialize();
            co_await io::send(client_fd, out.data(), out.size(), 0);
        }

        CancelRegistry::instance().remove(req_id);

        // Consume parsed bytes; shift any remaining data.
        auto consumed = static_cast<std::size_t>(parsed);
        if (consumed < buf_used) {
            std::memmove(buf, buf + consumed, buf_used - consumed);
            buf_used -= consumed;
        } else {
            buf_used = 0;
        }

        // Check Connection: close or HTTP/1.0 (no keep-alive).
        if (req.minor_version == 0 || req.has_header("connection", "close")) {
            break;
        }
    }

    // Note: the caller (tcp_listener) is responsible for closing
    // client_fd via io::close() after this coroutine returns.

#ifdef __linux__
    // Return freed heap pages to the OS.  Each request may decompress
    // large gzip buffers and build sizeable JSON responses; without
    // this call glibc's ptmalloc2 keeps those arenas mapped, causing
    // RSS to grow monotonically.
    ::malloc_trim(0);
#endif
}

}  // namespace dftracer::utils::server

1	#include <dftracer/utils/core/common/logging.h>
2	#include <dftracer/utils/core/io/ops.h>
3	#include <dftracer/utils/server/http_connection.h>
4	#include <dftracer/utils/server/http_request.h>
5	#include <dftracer/utils/server/http_response.h>
6	#include <dftracer/utils/server/router.h>
7	#include <sys/uio.h>
8
9	#include <cstring>
10	#include <vector>
11
12	#ifdef __linux__
13	#include <malloc.h> // malloc_trim
14	#endif
15
16	namespace dftracer::utils::server {
17
18	coro::CoroTask<void> handle_connection(int client_fd,	552!
19	struct sockaddr_in /addr/,
20	Router& router) {	64!
21	// 8 KiB receive buffer. For HTTP/1.1 GET requests this is plenty.
22	// Requests larger than this are rejected as "too large".
23	constexpr std::size_t BUF_SIZE = 8192;	64✔
24	char buf[BUF_SIZE];	64✔
25	std::size_t buf_used = 0;	64✔
26
27	while (true) {	64✔
28	// Read data from socket.
29	ssize_t n = co_await io::recv(client_fd, buf + buf_used,	256!
30	BUF_SIZE - buf_used, 0);	64✔
31	if (n <= 0) break; // Connection closed or error	174✔
32	buf_used += static_cast<std::size_t>(n);	165✔
33
34	// Try to parse a complete request.
35	HttpRequest req;	165✔
36	int parsed = req.parse(buf, buf_used);	165!
37	if (parsed == -2) {	165!
38	// Incomplete — need more data.
UNCOV 39	if (buf_used >= BUF_SIZE) {	×
40	// Buffer full but still no complete request.
UNCOV 41	auto resp = HttpResponse::bad_request("Request too large");	×
UNCOV 42	auto out = resp.serialize();	×
UNCOV 43	co_await io::send(client_fd, out.data(), out.size(), 0);	×
UNCOV 44	break;	×
UNCOV 45	}	×
UNCOV 46	continue;	×
47	}
48	if (parsed < 0) {	165!
UNCOV 49	auto resp = HttpResponse::bad_request("Malformed HTTP request");	×
UNCOV 50	auto out = resp.serialize();	×
UNCOV 51	co_await io::send(client_fd, out.data(), out.size(), 0);	×
UNCOV 52	break;	×
UNCOV 53	}	×
54
55	// Route and handle. Register a cancellation token keyed by the
56	// client-supplied request id so a separate cancel request can reach it.
57	std::string req_id(req.header("x-request-id"));	165!
58	CancelToken token =	165✔
59	CancelRegistry::instance().create(req_id, client_fd);	165✔
60	req.cancel_token = token;	165✔
61
62	HttpResponse resp;	165!
63	try {
64	resp = co_await router.handle(req);	220!
65	} catch (const std::exception& e) {	55!
UNCOV 66	DFTRACER_UTILS_LOG_ERROR("Handler exception: %s", e.what());	×
UNCOV 67	resp = HttpResponse::internal_error(e.what());	×
UNCOV 68	} catch (...) {	×
UNCOV 69	DFTRACER_UTILS_LOG_ERROR("Handler threw unknown exception");	×
UNCOV 70	resp = HttpResponse::internal_error("Internal server error");	×
UNCOV 71	}	×
72	if (resp.is_streaming()) {	55!
73	auto hdrs = resp.serialize_headers();	3!
74	auto hdr_rc =	6✔
75	co_await io::send(client_fd, hdrs.data(), hdrs.size(), 0);	6!
76	if (hdr_rc < 0) goto stream_done;	3!
77
78	{
79	static constexpr char newline_ch = '\n';
80	static constexpr char crlf[] = "\r\n";
81	char hex[24];	3✔
82	std::vector<struct iovec> iovs;	3✔
83
84	while (auto chunk = co_await resp.stream->next()) {	24!
85	if (chunk->views.empty()) continue;	9!
86
87	std::size_t payload_size = 0;	9✔
88	for (const auto& sv : chunk->views) {	2,069✔
89	payload_size += sv.size() + 1;	2,060✔
90	}	2,060✔
91
92	int hex_len = std::snprintf(hex, sizeof(hex), "%zx\r\n",	18✔
93	payload_size);	9✔
94
95	iovs.clear();	9✔
96	iovs.reserve(chunk->views.size() * 2 + 2);	9!
97	iovs.push_back({hex, static_cast<std::size_t>(hex_len)});	9!
98	for (const auto& sv : chunk->views) {	2,069✔
99	iovs.push_back(	2,060!
100	{const_cast<char*>(sv.data()), sv.size()});	2,060✔
101	iovs.push_back({const_cast<char*>(&newline_ch), 1});	2,060!
102	}	2,060✔
103	iovs.push_back({const_cast<char*>(crlf), 2});	9!
104
105	auto rc = co_await io::writev_all(	21!
106	client_fd, iovs.data(), static_cast<int>(iovs.size()));	9✔
107	if (rc < 0) {	3!
UNCOV 108	DFTRACER_UTILS_LOG_ERROR(	×
109	"Streaming write failed; closing connection");
NEW 110	token.cancel();	×
UNCOV 111	goto stream_done;	×
112	}
113	}	12!
114	}	9!
115	co_await io::send(client_fd, "0\r\n\r\n", 5, 0);	6!
116	stream_done:;
117	} else {	9✔
118	auto out = resp.serialize();	52!
119	co_await io::send(client_fd, out.data(), out.size(), 0);	104!
120	}	52!
121
122	CancelRegistry::instance().remove(req_id);	55!
123
124	// Consume parsed bytes; shift any remaining data.
125	auto consumed = static_cast<std::size_t>(parsed);	55✔
126	if (consumed < buf_used) {	55!
UNCOV 127	std::memmove(buf, buf + consumed, buf_used - consumed);	×
UNCOV 128	buf_used -= consumed;	×
UNCOV 129	} else {	×
130	buf_used = 0;	55✔
131	}
132
133	// Check Connection: close or HTTP/1.0 (no keep-alive).
134	if (req.minor_version == 0 \|\| req.has_header("connection", "close")) {	55!
135	break;	55✔
136	}
137	}	180!
138
139	// Note: the caller (tcp_listener) is responsible for closing
140	// client_fd via io::close() after this coroutine returns.
141
142	#ifdef __linux__
143	// Return freed heap pages to the OS. Each request may decompress
144	// large gzip buffers and build sizeable JSON responses; without
145	// this call glibc's ptmalloc2 keeps those arenas mapped, causing
146	// RSS to grow monotonically.
147	::malloc_trim(0);
148	#endif
149	}	760✔
150
151	} // 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