Commit d671caf2 authored by Dan Melnic's avatar Dan Melnic Committed by Facebook GitHub Bot

Add io_uring support for fsync/fdatasync

Summary: Add io_uring support for fsync/fdatasync

Reviewed By: kevin-vigor

Differential Revision: D22344375

fbshipit-source-id: c6e1de2d778568c4abbefa0a732a3cf8cffc7c97
parent 59701670
...@@ -376,7 +376,7 @@ void IoUringBackend::queueRead( ...@@ -376,7 +376,7 @@ void IoUringBackend::queueRead(
buf, nbytes buf, nbytes
}; };
auto* iocb = new ReadIoSqe(this, fd, &iov, offset, std::move(cb)); auto* iocb = new ReadIoSqe(this, fd, &iov, offset, std::move(cb));
iocb->backendCb_ = processReadWriteCB; iocb->backendCb_ = processFileOpCB;
incNumIoCbInUse(); incNumIoCbInUse();
submitList_.push_back(*iocb); submitList_.push_back(*iocb);
...@@ -393,7 +393,7 @@ void IoUringBackend::queueWrite( ...@@ -393,7 +393,7 @@ void IoUringBackend::queueWrite(
const_cast<void*>(buf), nbytes const_cast<void*>(buf), nbytes
}; };
auto* iocb = new WriteIoSqe(this, fd, &iov, offset, std::move(cb)); auto* iocb = new WriteIoSqe(this, fd, &iov, offset, std::move(cb));
iocb->backendCb_ = processReadWriteCB; iocb->backendCb_ = processFileOpCB;
incNumIoCbInUse(); incNumIoCbInUse();
submitList_.push_back(*iocb); submitList_.push_back(*iocb);
...@@ -406,7 +406,7 @@ void IoUringBackend::queueReadv( ...@@ -406,7 +406,7 @@ void IoUringBackend::queueReadv(
off_t offset, off_t offset,
FileOpCallback&& cb) { FileOpCallback&& cb) {
auto* iocb = new ReadvIoSqe(this, fd, iovecs, offset, std::move(cb)); auto* iocb = new ReadvIoSqe(this, fd, iovecs, offset, std::move(cb));
iocb->backendCb_ = processReadWriteCB; iocb->backendCb_ = processFileOpCB;
incNumIoCbInUse(); incNumIoCbInUse();
submitList_.push_back(*iocb); submitList_.push_back(*iocb);
...@@ -419,15 +419,32 @@ void IoUringBackend::queueWritev( ...@@ -419,15 +419,32 @@ void IoUringBackend::queueWritev(
off_t offset, off_t offset,
FileOpCallback&& cb) { FileOpCallback&& cb) {
auto* iocb = new WritevIoSqe(this, fd, iovecs, offset, std::move(cb)); auto* iocb = new WritevIoSqe(this, fd, iovecs, offset, std::move(cb));
iocb->backendCb_ = processReadWriteCB; iocb->backendCb_ = processFileOpCB;
incNumIoCbInUse(); incNumIoCbInUse();
submitList_.push_back(*iocb); submitList_.push_back(*iocb);
numInsertedEvents_++; numInsertedEvents_++;
} }
void IoUringBackend::processReadWrite(IoCb* ioCb, int64_t res) noexcept { void IoUringBackend::queueFsync(int fd, FileOpCallback&& cb) {
auto* ioSqe = reinterpret_cast<ReadWriteIoSqe*>(ioCb); queueFsync(fd, FSyncFlags::FLAGS_FSYNC, std::move(cb));
}
void IoUringBackend::queueFdatasync(int fd, FileOpCallback&& cb) {
queueFsync(fd, FSyncFlags::FLAGS_FDATASYNC, std::move(cb));
}
void IoUringBackend::queueFsync(int fd, FSyncFlags flags, FileOpCallback&& cb) {
auto* iocb = new FSyncIoSqe(this, fd, flags, std::move(cb));
iocb->backendCb_ = processFileOpCB;
incNumIoCbInUse();
submitList_.push_back(*iocb);
numInsertedEvents_++;
}
void IoUringBackend::processFileOp(IoCb* ioCb, int64_t res) noexcept {
auto* ioSqe = reinterpret_cast<FileOpIoSqe*>(ioCb);
// save the res // save the res
ioSqe->res_ = res; ioSqe->res_ = res;
activeEvents_.push_back(*ioCb); activeEvents_.push_back(*ioCb);
......
...@@ -52,9 +52,9 @@ class IoUringBackend : public PollIoBackend { ...@@ -52,9 +52,9 @@ class IoUringBackend : public PollIoBackend {
return fdRegistry_.free(rec); return fdRegistry_.free(rec);
} }
// read/write file operation callback // read/write/fsync/fdatasync file operation callback
// int param is the io_uring_cqe res field // int param is the io_uring_cqe res field
// i.e. the result of the read/write operation // i.e. the result of the file operation
using FileOpCallback = folly::Function<void(int)>; using FileOpCallback = folly::Function<void(int)>;
void queueRead( void queueRead(
...@@ -83,6 +83,14 @@ class IoUringBackend : public PollIoBackend { ...@@ -83,6 +83,14 @@ class IoUringBackend : public PollIoBackend {
off_t offset, off_t offset,
FileOpCallback&& cb); FileOpCallback&& cb);
// there is no ordering between the prev submitted write
// requests and the sync ops
// ordering can be achieved by calling queue*sync from one of
// the prev write callbacks, once all the write operations
// we have to wait for are done
void queueFsync(int fd, FileOpCallback&& cb);
void queueFdatasync(int fd, FileOpCallback&& cb);
protected: protected:
struct FdRegistry { struct FdRegistry {
FdRegistry() = delete; FdRegistry() = delete;
...@@ -205,18 +213,32 @@ class IoUringBackend : public PollIoBackend { ...@@ -205,18 +213,32 @@ class IoUringBackend : public PollIoBackend {
} }
}; };
struct ReadWriteIoSqe : public IoSqe { struct FileOpIoSqe : public IoSqe {
FileOpIoSqe(PollIoBackend* backend, int fd, FileOpCallback&& cb)
: IoSqe(backend, false), fd_(fd), cb_(std::move(cb)) {}
~FileOpIoSqe() override = default;
void processActive() override {
cb_(res_);
}
int fd_{-1};
int res_{-1};
FileOpCallback cb_;
};
struct ReadWriteIoSqe : public FileOpIoSqe {
ReadWriteIoSqe( ReadWriteIoSqe(
PollIoBackend* backend, PollIoBackend* backend,
int fd, int fd,
const struct iovec* iov, const struct iovec* iov,
off_t offset, off_t offset,
FileOpCallback&& cb) FileOpCallback&& cb)
: IoSqe(backend, false), : FileOpIoSqe(backend, fd, std::move(cb)),
fd_(fd),
iov_(iov, iov + 1), iov_(iov, iov + 1),
offset_(offset), offset_(offset) {}
cb_(std::move(cb)) {}
ReadWriteIoSqe( ReadWriteIoSqe(
PollIoBackend* backend, PollIoBackend* backend,
...@@ -224,11 +246,7 @@ class IoUringBackend : public PollIoBackend { ...@@ -224,11 +246,7 @@ class IoUringBackend : public PollIoBackend {
Range<const struct iovec*> iov, Range<const struct iovec*> iov,
off_t offset, off_t offset,
FileOpCallback&& cb) FileOpCallback&& cb)
: IoSqe(backend, false), : FileOpIoSqe(backend, fd, std::move(cb)), iov_(iov), offset_(offset) {}
fd_(fd),
iov_(iov),
offset_(offset),
cb_(std::move(cb)) {}
~ReadWriteIoSqe() override = default; ~ReadWriteIoSqe() override = default;
...@@ -236,13 +254,9 @@ class IoUringBackend : public PollIoBackend { ...@@ -236,13 +254,9 @@ class IoUringBackend : public PollIoBackend {
cb_(res_); cb_(res_);
} }
int fd_{-1};
int res_{-1};
static constexpr size_t kNumInlineIoVec = 4; static constexpr size_t kNumInlineIoVec = 4;
folly::small_vector<struct iovec> iov_; folly::small_vector<struct iovec> iov_;
off_t offset_; off_t offset_;
FileOpCallback cb_;
}; };
struct ReadIoSqe : public ReadWriteIoSqe { struct ReadIoSqe : public ReadWriteIoSqe {
...@@ -292,11 +306,48 @@ class IoUringBackend : public PollIoBackend { ...@@ -292,11 +306,48 @@ class IoUringBackend : public PollIoBackend {
} }
}; };
void processReadWrite(IoCb* ioCb, int64_t res) noexcept; enum class FSyncFlags {
FLAGS_FSYNC = 0,
FLAGS_FDATASYNC = 1,
};
struct FSyncIoSqe : public FileOpIoSqe {
FSyncIoSqe(
PollIoBackend* backend,
int fd,
FSyncFlags flags,
FileOpCallback&& cb)
: FileOpIoSqe(backend, fd, std::move(cb)), flags_(flags) {}
~FSyncIoSqe() override = default;
bool processSubmit(void* entry) override {
struct io_uring_sqe* sqe = reinterpret_cast<struct io_uring_sqe*>(entry);
unsigned int fsyncFlags = 0;
switch (flags_) {
case FSyncFlags::FLAGS_FSYNC:
fsyncFlags = 0;
break;
case FSyncFlags::FLAGS_FDATASYNC:
fsyncFlags = IORING_FSYNC_DATASYNC;
break;
}
::io_uring_prep_fsync(sqe, fd_, fsyncFlags);
::io_uring_sqe_set_data(sqe, this);
return true;
}
FSyncFlags flags_;
};
void queueFsync(int fd, FSyncFlags flags, FileOpCallback&& cb);
void processFileOp(IoCb* ioCb, int64_t res) noexcept;
static void static void processFileOpCB(PollIoBackend* backend, IoCb* ioCb, int64_t res) {
processReadWriteCB(PollIoBackend* backend, IoCb* ioCb, int64_t res) { static_cast<IoUringBackend*>(backend)->processFileOp(ioCb, res);
static_cast<IoUringBackend*>(backend)->processReadWrite(ioCb, res);
} }
PollIoBackend::IoCb* allocNewIoCb(const EventCallback& /*cb*/) override { PollIoBackend::IoCb* allocNewIoCb(const EventCallback& /*cb*/) override {
......
...@@ -896,11 +896,22 @@ TEST(IoUringBackend, FileWriteMany) { ...@@ -896,11 +896,22 @@ TEST(IoUringBackend, FileWriteMany) {
AlignedBuf bigWriteData(kBigBlockSize, 'A'); AlignedBuf bigWriteData(kBigBlockSize, 'A');
bool bFdatasync = false;
for (size_t i = 0; i < kNumBlocks; i++) { for (size_t i = 0; i < kNumBlocks; i++) {
folly::IoUringBackend::FileOpCallback writeCb = [&, i](int res) { folly::IoUringBackend::FileOpCallback writeCb = [&, i](int res) {
CHECK_EQ(res, writeDataVec[i].size()); CHECK_EQ(res, writeDataVec[i].size());
++num; ++num;
if (num == kNumBlocks) {
folly::IoUringBackend::FileOpCallback fdatasyncCb = [&](int res) {
CHECK_EQ(res, 0);
bFdatasync = true;
};
backendPtr->queueFdatasync(fd, std::move(fdatasyncCb));
}
}; };
backendPtr->queueWrite( backendPtr->queueWrite(
fd, fd,
writeDataVec[i].data(), writeDataVec[i].data(),
...@@ -908,8 +919,21 @@ TEST(IoUringBackend, FileWriteMany) { ...@@ -908,8 +919,21 @@ TEST(IoUringBackend, FileWriteMany) {
i * kBlockSize, i * kBlockSize,
std::move(writeCb)); std::move(writeCb));
} }
evb.loop();
EXPECT_EQ(num, kNumBlocks);
EXPECT_EQ(bFdatasync, true);
bool bFsync = false;
folly::IoUringBackend::FileOpCallback bigWriteCb = [&](int res) { folly::IoUringBackend::FileOpCallback bigWriteCb = [&](int res) {
CHECK_EQ(res, bigWriteData.size()); CHECK_EQ(res, bigWriteData.size());
folly::IoUringBackend::FileOpCallback fsyncCb = [&](int res) {
CHECK_EQ(res, 0);
bFsync = true;
};
backendPtr->queueFsync(fd, std::move(fsyncCb));
}; };
backendPtr->queueWrite( backendPtr->queueWrite(
...@@ -917,7 +941,7 @@ TEST(IoUringBackend, FileWriteMany) { ...@@ -917,7 +941,7 @@ TEST(IoUringBackend, FileWriteMany) {
evb.loop(); evb.loop();
EXPECT_EQ(num, kNumBlocks); EXPECT_EQ(bFsync, true);
} }
namespace folly { namespace folly {
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment