async_{read,write} support rw_flags

include/condy/async_operations.hpp

@@ -350,20 +350,21 @@ inline auto async_close(detail::FixedFd fd) {
  * @brief See io_uring_prep_read
  */
 template <FdLike Fd, BufferLike Buffer>
-inline auto async_read(Fd fd, const Buffer &buf, __u64 offset) {
-    auto op = detail::make_op_awaiter(io_uring_prep_read, fd, buf.data(),
-                                      buf.size(), offset);
+inline auto async_read(Fd fd, const Buffer &buf, __u64 offset, int flags = 0) {
+    auto op = detail::make_op_awaiter(detail::prep_read, fd, buf.data(),
+                                      buf.size(), offset, flags);
     return detail::maybe_flag_fixed_fd(std::move(op), fd);
 }
 
 /**
  * @brief See io_uring_prep_read_fixed
  */
 template <FdLike Fd, BufferLike Buffer>
-inline auto async_read(Fd fd, detail::FixedBuffer<Buffer> buf, __u64 offset) {
+inline auto async_read(Fd fd, detail::FixedBuffer<Buffer> buf, __u64 offset,
+                       int flags = 0) {
     auto op =
-        detail::make_op_awaiter(io_uring_prep_read_fixed, fd, buf.value.data(),
-                                buf.value.size(), offset, buf.buf_index);
+        detail::make_op_awaiter(detail::prep_read_fixed, fd, buf.value.data(),
+                                buf.value.size(), offset, flags, buf.buf_index);
     return detail::maybe_flag_fixed_fd(std::move(op), fd);
 }
 
@@ -396,20 +397,21 @@ inline auto async_read_multishot(Fd fd, Buffer &buf, __u64 offset,
  * @brief See io_uring_prep_write
  */
 template <FdLike Fd, BufferLike Buffer>
-inline auto async_write(Fd fd, const Buffer &buf, __u64 offset) {
-    auto op = detail::make_op_awaiter(io_uring_prep_write, fd, buf.data(),
-                                      buf.size(), offset);
+inline auto async_write(Fd fd, const Buffer &buf, __u64 offset, int flags = 0) {
+    auto op = detail::make_op_awaiter(detail::prep_write, fd, buf.data(),
+                                      buf.size(), offset, flags);
     return detail::maybe_flag_fixed_fd(std::move(op), fd);
 }
 
 /**
  * @brief See io_uring_prep_write_fixed
  */
 template <FdLike Fd, BufferLike Buffer>
-inline auto async_write(Fd fd, detail::FixedBuffer<Buffer> buf, __u64 offset) {
+inline auto async_write(Fd fd, detail::FixedBuffer<Buffer> buf, __u64 offset,
+                        int flags = 0) {
     auto op =
-        detail::make_op_awaiter(io_uring_prep_write_fixed, fd, buf.value.data(),
-                                buf.value.size(), offset, buf.buf_index);
+        detail::make_op_awaiter(detail::prep_write_fixed, fd, buf.value.data(),
+                                buf.value.size(), offset, flags, buf.buf_index);
     return detail::maybe_flag_fixed_fd(std::move(op), fd);
 }
 

include/condy/detail/async_operations.hpp

@@ -221,5 +221,31 @@ inline void prep_recv_zc_multishot(io_uring_sqe *sqe, int fd,
 }
 #endif
 
+inline void prep_read(io_uring_sqe *sqe, int fd, void *buf, unsigned nbytes,
+                      __u64 offset, int flags) noexcept {
+    io_uring_prep_read(sqe, fd, buf, nbytes, offset);
+    sqe->rw_flags = flags;
+}
+
+inline void prep_read_fixed(io_uring_sqe *sqe, int fd, void *buf,
+                            unsigned nbytes, __u64 offset, int flags,
+                            int buf_index) noexcept {
+    io_uring_prep_read_fixed(sqe, fd, buf, nbytes, offset, buf_index);
+    sqe->rw_flags = flags;
+}
+
+inline void prep_write(io_uring_sqe *sqe, int fd, const void *buf,
+                       unsigned nbytes, __u64 offset, int flags) noexcept {
+    io_uring_prep_write(sqe, fd, buf, nbytes, offset);
+    sqe->rw_flags = flags;
+}
+
+inline void prep_write_fixed(io_uring_sqe *sqe, int fd, const void *buf,
+                             unsigned nbytes, __u64 offset, int flags,
+                             int buf_index) noexcept {
+    io_uring_prep_write_fixed(sqe, fd, buf, nbytes, offset, buf_index);
+    sqe->rw_flags = flags;
+}
+
 } // namespace detail
 } // namespace condy

tests/test_async_operations.1.cpp

@@ -251,6 +251,26 @@ TEST_CASE("test async_operations - read incr provided buffer") {
 }
 #endif
 
+TEST_CASE("test async_operations - test read - rw_flags") {
+    int pipe_fds[2];
+    REQUIRE(pipe(pipe_fds) == 0);
+
+    int fl = fcntl(pipe_fds[0], F_GETFL, 0);
+    REQUIRE(fl >= 0);
+    REQUIRE(fcntl(pipe_fds[0], F_SETFL, fl | O_NONBLOCK) == 0);
+
+    auto func = [&]() -> condy::Coro<void> {
+        char buf[64] = {0};
+        ssize_t n = co_await condy::async_read(
+            pipe_fds[0], condy::buffer(buf, 64), 0, RWF_NOWAIT);
+        REQUIRE(n == -EAGAIN);
+    };
+    condy::sync_wait(func());
+
+    close(pipe_fds[0]);
+    close(pipe_fds[1]);
+}
+
 #if CONDY_URING_VERSION_GE(2, 8) // >= 2.8
 TEST_CASE("test async_operations - provided buffer queue check - incr") {
     int sv[2];
@@ -723,6 +743,29 @@ TEST_CASE("test async_operations - writev fixed buffer") {
 }
 #endif
 
+TEST_CASE("test async_operations - test write - rw_flags") {
+    int pipe_fds[2];
+    REQUIRE(pipe(pipe_fds) == 0);
+
+    const char *msg = "Hello, condy write dsync!";
+    size_t msg_len = std::strlen(msg);
+
+    auto func = [&]() -> condy::Coro<void> {
+        ssize_t n = co_await condy::async_write(
+            pipe_fds[1], condy::buffer(msg, msg_len), 0, RWF_DSYNC);
+        REQUIRE(n == static_cast<ssize_t>(msg_len));
+    };
+    condy::sync_wait(func());
+
+    char read_buf[64];
+    ssize_t n = ::read(pipe_fds[0], read_buf, sizeof(read_buf));
+    REQUIRE(n == static_cast<ssize_t>(msg_len));
+    REQUIRE(std::memcmp(read_buf, msg, msg_len) == 0);
+
+    close(pipe_fds[0]);
+    close(pipe_fds[1]);
+}
+
 #if CONDY_URING_VERSION_GE(2, 7) // >= 2.7
 TEST_CASE("test async_operations - send provided buffer") {
     int sv[2];