Skip to content

Commit 18f657e

Browse files
Cra3zCra3z
committed
refactor: refactor asyncio
1 parent a319ffc commit 18f657e

21 files changed

Lines changed: 1847 additions & 118 deletions

‎include/coio/asyncio/epoll_context.h‎

Lines changed: 13 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -102,6 +102,7 @@ namespace coio {
102102
io_object(io_object&& other) noexcept :
103103
ctx_(other.ctx_),
104104
fd_(std::exchange(other.fd_, -1)),
105+
stream_oriented_(std::exchange(other.stream_oriented_, false)),
105106
data_(std::exchange(other.data_, {})) {}
106107

107108
~io_object();
@@ -114,6 +115,7 @@ namespace coio {
114115
auto swap(io_object& other) noexcept -> void {
115116
std::ranges::swap(ctx_, other.ctx_);
116117
std::ranges::swap(fd_, other.fd_);
118+
std::ranges::swap(stream_oriented_, other.stream_oriented_);
117119
std::ranges::swap(data_, other.data_);
118120
}
119121

@@ -147,6 +149,11 @@ namespace coio {
147149
[[nodiscard]]
148150
auto send_to(std::span<const std::byte> buffer, const endpoint& dest) -> std::size_t;
149151

152+
auto connect(const endpoint& peer) -> void;
153+
154+
[[nodiscard]]
155+
auto accept() -> detail::socket_native_handle_type;
156+
150157
auto read_some(std::span<std::byte> buffer) -> std::size_t;
151158

152159
auto write_some(std::span<const std::byte> buffer) -> std::size_t;
@@ -173,7 +180,7 @@ namespace coio {
173180
public:
174181
[[nodiscard]]
175182
COIO_ALWAYS_INLINE auto async_receive(std::span<std::byte> buffer) noexcept {
176-
return async_initiate<detail::receive_tag>(buffer);
183+
return async_initiate<detail::receive_tag>(buffer, stream_oriented_);
177184
}
178185

179186
[[nodiscard]]
@@ -224,6 +231,7 @@ namespace coio {
224231
private:
225232
epoll_context* ctx_;
226233
int fd_ = -1;
234+
bool stream_oriented_ = false;
227235
per_fd_data* data_ = nullptr;
228236
};
229237

@@ -409,9 +417,10 @@ namespace coio {
409417
template<>
410418
class epoll_state_base_for<receive_tag> : public epoll_node_for<receive_tag> {
411419
public:
412-
epoll_state_base_for(int fd, epoll_context& context, epoll_context::per_fd_data* data, std::span<std::byte> buffer) noexcept :
420+
epoll_state_base_for(int fd, epoll_context& context, epoll_context::per_fd_data* data, std::span<std::byte> buffer, bool stream_oriented) noexcept :
413421
epoll_node_for(fd, context, data),
414-
buffer_(buffer) {}
422+
buffer_(buffer),
423+
stream_oriented_(stream_oriented) {}
415424

416425
protected:
417426
auto do_start() noexcept -> start_result;
@@ -423,6 +432,7 @@ namespace coio {
423432

424433
private:
425434
std::span<std::byte> buffer_;
435+
bool stream_oriented_;
426436
};
427437

428438
/// async_send

‎include/coio/asyncio/file.h‎

Lines changed: 2 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -201,19 +201,7 @@ namespace coio {
201201
*/
202202
[[nodiscard]]
203203
COIO_ALWAYS_INLINE auto async_read_some(std::span<std::byte> buffer) {
204-
return let_value(
205-
this->impl_.async_read_some(buffer),
206-
[total = buffer.size()](std::size_t bytes_transferred) noexcept {
207-
async_result<execution::set_value_t(std::size_t), execution::set_error_t(std::error_code)> result;
208-
if (bytes_transferred == 0 and total > 0) [[unlikely]] {
209-
result.set_error(error::eof);
210-
}
211-
else {
212-
result.set_value(bytes_transferred);
213-
}
214-
return result;
215-
}
216-
);
204+
return this->impl_.async_read_some(buffer);
217205
}
218206

219207
/**
@@ -272,19 +260,7 @@ namespace coio {
272260
* \return a sender of `std::size_t` representing the number of bytes read.
273261
*/
274262
COIO_ALWAYS_INLINE auto async_read_some_at(std::size_t offset, std::span<std::byte> buffer) {
275-
return let_value(
276-
this->impl_.async_read_some_at(offset, buffer),
277-
[total = buffer.size()](std::size_t bytes_transferred) noexcept {
278-
async_result<execution::set_value_t(std::size_t), execution::set_error_t(std::error_code)> result;
279-
if (bytes_transferred == 0 and total > 0) [[unlikely]] {
280-
result.set_error(error::eof);
281-
}
282-
else {
283-
result.set_value(bytes_transferred);
284-
}
285-
return result;
286-
}
287-
);
263+
return this->impl_.async_read_some_at(offset, buffer);
288264
}
289265

290266
/**

‎include/coio/asyncio/iocp_context.h‎

Lines changed: 20 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -234,7 +234,9 @@ namespace coio {
234234
public:
235235
socket_object(iocp_context& ctx, detail::socket_native_handle_type sock);
236236

237-
socket_object(socket_object&& other) noexcept : io_object(std::move(other)) {}
237+
socket_object(socket_object&& other) noexcept :
238+
io_object(std::move(other)),
239+
stream_oriented_(std::exchange(other.stream_oriented_, false)) {}
238240

239241
~socket_object();
240242

@@ -245,6 +247,7 @@ namespace coio {
245247

246248
auto swap(socket_object& other) noexcept -> void {
247249
swap_handle(other);
250+
std::ranges::swap(stream_oriented_, other.stream_oriented_);
248251
}
249252

250253
friend auto swap(socket_object& lhs, socket_object& rhs) noexcept -> void {
@@ -270,14 +273,19 @@ namespace coio {
270273
[[nodiscard]]
271274
auto send_to(std::span<const std::byte> buffer, const endpoint& dest) -> std::size_t;
272275

276+
auto connect(const endpoint& peer) -> void;
277+
278+
[[nodiscard]]
279+
auto accept() -> detail::socket_native_handle_type;
280+
273281
[[nodiscard]]
274282
COIO_ALWAYS_INLINE auto async_receive(std::span<std::byte> buffer) noexcept {
275-
return async_initiate<detail::receive_tag>(buffer);
283+
return async_initiate<detail::receive_tag>(buffer, stream_oriented_);
276284
}
277285

278286
[[nodiscard]]
279287
COIO_ALWAYS_INLINE auto async_send(std::span<const std::byte> buffer) noexcept {
280-
return async_initiate<detail::send_tag>(buffer);
288+
return async_initiate<detail::send_tag>(buffer, stream_oriented_);
281289
}
282290

283291
[[nodiscard]]
@@ -299,6 +307,9 @@ namespace coio {
299307
COIO_ALWAYS_INLINE auto async_connect(const endpoint& peer) noexcept {
300308
return async_initiate<detail::connect_tag>(peer);
301309
}
310+
311+
private:
312+
bool stream_oriented_;
302313
};
303314

304315
public:
@@ -379,8 +390,8 @@ namespace coio {
379390
template<>
380391
class iocp_state_base_for<receive_tag> : public iocp_context::iocp_node {
381392
public:
382-
iocp_state_base_for(::HANDLE handle_, iocp_context& ctx, std::span<std::byte> buffer) noexcept
383-
: iocp_node(ctx, handle_), buffer_(buffer) {}
393+
iocp_state_base_for(::HANDLE handle_, iocp_context& ctx, std::span<std::byte> buffer, bool stream_oriented_) noexcept
394+
: iocp_node(ctx, handle_), buffer_(buffer), stream_oriented_(stream_oriented_) {}
384395

385396
protected:
386397
auto do_start() noexcept -> start_result;
@@ -392,14 +403,15 @@ namespace coio {
392403

393404
private:
394405
std::span<std::byte> buffer_;
406+
bool stream_oriented_;
395407
};
396408

397409
/// async_send
398410
template<>
399411
class iocp_state_base_for<send_tag> : public iocp_context::iocp_node {
400412
public:
401-
iocp_state_base_for(::HANDLE handle_, iocp_context& ctx, std::span<const std::byte> buffer) noexcept
402-
: iocp_node(ctx, handle_), buffer_(buffer) {}
413+
iocp_state_base_for(::HANDLE handle_, iocp_context& ctx, std::span<const std::byte> buffer, bool stream_oriented_) noexcept
414+
: iocp_node(ctx, handle_), buffer_(buffer), stream_oriented_(stream_oriented_) {}
403415

404416
protected:
405417
auto do_start() noexcept -> start_result;
@@ -411,6 +423,7 @@ namespace coio {
411423

412424
private:
413425
std::span<const std::byte> buffer_;
426+
bool stream_oriented_;
414427
};
415428

416429
/// async_receive_from

‎include/coio/asyncio/uring_context.h‎

Lines changed: 41 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -53,13 +53,14 @@ namespace coio {
5353

5454
class io_object {
5555
public:
56-
io_object(uring_context& ctx, int fd) noexcept : ctx_(&ctx), fd_(fd) {}
56+
io_object(uring_context& ctx, int fd);
5757

5858
io_object(const io_object&) = delete;
5959

6060
io_object(io_object&& other) noexcept :
6161
ctx_(other.ctx_),
62-
fd_(std::exchange(other.fd_, -1)) {}
62+
fd_(std::exchange(other.fd_, -1)),
63+
stream_oriented_(std::exchange(other.stream_oriented_, false)) {}
6364

6465
~io_object();
6566

@@ -71,6 +72,7 @@ namespace coio {
7172
auto swap(io_object& other) noexcept -> void {
7273
std::ranges::swap(ctx_, other.ctx_);
7374
std::ranges::swap(fd_, other.fd_);
75+
std::ranges::swap(stream_oriented_, other.stream_oriented_);
7476
}
7577

7678
friend auto swap(io_object& lhs, io_object& rhs) noexcept -> void {
@@ -104,6 +106,11 @@ namespace coio {
104106
[[nodiscard]]
105107
auto send_to(std::span<const std::byte> buffer, const endpoint& dest) -> std::size_t;
106108

109+
auto connect(const endpoint& peer) -> void;
110+
111+
[[nodiscard]]
112+
auto accept() -> detail::socket_native_handle_type;
113+
107114
auto read_some(std::span<std::byte> buffer) -> std::size_t;
108115

109116
auto write_some(std::span<const std::byte> buffer) -> std::size_t;
@@ -130,7 +137,7 @@ namespace coio {
130137
public:
131138
[[nodiscard]]
132139
COIO_ALWAYS_INLINE auto async_receive(std::span<std::byte> buffer) noexcept {
133-
return async_initiate<detail::receive_tag>(buffer);
140+
return async_initiate<detail::receive_tag>(buffer, stream_oriented_);
134141
}
135142

136143
[[nodiscard]]
@@ -181,6 +188,7 @@ namespace coio {
181188
private:
182189
uring_context* ctx_;
183190
int fd_ = -1;
191+
bool stream_oriented_ = false;
184192
};
185193

186194
template<typename Tag, typename... Args>
@@ -243,9 +251,7 @@ namespace coio {
243251
using scheduler_base::scheduler_base;
244252

245253
[[nodiscard]]
246-
COIO_ALWAYS_INLINE auto make_io_object(int fd) const -> io_object {
247-
return io_object{*ctx_, fd};
248-
}
254+
auto make_io_object(int fd) const -> io_object ;
249255
};
250256

251257
template<typename T = void, typename Alloc = void>
@@ -291,14 +297,27 @@ namespace coio {
291297
public:
292298
uring_node_for(int fd, uring_context& context) noexcept : uring_node(context, fd) {}
293299

300+
protected:
294301
auto do_start() noexcept -> start_result {
302+
if (fd == -1) [[unlikely]] {
303+
result.set_error(std::make_error_code(std::errc::bad_file_descriptor));
304+
return start_result::completed;
305+
}
306+
307+
auto derived = static_cast<uring_state_base_for<Tag>*>(this);
308+
if constexpr (requires { derived->try_complete(); }) {
309+
if (derived->try_complete()) {
310+
return start_result::completed;
311+
}
312+
}
313+
295314
std::scoped_lock _{context_.uring_mtx_};
296315
auto sqe = context_.allocate_sqe();
297316
if (sqe == nullptr) {
298317
result.set_error(std::make_error_code(std::errc::no_buffer_space));
299318
return start_result::completed;
300319
}
301-
static_cast<uring_state_base_for<Tag>*>(this)->prepare(sqe);
320+
derived->prepare(sqe);
302321
::io_uring_sqe_set_data(sqe, static_cast<uring_node*>(this));
303322
// TODO: To suppress TSAN false positives, we need to add more TSAN annotations! see https://github.com/axboe/liburing/issues/1514
304323
COIO_TSAN_RELEASE(static_cast<uring_node*>(this));
@@ -355,6 +374,9 @@ namespace coio {
355374

356375
auto prepare(::io_uring_sqe* sqe) noexcept -> void;
357376

377+
private:
378+
auto complete(int cqe_res) noexcept -> void override;
379+
358380
private:
359381
std::span<std::byte> buffer_;
360382
};
@@ -386,6 +408,9 @@ namespace coio {
386408

387409
auto prepare(::io_uring_sqe* sqe) noexcept -> void;
388410

411+
private:
412+
auto complete(int cqe_res) noexcept -> void override;
413+
389414
private:
390415
std::size_t offset_;
391416
std::span<std::byte> buffer_;
@@ -413,14 +438,21 @@ namespace coio {
413438
template<>
414439
class uring_state_base_for<receive_tag> : public uring_node_for<receive_tag> {
415440
public:
416-
uring_state_base_for(int fd, uring_context& context, std::span<std::byte> buffer) noexcept :
441+
uring_state_base_for(int fd, uring_context& context, std::span<std::byte> buffer, bool stream_oriented) noexcept :
417442
uring_node_for(fd, context),
418-
buffer_(buffer) {}
443+
buffer_(buffer),
444+
stream_oriented_(stream_oriented) {}
419445

420446
auto prepare(::io_uring_sqe* sqe) noexcept -> void;
421447

448+
auto try_complete() noexcept -> bool;
449+
450+
private:
451+
auto complete(int cqe_res) noexcept -> void override;
452+
422453
private:
423454
std::span<std::byte> buffer_;
455+
bool stream_oriented_;
424456
};
425457

426458

‎include/coio/net/basic.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@ namespace coio {
2222
#endif
2323
inline constexpr socket_native_handle_type invalid_socket_handle = socket_native_handle_type(-1);
2424

25-
enum class shutdown_type : short {
25+
enum class shutdown_type : unsigned char {
2626
shutdown_send,
2727
shutdown_receive,
2828
shutdown_both,

0 commit comments

Comments
 (0)