27052412546

Committed 06 Jun 2026 04:20AM UTC coverage: 50.862% (+1.0%) from 49.905%

Build # 27052412546

Build Type

Pull #73

github

Committed by

web-flow

Commit Message

Merge 734572730 into 88a3c8457

Pull Request Pull Request #73: add portable dependencies wheel support

Coverage Stats

31801 of 79859 branches covered (39.82%)

Branch coverage included in aggregate %.

32491 of 46545 relevant lines covered (69.81%)

9947.11 hits per line

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

46.53

/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>

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

llnl / dftracer-utils / 27052412546

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