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

77.98

/src/dftracer/utils/core/io/thread_pool_file_ops.cpp

#include <dftracer/utils/core/io/io_sendfile.h>
#include <dftracer/utils/core/io/thread_pool_file_ops.h>
#include <dftracer/utils/core/pipeline/executor.h>
#include <fcntl.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/uio.h>
#include <unistd.h>

#include <cerrno>

namespace dftracer::utils::io {

ThreadPoolFileOps::ThreadPoolFileOps(Executor& executor, std::size_t pool_size,
                                     unsigned batch_threshold)
    : executor_(executor), pool_(pool_size, batch_threshold) {}

IoAwaitable ThreadPoolFileOps::make_request(IoOp op, int fd, void* buf,
                                            std::size_t len, off_t offset,
                                            const char* path, int flags,
                                            mode_t mode, Executor* executor,
                                            IoThreadPool* pool) {
    auto* req = new IoRequest{};
    req->submit = &ThreadPoolFileOps::submit_to_pool;
    req->op = op;
    req->fd = fd;
    req->buf = buf;
    req->len = len;
    req->offset = offset;
    req->path = path;
    req->flags = flags;
    req->mode = mode;
    req->executor = executor;
    req->pool = pool;

    IoAwaitable awaitable;
    // req->awaitable will be updated in submit_to_pool with the
    // stable address from await_suspend (the IoAwaitable may move
    // before the coroutine frame is allocated).
    req->awaitable = nullptr;
    awaitable.submit_ctx_ = req;
    return awaitable;
}

IoAwaitable ThreadPoolFileOps::submit_read(int fd, void* buf, std::size_t len) {
    return make_request(IoOp::READ, fd, buf, len, 0, nullptr, 0, 0, &executor_,
                        &pool_);
}

IoAwaitable ThreadPoolFileOps::submit_write(int fd, const void* buf,
                                            std::size_t len) {
    return make_request(IoOp::WRITE, fd, const_cast<void*>(buf), len, 0,
                        nullptr, 0, 0, &executor_, &pool_);
}

IoAwaitable ThreadPoolFileOps::submit_pread(int fd, void* buf, std::size_t len,
                                            off_t offset) {
    return make_request(IoOp::PREAD, fd, buf, len, offset, nullptr, 0, 0,
                        &executor_, &pool_);
}

void ThreadPoolFileOps::submit_pread_callback(int fd, void* buf,
                                              std::size_t len, off_t offset,
                                              IoCompletionFn completion,
                                              void* context) {
    auto* req = new IoRequest{};
    req->op = IoOp::PREAD;
    req->fd = fd;
    req->buf = buf;
    req->len = len;
    req->offset = offset;
    req->completion = completion;
    req->completion_ctx = context;
    req->pool = &pool_;
    pool_.submit([req] { execute_request(req); });
}

IoAwaitable ThreadPoolFileOps::submit_pwrite(int fd, const void* buf,
                                             std::size_t len, off_t offset) {
    return make_request(IoOp::PWRITE, fd, const_cast<void*>(buf), len, offset,
                        nullptr, 0, 0, &executor_, &pool_);
}

IoAwaitable ThreadPoolFileOps::submit_open(const char* path, int flags,
                                           mode_t mode) {
    return make_request(IoOp::OPEN, -1, nullptr, 0, 0, path, flags, mode,
                        &executor_, &pool_);
}

IoAwaitable ThreadPoolFileOps::submit_close(int fd) {
    return make_request(IoOp::CLOSE, fd, nullptr, 0, 0, nullptr, 0, 0,
                        &executor_, &pool_);
}

IoAwaitable ThreadPoolFileOps::submit_fsync(int fd) {
    return make_request(IoOp::FSYNC, fd, nullptr, 0, 0, nullptr, 0, 0,
                        &executor_, &pool_);
}

IoAwaitable ThreadPoolFileOps::submit_ftruncate(int fd, off_t length) {
    return make_request(IoOp::FTRUNCATE, fd, nullptr, 0, length, nullptr, 0, 0,
                        &executor_, &pool_);
}

IoAwaitable ThreadPoolFileOps::submit_fstat(int fd, struct stat* buf) {
    auto req_awaitable = make_request(IoOp::FSTAT, fd, nullptr, 0, 0, nullptr,
                                      0, 0, &executor_, &pool_);
    auto* req = static_cast<IoRequest*>(req_awaitable.submit_ctx_);
    req->stat_buf = buf;
    return req_awaitable;
}

IoAwaitable ThreadPoolFileOps::submit_accept(int listen_fd,
                                             struct sockaddr* addr,
                                             socklen_t* addrlen) {
    auto req_awaitable = make_request(IoOp::ACCEPT, listen_fd, nullptr, 0, 0,
                                      nullptr, 0, 0, &executor_, &pool_);
    auto* req = static_cast<IoRequest*>(req_awaitable.submit_ctx_);
    req->addr = addr;
    req->addrlen = addrlen;
    return req_awaitable;
}

IoAwaitable ThreadPoolFileOps::submit_recv(int fd, void* buf, std::size_t len,
                                           int flags) {
    auto req_awaitable = make_request(IoOp::RECV, fd, buf, len, 0, nullptr, 0,
                                      0, &executor_, &pool_);
    auto* req = static_cast<IoRequest*>(req_awaitable.submit_ctx_);
    req->msg_flags = flags;
    return req_awaitable;
}

IoAwaitable ThreadPoolFileOps::submit_send(int fd, const void* buf,
                                           std::size_t len, int flags) {
    auto req_awaitable =
        make_request(IoOp::SEND, fd, const_cast<void*>(buf), len, 0, nullptr, 0,
                     0, &executor_, &pool_);
    auto* req = static_cast<IoRequest*>(req_awaitable.submit_ctx_);
    req->msg_flags = flags;
    return req_awaitable;
}

IoAwaitable ThreadPoolFileOps::submit_readv(int fd, const struct iovec* iov,
                                            int iovcnt) {
    auto req_awaitable = make_request(IoOp::READV, fd, nullptr, 0, 0, nullptr,
                                      0, 0, &executor_, &pool_);
    auto* req = static_cast<IoRequest*>(req_awaitable.submit_ctx_);
    req->iov = iov;
    req->iovcnt = iovcnt;
    return req_awaitable;
}

IoAwaitable ThreadPoolFileOps::submit_writev(int fd, const struct iovec* iov,
                                             int iovcnt) {
    auto req_awaitable = make_request(IoOp::WRITEV, fd, nullptr, 0, 0, nullptr,
                                      0, 0, &executor_, &pool_);
    auto* req = static_cast<IoRequest*>(req_awaitable.submit_ctx_);
    req->iov = iov;
    req->iovcnt = iovcnt;
    return req_awaitable;
}

IoAwaitable ThreadPoolFileOps::submit_preadv(int fd, const struct iovec* iov,
                                             int iovcnt, off_t offset) {
    auto req_awaitable = make_request(IoOp::PREADV, fd, nullptr, 0, offset,
                                      nullptr, 0, 0, &executor_, &pool_);
    auto* req = static_cast<IoRequest*>(req_awaitable.submit_ctx_);
    req->iov = iov;
    req->iovcnt = iovcnt;
    return req_awaitable;
}

IoAwaitable ThreadPoolFileOps::submit_pwritev(int fd, const struct iovec* iov,
                                              int iovcnt, off_t offset) {
    auto req_awaitable = make_request(IoOp::PWRITEV, fd, nullptr, 0, offset,
                                      nullptr, 0, 0, &executor_, &pool_);
    auto* req = static_cast<IoRequest*>(req_awaitable.submit_ctx_);
    req->iov = iov;
    req->iovcnt = iovcnt;
    return req_awaitable;
}

IoAwaitable ThreadPoolFileOps::submit_lseek(int fd, off_t offset, int whence) {
    auto req_awaitable = make_request(IoOp::LSEEK, fd, nullptr, 0, offset,
                                      nullptr, 0, 0, &executor_, &pool_);
    auto* req = static_cast<IoRequest*>(req_awaitable.submit_ctx_);
    req->whence = whence;
    return req_awaitable;
}

IoAwaitable ThreadPoolFileOps::submit_sendfile(int out_fd, int in_fd,
                                               off_t offset,
                                               std::size_t count) {
    auto req_awaitable =
        make_request(IoOp::SENDFILE, in_fd, nullptr, count, offset, nullptr, 0,
                     0, &executor_, &pool_);
    auto* req = static_cast<IoRequest*>(req_awaitable.submit_ctx_);
    req->dest_fd = out_fd;
    return req_awaitable;
}

void ThreadPoolFileOps::submit_to_pool(SubmitContext* ctx,
                                       IoAwaitable* awaitable) {
    auto* req = static_cast<IoRequest*>(ctx);
    // Update the awaitable pointer -- await_suspend passes the real,
    // stable address of the IoAwaitable in the coroutine frame.
    req->awaitable = awaitable;
    req->pool->submit([req] { execute_request(req); });
}

void ThreadPoolFileOps::execute_request(IoRequest* req) {
    ssize_t result = 0;
    switch (req->op) {
        case IoOp::READ:
            result = ::read(req->fd, req->buf, req->len);
            break;
        case IoOp::WRITE:
            result = ::write(req->fd, req->buf, req->len);
            break;
        case IoOp::PREAD:
            result = ::pread(req->fd, req->buf, req->len, req->offset);
            break;
        case IoOp::PWRITE:
            result = ::pwrite(req->fd, req->buf, req->len, req->offset);
            break;
        case IoOp::OPEN:
            result = ::open(req->path, req->flags, req->mode);
            break;
        case IoOp::CLOSE:
            result = ::close(req->fd);
            break;
        case IoOp::FSYNC:
            result = ::fsync(req->fd);
            break;
        case IoOp::FTRUNCATE:
            result = ::ftruncate(req->fd, req->offset);
            break;
        case IoOp::FSTAT:
            result = ::fstat(req->fd, req->stat_buf);
            break;
        case IoOp::ACCEPT:
#ifdef __linux__
            result = ::accept4(req->fd, req->addr, req->addrlen,
                               SOCK_NONBLOCK | SOCK_CLOEXEC);
#else
            result = ::accept(req->fd, req->addr, req->addrlen);
            if (result >= 0) {
                int fd = static_cast<int>(result);
                int fl = ::fcntl(fd, F_GETFL, 0);
                ::fcntl(fd, F_SETFL, fl | O_NONBLOCK);
                ::fcntl(fd, F_SETFD, FD_CLOEXEC);
            }
#endif
            break;
        case IoOp::RECV:
            result = ::recv(req->fd, req->buf, req->len, req->msg_flags);
            break;
        case IoOp::SEND:
            result = ::send(req->fd, req->buf, req->len, req->msg_flags);
            break;
        case IoOp::READV:
            result = ::readv(req->fd, req->iov, req->iovcnt);
            break;
        case IoOp::WRITEV:
            result = ::writev(req->fd, req->iov, req->iovcnt);
            break;
        case IoOp::PREADV:
            result = ::preadv(req->fd, req->iov, req->iovcnt, req->offset);
            break;
        case IoOp::PWRITEV:
            result = ::pwritev(req->fd, req->iov, req->iovcnt, req->offset);
            break;
        case IoOp::LSEEK:
            result = ::lseek(req->fd, req->offset, req->whence);
            break;
        case IoOp::SENDFILE:
            result =
                platform_sendfile(req->dest_fd, req->fd, req->offset, req->len);
            break;
    }
    if (result < 0) result = -errno;

    if (req->awaitable != nullptr) {
        req->awaitable->result_ = result;
        req->executor->enqueue(req->awaitable->handle_);
    } else if (req->completion != nullptr) {
        req->completion(req->completion_ctx, result);
    }
    delete req;
}

std::size_t ThreadPoolFileOps::poll(int /*timeout_ms*/) {
    return 0;  // Thread pool backends: completions fire via callbacks
}

int ThreadPoolFileOps::flush() { return static_cast<int>(pool_.flush()); }

}  // namespace dftracer::utils::io

1	#include <dftracer/utils/core/io/io_sendfile.h>
2	#include <dftracer/utils/core/io/thread_pool_file_ops.h>
3	#include <dftracer/utils/core/pipeline/executor.h>
4	#include <fcntl.h>
5	#include <sys/socket.h>
6	#include <sys/stat.h>
7	#include <sys/uio.h>
8	#include <unistd.h>
9
10	#include <cerrno>
11
12	namespace dftracer::utils::io {
13
14	ThreadPoolFileOps::ThreadPoolFileOps(Executor& executor, std::size_t pool_size,	570✔
15	unsigned batch_threshold)
16	: executor_(executor), pool_(pool_size, batch_threshold) {}	570!
17
18	IoAwaitable ThreadPoolFileOps::make_request(IoOp op, int fd, void* buf,	4,893✔
19	std::size_t len, off_t offset,
20	const char* path, int flags,
21	mode_t mode, Executor* executor,
22	IoThreadPool* pool) {
23	auto* req = new IoRequest{};	4,893✔
24	req->submit = &ThreadPoolFileOps::submit_to_pool;	4,893✔
25	req->op = op;	4,893✔
26	req->fd = fd;	4,893✔
27	req->buf = buf;	4,893✔
28	req->len = len;	4,893✔
29	req->offset = offset;	4,893✔
30	req->path = path;	4,893✔
31	req->flags = flags;	4,893✔
32	req->mode = mode;	4,893✔
33	req->executor = executor;	4,893✔
34	req->pool = pool;	4,893✔
35
36	IoAwaitable awaitable;	4,893✔
37	// req->awaitable will be updated in submit_to_pool with the
38	// stable address from await_suspend (the IoAwaitable may move
39	// before the coroutine frame is allocated).
40	req->awaitable = nullptr;	4,893✔
41	awaitable.submit_ctx_ = req;	4,893✔
42	return awaitable;	4,893✔
43	}
44
45	IoAwaitable ThreadPoolFileOps::submit_read(int fd, void* buf, std::size_t len) {	626✔
46	return make_request(IoOp::READ, fd, buf, len, 0, nullptr, 0, 0, &executor_,	1,252✔
47	&pool_);	626✔
48	}
49
50	IoAwaitable ThreadPoolFileOps::submit_write(int fd, const void* buf,	649✔
51	std::size_t len) {
52	return make_request(IoOp::WRITE, fd, const_cast<void*>(buf), len, 0,	1,298✔
53	nullptr, 0, 0, &executor_, &pool_);	649✔
54	}
55
56	IoAwaitable ThreadPoolFileOps::submit_pread(int fd, void* buf, std::size_t len,	1,401✔
57	off_t offset) {
58	return make_request(IoOp::PREAD, fd, buf, len, offset, nullptr, 0, 0,	2,802✔
59	&executor_, &pool_);	1,401✔
60	}
61
62	void ThreadPoolFileOps::submit_pread_callback(int fd, void* buf,	×
63	std::size_t len, off_t offset,
64	IoCompletionFn completion,
65	void* context) {
66	auto* req = new IoRequest{};	×
67	req->op = IoOp::PREAD;	×
68	req->fd = fd;	×
69	req->buf = buf;	×
70	req->len = len;	×
71	req->offset = offset;	×
72	req->completion = completion;	×
73	req->completion_ctx = context;	×
74	req->pool = &pool_;	×
75	pool_.submit([req] { execute_request(req); });	×
76	}	×
77
78	IoAwaitable ThreadPoolFileOps::submit_pwrite(int fd, const void* buf,	176✔
79	std::size_t len, off_t offset) {
80	return make_request(IoOp::PWRITE, fd, const_cast<void*>(buf), len, offset,	352✔
81	nullptr, 0, 0, &executor_, &pool_);	176✔
82	}
83
84	IoAwaitable ThreadPoolFileOps::submit_open(const char* path, int flags,	931✔
85	mode_t mode) {
86	return make_request(IoOp::OPEN, -1, nullptr, 0, 0, path, flags, mode,	1,862✔
87	&executor_, &pool_);	931✔
88	}
89
90	IoAwaitable ThreadPoolFileOps::submit_close(int fd) {	780✔
91	return make_request(IoOp::CLOSE, fd, nullptr, 0, 0, nullptr, 0, 0,	1,560✔
92	&executor_, &pool_);	780✔
93	}
94
95	IoAwaitable ThreadPoolFileOps::submit_fsync(int fd) {	56✔
96	return make_request(IoOp::FSYNC, fd, nullptr, 0, 0, nullptr, 0, 0,	112✔
97	&executor_, &pool_);	56✔
98	}
99
100	IoAwaitable ThreadPoolFileOps::submit_ftruncate(int fd, off_t length) {	×
101	return make_request(IoOp::FTRUNCATE, fd, nullptr, 0, length, nullptr, 0, 0,	×
102	&executor_, &pool_);	×
103	}
104
105	IoAwaitable ThreadPoolFileOps::submit_fstat(int fd, struct stat* buf) {	×
106	auto req_awaitable = make_request(IoOp::FSTAT, fd, nullptr, 0, 0, nullptr,	×
107	0, 0, &executor_, &pool_);	×
108	auto* req = static_cast<IoRequest*>(req_awaitable.submit_ctx_);	×
109	req->stat_buf = buf;	×
110	return req_awaitable;	×
111	}
112
113	IoAwaitable ThreadPoolFileOps::submit_accept(int listen_fd,	71✔
114	struct sockaddr* addr,
115	socklen_t* addrlen) {
116	auto req_awaitable = make_request(IoOp::ACCEPT, listen_fd, nullptr, 0, 0,	142✔
117	nullptr, 0, 0, &executor_, &pool_);	71✔
118	auto* req = static_cast<IoRequest*>(req_awaitable.submit_ctx_);	71✔
119	req->addr = addr;	71✔
120	req->addrlen = addrlen;	71✔
121	return req_awaitable;	71✔
122	}
123
124	IoAwaitable ThreadPoolFileOps::submit_recv(int fd, void* buf, std::size_t len,	64✔
125	int flags) {
126	auto req_awaitable = make_request(IoOp::RECV, fd, buf, len, 0, nullptr, 0,	128✔
127	0, &executor_, &pool_);	64✔
128	auto* req = static_cast<IoRequest*>(req_awaitable.submit_ctx_);	64✔
129	req->msg_flags = flags;	64✔
130	return req_awaitable;	64✔
131	}
132
133	IoAwaitable ThreadPoolFileOps::submit_send(int fd, const void* buf,	58✔
134	std::size_t len, int flags) {
135	auto req_awaitable =
136	make_request(IoOp::SEND, fd, const_cast<void*>(buf), len, 0, nullptr, 0,	116✔
137	0, &executor_, &pool_);	58✔
138	auto* req = static_cast<IoRequest*>(req_awaitable.submit_ctx_);	58✔
139	req->msg_flags = flags;	58✔
140	return req_awaitable;	58✔
141	}
142
143	IoAwaitable ThreadPoolFileOps::submit_readv(int fd, const struct iovec* iov,	×
144	int iovcnt) {
145	auto req_awaitable = make_request(IoOp::READV, fd, nullptr, 0, 0, nullptr,	×
146	0, 0, &executor_, &pool_);	×
147	auto* req = static_cast<IoRequest*>(req_awaitable.submit_ctx_);	×
148	req->iov = iov;	×
149	req->iovcnt = iovcnt;	×
150	return req_awaitable;	×
151	}
152
153	IoAwaitable ThreadPoolFileOps::submit_writev(int fd, const struct iovec* iov,	7✔
154	int iovcnt) {
155	auto req_awaitable = make_request(IoOp::WRITEV, fd, nullptr, 0, 0, nullptr,	14✔
156	0, 0, &executor_, &pool_);	7✔
157	auto* req = static_cast<IoRequest*>(req_awaitable.submit_ctx_);	7✔
158	req->iov = iov;	7✔
159	req->iovcnt = iovcnt;	7✔
160	return req_awaitable;	7✔
161	}
162
163	IoAwaitable ThreadPoolFileOps::submit_preadv(int fd, const struct iovec* iov,	1✔
164	int iovcnt, off_t offset) {
165	auto req_awaitable = make_request(IoOp::PREADV, fd, nullptr, 0, offset,	2✔
166	nullptr, 0, 0, &executor_, &pool_);	1✔
167	auto* req = static_cast<IoRequest*>(req_awaitable.submit_ctx_);	1✔
168	req->iov = iov;	1✔
169	req->iovcnt = iovcnt;	1✔
170	return req_awaitable;	1✔
171	}
172
173	IoAwaitable ThreadPoolFileOps::submit_pwritev(int fd, const struct iovec* iov,	71✔
174	int iovcnt, off_t offset) {
175	auto req_awaitable = make_request(IoOp::PWRITEV, fd, nullptr, 0, offset,	142✔
176	nullptr, 0, 0, &executor_, &pool_);	71✔
177	auto* req = static_cast<IoRequest*>(req_awaitable.submit_ctx_);	71✔
178	req->iov = iov;	71✔
179	req->iovcnt = iovcnt;	71✔
180	return req_awaitable;	71✔
181	}
182
183	IoAwaitable ThreadPoolFileOps::submit_lseek(int fd, off_t offset, int whence) {	1✔
184	auto req_awaitable = make_request(IoOp::LSEEK, fd, nullptr, 0, offset,	2✔
185	nullptr, 0, 0, &executor_, &pool_);	1✔
186	auto* req = static_cast<IoRequest*>(req_awaitable.submit_ctx_);	1✔
187	req->whence = whence;	1✔
188	return req_awaitable;	1✔
189	}
190
191	IoAwaitable ThreadPoolFileOps::submit_sendfile(int out_fd, int in_fd,	1✔
192	off_t offset,
193	std::size_t count) {
194	auto req_awaitable =
195	make_request(IoOp::SENDFILE, in_fd, nullptr, count, offset, nullptr, 0,	2✔
196	0, &executor_, &pool_);	1✔
197	auto* req = static_cast<IoRequest*>(req_awaitable.submit_ctx_);	1✔
198	req->dest_fd = out_fd;	1✔
199	return req_awaitable;	1✔
200	}
201
202	void ThreadPoolFileOps::submit_to_pool(SubmitContext* ctx,	4,892✔
203	IoAwaitable* awaitable) {
204	auto* req = static_cast<IoRequest*>(ctx);	4,892✔
205	// Update the awaitable pointer -- await_suspend passes the real,
206	// stable address of the IoAwaitable in the coroutine frame.
207	req->awaitable = awaitable;	4,892✔
208	req->pool->submit([req] { execute_request(req); });	9,785!
209	}	4,892✔
210
211	void ThreadPoolFileOps::execute_request(IoRequest* req) {	4,893✔
212	ssize_t result = 0;	4,893✔
213	switch (req->op) {	4,893!
214	case IoOp::READ:
215	result = ::read(req->fd, req->buf, req->len);	626✔
216	break;	626✔
217	case IoOp::WRITE:
218	result = ::write(req->fd, req->buf, req->len);	649✔
219	break;	649✔
220	case IoOp::PREAD:
221	result = ::pread(req->fd, req->buf, req->len, req->offset);	1,401✔
222	break;	1,401✔
223	case IoOp::PWRITE:
224	result = ::pwrite(req->fd, req->buf, req->len, req->offset);	176✔
225	break;	176✔
226	case IoOp::OPEN:
227	result = ::open(req->path, req->flags, req->mode);	931✔
228	break;	931✔
229	case IoOp::CLOSE:
230	result = ::close(req->fd);	780✔
231	break;	780✔
232	case IoOp::FSYNC:
233	result = ::fsync(req->fd);	56✔
234	break;	56✔
235	case IoOp::FTRUNCATE:
236	result = ::ftruncate(req->fd, req->offset);	×
237	break;	×
238	case IoOp::FSTAT:
239	result = ::fstat(req->fd, req->stat_buf);	×
240	break;	×
241	case IoOp::ACCEPT:
242	#ifdef __linux__
243	result = ::accept4(req->fd, req->addr, req->addrlen,
244	SOCK_NONBLOCK \| SOCK_CLOEXEC);
245	#else
246	result = ::accept(req->fd, req->addr, req->addrlen);	71✔
247	if (result >= 0) {	71✔
248	int fd = static_cast<int>(result);	64✔
249	int fl = ::fcntl(fd, F_GETFL, 0);	64✔
250	::fcntl(fd, F_SETFL, fl \| O_NONBLOCK);	64✔
251	::fcntl(fd, F_SETFD, FD_CLOEXEC);	64✔
252	}	64✔
253	#endif
254	break;	71✔
255	case IoOp::RECV:
256	result = ::recv(req->fd, req->buf, req->len, req->msg_flags);	64✔
257	break;	64✔
258	case IoOp::SEND:
259	result = ::send(req->fd, req->buf, req->len, req->msg_flags);	58✔
260	break;	58✔
261	case IoOp::READV:
262	result = ::readv(req->fd, req->iov, req->iovcnt);	×
263	break;	×
264	case IoOp::WRITEV:
265	result = ::writev(req->fd, req->iov, req->iovcnt);	7✔
266	break;	7✔
267	case IoOp::PREADV:
268	result = ::preadv(req->fd, req->iov, req->iovcnt, req->offset);	1✔
269	break;	1✔
270	case IoOp::PWRITEV:
271	result = ::pwritev(req->fd, req->iov, req->iovcnt, req->offset);	71✔
272	break;	71✔
273	case IoOp::LSEEK:
274	result = ::lseek(req->fd, req->offset, req->whence);	1✔
275	break;	1✔
276	case IoOp::SENDFILE:
277	result =	1✔
278	platform_sendfile(req->dest_fd, req->fd, req->offset, req->len);	1✔
279	break;	1✔
280	}
281	if (result < 0) result = -errno;	4,893✔
282
283	if (req->awaitable != nullptr) {	4,893!
284	req->awaitable->result_ = result;	4,893✔
285	req->executor->enqueue(req->awaitable->handle_);	4,893✔
286	} else if (req->completion != nullptr) {	4,893!
287	req->completion(req->completion_ctx, result);	×
UNCOV 288	}	×
289	delete req;	4,893!
290	}	4,893✔
291
292	std::size_t ThreadPoolFileOps::poll(int /timeout_ms/) {	24,130✔
293	return 0; // Thread pool backends: completions fire via callbacks	24,130✔
294	}
295
296	int ThreadPoolFileOps::flush() { return static_cast<int>(pool_.flush()); }	24,083✔
297
298	} // namespace dftracer::utils::io

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