29187058583

Committed 12 Jul 2026 09:11AM UTC coverage: 51.324% (-1.4%) from 52.754%

Build # 29187058583

Build Type

Pull #96

github

Committed by

web-flow

Commit Message

Merge 7e90b76dd into 056c79287

Pull Request Pull Request #96: fix: consolidate stale-index rebuilds across consumers and fix multi-run breaks

Coverage Stats

33807 of 84432 branches covered (40.04%)

Branch coverage included in aggregate %.

145 of 152 new or added lines in 16 files covered. (95.39%)

5186 existing lines in 197 files now uncovered.

34558 of 48770 relevant lines covered (70.86%)

10797.15 hits per line

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

46.82

/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.
        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");
                        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);
        }

        // 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,	425!
19	struct sockaddr_in /addr/,
20	Router& router) {	49!
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;	49✔
24	char buf[BUF_SIZE];	49✔
25	std::size_t buf_used = 0;	49✔
26
27	while (true) {	49✔
28	// Read data from socket.
29	ssize_t n = co_await io::recv(client_fd, buf + buf_used,	196!
30	BUF_SIZE - buf_used, 0);	49✔
31	if (n <= 0) break; // Connection closed or error	133✔
32	buf_used += static_cast<std::size_t>(n);	126✔
33
34	// Try to parse a complete request.
35	HttpRequest req;	126✔
36	int parsed = req.parse(buf, buf_used);	126!
37	if (parsed == -2) {	126!
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) {	126!
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.
56	HttpResponse resp;	126!
57	try {
58	resp = co_await router.handle(req);	168!
59	} catch (const std::exception& e) {	42!
UNCOV 60	DFTRACER_UTILS_LOG_ERROR("Handler exception: %s", e.what());	×
UNCOV 61	resp = HttpResponse::internal_error(e.what());	×
UNCOV 62	} catch (...) {	×
UNCOV 63	DFTRACER_UTILS_LOG_ERROR("Handler threw unknown exception");	×
UNCOV 64	resp = HttpResponse::internal_error("Internal server error");	×
UNCOV 65	}	×
66	if (resp.is_streaming()) {	42!
67	auto hdrs = resp.serialize_headers();	3!
68	auto hdr_rc =	6✔
69	co_await io::send(client_fd, hdrs.data(), hdrs.size(), 0);	6!
70	if (hdr_rc < 0) goto stream_done;	3!
71
72	{
73	static constexpr char newline_ch = '\n';
74	static constexpr char crlf[] = "\r\n";
75	char hex[24];	3✔
76	std::vector<struct iovec> iovs;	3✔
77
78	while (auto chunk = co_await resp.stream->next()) {	24!
79	if (chunk->views.empty()) continue;	9!
80
81	std::size_t payload_size = 0;	9✔
82	for (const auto& sv : chunk->views) {	2,069✔
83	payload_size += sv.size() + 1;	2,060✔
84	}	2,060✔
85
86	int hex_len = std::snprintf(hex, sizeof(hex), "%zx\r\n",	18✔
87	payload_size);	9✔
88
89	iovs.clear();	9✔
90	iovs.reserve(chunk->views.size() * 2 + 2);	9!
91	iovs.push_back({hex, static_cast<std::size_t>(hex_len)});	9!
92	for (const auto& sv : chunk->views) {	2,069✔
93	iovs.push_back(	2,060!
94	{const_cast<char*>(sv.data()), sv.size()});	2,060✔
95	iovs.push_back({const_cast<char*>(&newline_ch), 1});	2,060!
96	}	2,060✔
97	iovs.push_back({const_cast<char*>(crlf), 2});	9!
98
99	auto rc = co_await io::writev_all(	21!
100	client_fd, iovs.data(), static_cast<int>(iovs.size()));	9✔
101	if (rc < 0) {	3!
UNCOV 102	DFTRACER_UTILS_LOG_ERROR(	×
103	"Streaming write failed; closing connection");
UNCOV 104	goto stream_done;	×
105	}
106	}	12!
107	}	9!
108	co_await io::send(client_fd, "0\r\n\r\n", 5, 0);	6!
109	stream_done:;
110	} else {	9✔
111	auto out = resp.serialize();	39!
112	co_await io::send(client_fd, out.data(), out.size(), 0);	78!
113	}	39!
114
115	// Consume parsed bytes; shift any remaining data.
116	auto consumed = static_cast<std::size_t>(parsed);	42✔
117	if (consumed < buf_used) {	42!
UNCOV 118	std::memmove(buf, buf + consumed, buf_used - consumed);	×
UNCOV 119	buf_used -= consumed;	×
UNCOV 120	} else {	×
121	buf_used = 0;	42✔
122	}
123
124	// Check Connection: close or HTTP/1.0 (no keep-alive).
125	if (req.minor_version == 0 \|\| req.has_header("connection", "close")) {	42!
126	break;	42✔
127	}
128	}	139!
129
130	// Note: the caller (tcp_listener) is responsible for closing
131	// client_fd via io::close() after this coroutine returns.
132
133	#ifdef __linux__
134	// Return freed heap pages to the OS. Each request may decompress
135	// large gzip buffers and build sizeable JSON responses; without
136	// this call glibc's ptmalloc2 keeps those arenas mapped, causing
137	// RSS to grow monotonically.
138	::malloc_trim(0);
139	#endif
140	}	253✔
141
142	} // namespace dftracer::utils::server

llnl / dftracer-utils / 29187058583

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