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

Expose async read/write functionality

Summary: Expose async read/write functionality

Reviewed By: yfeldblum, kevin-vigor

Differential Revision: D21731597

fbshipit-source-id: 95a0af53c7b85e1bf1a2619586091b6633bc7b8a
parent 8f91f848
......@@ -308,36 +308,45 @@ size_t IoUringBackend::submitList(
CHECK(sqe); // this should not happen
auto* ev = entry->event_->getEvent();
const auto& cb = entry->event_->getCallback();
bool processed = false;
switch (cb.type_) {
case EventCallback::Type::TYPE_NONE:
break;
case EventCallback::Type::TYPE_READ:
if (auto* iov = cb.readCb_->allocateData()) {
processed = true;
entry->prepRead(
sqe, ev->ev_fd, &iov->data_, (ev->ev_events & EV_PERSIST) != 0);
entry->cbData_.set(iov);
}
break;
case EventCallback::Type::TYPE_RECVMSG:
if (auto* msg = cb.recvmsgCb_->allocateData()) {
processed = true;
entry->prepRecvmsg(
sqe, ev->ev_fd, &msg->data_, (ev->ev_events & EV_PERSIST) != 0);
entry->cbData_.set(msg);
}
break;
}
if (ev) {
const auto& cb = entry->event_->getCallback();
bool processed = false;
switch (cb.type_) {
case EventCallback::Type::TYPE_NONE:
processed = entry->processSubmit(sqe);
break;
case EventCallback::Type::TYPE_READ:
if (auto* iov = cb.readCb_->allocateData()) {
processed = true;
entry->prepRead(
sqe,
ev->ev_fd,
&iov->data_,
0,
(ev->ev_events & EV_PERSIST) != 0);
entry->cbData_.set(iov);
}
break;
case EventCallback::Type::TYPE_RECVMSG:
if (auto* msg = cb.recvmsgCb_->allocateData()) {
processed = true;
entry->prepRecvmsg(
sqe, ev->ev_fd, &msg->data_, (ev->ev_events & EV_PERSIST) != 0);
entry->cbData_.set(msg);
}
break;
}
if (!processed) {
entry->cbData_.reset();
entry->prepPollAdd(
sqe,
ev->ev_fd,
getPollFlags(ev->ev_events),
(ev->ev_events & EV_PERSIST) != 0);
if (!processed) {
entry->cbData_.reset();
entry->prepPollAdd(
sqe,
ev->ev_fd,
getPollFlags(ev->ev_events),
(ev->ev_events & EV_PERSIST) != 0);
}
} else {
entry->processSubmit(sqe);
}
i++;
if (ioCbs.empty()) {
......@@ -356,4 +365,73 @@ size_t IoUringBackend::submitList(
return ret;
}
void IoUringBackend::queueRead(
int fd,
void* buf,
unsigned int nbytes,
off_t offset,
FileOpCallback&& cb) {
struct iovec iov {
buf, nbytes
};
auto* iocb = new ReadIoSqe(this, fd, &iov, offset, std::move(cb));
iocb->backendCb_ = processReadWriteCB;
incNumIoCbInUse();
submitList_.push_back(*iocb);
numInsertedEvents_++;
}
void IoUringBackend::queueWrite(
int fd,
const void* buf,
unsigned int nbytes,
off_t offset,
FileOpCallback&& cb) {
struct iovec iov {
const_cast<void*>(buf), nbytes
};
auto* iocb = new WriteIoSqe(this, fd, &iov, offset, std::move(cb));
iocb->backendCb_ = processReadWriteCB;
incNumIoCbInUse();
submitList_.push_back(*iocb);
numInsertedEvents_++;
}
void IoUringBackend::queueReadv(
int fd,
Range<const struct iovec*> iovecs,
off_t offset,
FileOpCallback&& cb) {
auto* iocb = new ReadvIoSqe(this, fd, iovecs, offset, std::move(cb));
iocb->backendCb_ = processReadWriteCB;
incNumIoCbInUse();
submitList_.push_back(*iocb);
numInsertedEvents_++;
}
void IoUringBackend::queueWritev(
int fd,
Range<const struct iovec*> iovecs,
off_t offset,
FileOpCallback&& cb) {
auto* iocb = new WritevIoSqe(this, fd, iovecs, offset, std::move(cb));
iocb->backendCb_ = processReadWriteCB;
incNumIoCbInUse();
submitList_.push_back(*iocb);
numInsertedEvents_++;
}
void IoUringBackend::processReadWrite(IoCb* ioCb, int64_t res) noexcept {
auto* ioSqe = reinterpret_cast<ReadWriteIoSqe*>(ioCb);
// save the res
ioSqe->res_ = res;
activeEvents_.push_back(*ioCb);
numInsertedEvents_--;
}
} // namespace folly
......@@ -20,7 +20,11 @@ extern "C" {
#include <liburing.h>
}
#include <folly/Function.h>
#include <folly/Range.h>
#include <folly/experimental/io/PollIoBackend.h>
#include <folly/small_vector.h>
#include <glog/logging.h>
namespace folly {
......@@ -48,6 +52,37 @@ class IoUringBackend : public PollIoBackend {
return fdRegistry_.free(rec);
}
// read/write file operation callback
// int param is the io_uring_cqe res field
// i.e. the result of the read/write operation
using FileOpCallback = folly::Function<void(int)>;
void queueRead(
int fd,
void* buf,
unsigned int nbytes,
off_t offset,
FileOpCallback&& cb);
void queueWrite(
int fd,
const void* buf,
unsigned int nbytes,
off_t offset,
FileOpCallback&& cb);
void queueReadv(
int fd,
Range<const struct iovec*> iovecs,
off_t offset,
FileOpCallback&& cb);
void queueWritev(
int fd,
Range<const struct iovec*> iovecs,
off_t offset,
FileOpCallback&& cb);
protected:
struct FdRegistry {
FdRegistry() = delete;
......@@ -100,8 +135,12 @@ class IoUringBackend : public PollIoBackend {
::io_uring_sqe_set_data(sqe, this);
}
void prepRead(void* entry, int fd, const struct iovec* iov, bool registerFd)
override {
void prepRead(
void* entry,
int fd,
const struct iovec* iov,
off_t offset,
bool registerFd) override {
CHECK(entry);
struct io_uring_sqe* sqe = reinterpret_cast<struct io_uring_sqe*>(entry);
if (registerFd && !fdRecord_) {
......@@ -110,11 +149,32 @@ class IoUringBackend : public PollIoBackend {
if (fdRecord_) {
::io_uring_prep_read(
sqe, fdRecord_->idx_, iov->iov_base, iov->iov_len, 0 /*offset*/);
sqe, fdRecord_->idx_, iov->iov_base, iov->iov_len, offset);
sqe->flags |= IOSQE_FIXED_FILE;
} else {
::io_uring_prep_read(
sqe, fd, iov->iov_base, iov->iov_len, 0 /*offset*/);
::io_uring_prep_read(sqe, fd, iov->iov_base, iov->iov_len, offset);
}
::io_uring_sqe_set_data(sqe, this);
}
void prepWrite(
void* entry,
int fd,
const struct iovec* iov,
off_t offset,
bool registerFd) override {
CHECK(entry);
struct io_uring_sqe* sqe = reinterpret_cast<struct io_uring_sqe*>(entry);
if (registerFd && !fdRecord_) {
fdRecord_ = backend_->registerFd(fd);
}
if (fdRecord_) {
::io_uring_prep_write(
sqe, fdRecord_->idx_, iov->iov_base, iov->iov_len, offset);
sqe->flags |= IOSQE_FIXED_FILE;
} else {
::io_uring_prep_write(sqe, fd, iov->iov_base, iov->iov_len, offset);
}
::io_uring_sqe_set_data(sqe, this);
}
......@@ -145,6 +205,100 @@ class IoUringBackend : public PollIoBackend {
}
};
struct ReadWriteIoSqe : public IoSqe {
ReadWriteIoSqe(
PollIoBackend* backend,
int fd,
const struct iovec* iov,
off_t offset,
FileOpCallback&& cb)
: IoSqe(backend, false),
fd_(fd),
iov_(iov, iov + 1),
offset_(offset),
cb_(std::move(cb)) {}
ReadWriteIoSqe(
PollIoBackend* backend,
int fd,
Range<const struct iovec*> iov,
off_t offset,
FileOpCallback&& cb)
: IoSqe(backend, false),
fd_(fd),
iov_(iov),
offset_(offset),
cb_(std::move(cb)) {}
~ReadWriteIoSqe() override = default;
void processActive() override {
cb_(res_);
}
int fd_{-1};
int res_{-1};
static constexpr size_t kNumInlineIoVec = 4;
folly::small_vector<struct iovec> iov_;
off_t offset_;
FileOpCallback cb_;
};
struct ReadIoSqe : public ReadWriteIoSqe {
using ReadWriteIoSqe::ReadWriteIoSqe;
~ReadIoSqe() override = default;
bool processSubmit(void* entry) override {
prepRead(entry, fd_, iov_.data(), offset_, false);
return true;
}
};
struct WriteIoSqe : public ReadWriteIoSqe {
using ReadWriteIoSqe::ReadWriteIoSqe;
~WriteIoSqe() override = default;
bool processSubmit(void* entry) override {
prepWrite(entry, fd_, iov_.data(), offset_, false);
return true;
}
};
struct ReadvIoSqe : public ReadWriteIoSqe {
using ReadWriteIoSqe::ReadWriteIoSqe;
~ReadvIoSqe() override = default;
bool processSubmit(void* entry) override {
struct io_uring_sqe* sqe = reinterpret_cast<struct io_uring_sqe*>(entry);
::io_uring_prep_readv(sqe, fd_, iov_.data(), iov_.size(), offset_);
::io_uring_sqe_set_data(sqe, this);
return true;
}
};
struct WritevIoSqe : public ReadWriteIoSqe {
using ReadWriteIoSqe::ReadWriteIoSqe;
~WritevIoSqe() override = default;
bool processSubmit(void* entry) override {
struct io_uring_sqe* sqe = reinterpret_cast<struct io_uring_sqe*>(entry);
::io_uring_prep_writev(sqe, fd_, iov_.data(), iov_.size(), offset_);
::io_uring_sqe_set_data(sqe, this);
return true;
}
};
void processReadWrite(IoCb* ioCb, int64_t res) noexcept;
static void
processReadWriteCB(PollIoBackend* backend, IoCb* ioCb, int64_t res) {
static_cast<IoUringBackend*>(backend)->processReadWrite(ioCb, res);
}
PollIoBackend::IoCb* allocNewIoCb(const EventCallback& /*cb*/) override {
auto* ret = new IoSqe(this, false);
ret->backendCb_ = PollIoBackend::processPollIoCb;
......
......@@ -353,6 +353,7 @@ PollIoBackend::IoCb* PollIoBackend::allocIoCb(const EventCallback& cb) {
}
void PollIoBackend::releaseIoCb(PollIoBackend::IoCb* aioIoCb) {
CHECK_GT(numIoCbInUse_, 0);
numIoCbInUse_--;
aioIoCb->cbData_.releaseData();
// unregister the file descriptor record
......@@ -430,6 +431,8 @@ size_t PollIoBackend::processActiveEvents() {
eb_event_modify_inserted(*event, ioCb);
}
ioCb->useCount_--;
} else {
ioCb->processActive();
}
if (release) {
releaseIoCb(ioCb);
......
......@@ -110,6 +110,12 @@ class PollIoBackend : public EventBaseBackendBase {
: backend_(backend), poolAlloc_(poolAlloc) {}
virtual ~IoCb() = default;
virtual bool processSubmit(void* /*entry*/) {
return false;
}
virtual void processActive() {}
PollIoBackend* backend_;
BackendCb* backendCb_{nullptr};
const bool poolAlloc_;
......@@ -134,6 +140,14 @@ class PollIoBackend : public EventBaseBackendBase {
void* /*entry*/,
int /*fd*/,
const struct iovec* /*iov*/,
off_t /*offset*/,
bool /*registerFd*/) {}
virtual void prepWrite(
void* /*entry*/,
int /*fd*/,
const struct iovec* /*iov*/,
off_t /*offset*/,
bool /*registerFd*/) {}
virtual void prepRecvmsg(
......@@ -343,6 +357,9 @@ class PollIoBackend : public EventBaseBackendBase {
IoCb* FOLLY_NULLABLE allocIoCb(const EventCallback& cb);
void releaseIoCb(IoCb* aioIoCb);
void incNumIoCbInUse() {
numIoCbInUse_++;
}
virtual IoCb* allocNewIoCb(const EventCallback& cb) = 0;
......
......@@ -18,6 +18,7 @@
namespace folly {
namespace test {
namespace async_base_test_lib_detail {
void TestUtil::waitUntilReadable(int fd) {
pollfd pfd;
pfd.fd = fd;
......@@ -44,51 +45,11 @@ folly::Range<folly::AsyncBase::Op**> TestUtil::readerWait(
TestUtil::ManagedBuffer TestUtil::allocateAligned(size_t size) {
void* buf;
int rc = posix_memalign(&buf, kAlign, size);
int rc = posix_memalign(&buf, kODirectAlign, size);
CHECK_EQ(rc, 0) << folly::errnoStr(rc);
return ManagedBuffer(reinterpret_cast<char*>(buf), free);
}
TemporaryFile::TemporaryFile(size_t size)
: path_(folly::fs::temp_directory_path() / folly::fs::unique_path()) {
CHECK_EQ(size % sizeof(uint32_t), 0);
size /= sizeof(uint32_t);
const uint32_t seed = 42;
std::mt19937 rnd(seed);
const size_t bufferSize = 1U << 16;
uint32_t buffer[bufferSize];
FILE* fp = ::fopen(path_.c_str(), "wb");
PCHECK(fp != nullptr);
while (size) {
size_t n = std::min(size, bufferSize);
for (size_t i = 0; i < n; ++i) {
buffer[i] = rnd();
}
size_t written = ::fwrite(buffer, sizeof(uint32_t), n, fp);
PCHECK(written == n);
size -= written;
}
// make sure the buffers are flushed
int ret = ::fflush(fp);
PCHECK(ret == 0);
ret = ::fdatasync(::fileno(fp));
PCHECK(ret == 0);
PCHECK(::fclose(fp) == 0);
}
TemporaryFile::~TemporaryFile() {
try {
folly::fs::remove(path_);
} catch (const folly::fs::filesystem_error& e) {
LOG(ERROR) << "fs::remove: " << folly::exceptionStr(e);
}
}
TemporaryFile& TemporaryFile::getTempFile() {
static TemporaryFile sTempFile(6 << 20); // 6MiB
return sTempFile;
}
} // namespace async_base_test_lib_detail
} // namespace test
} // namespace folly
......@@ -21,6 +21,8 @@ using folly::AsyncIO;
namespace folly {
namespace test {
namespace async_base_test_lib_detail {
INSTANTIATE_TYPED_TEST_CASE_P(AsyncTest, AsyncTest, AsyncIO);
} // namespace async_base_test_lib_detail
} // namespace test
} // namespace folly
/*
* Copyright (c) Facebook, Inc. and its affiliates.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <folly/experimental/io/test/IoTestTempFileUtil.h>
#include <folly/FileUtil.h>
#include <folly/String.h>
#include <random>
namespace folly {
namespace test {
TemporaryFile TempFileUtil::getTempFile(size_t size) {
CHECK_EQ(size % sizeof(uint32_t), 0);
size /= sizeof(uint32_t);
TemporaryFile tmpFile;
int fd = tmpFile.fd();
CHECK_GE(fd, 0);
// fill the file the file with random data
const uint32_t seed = 42;
std::mt19937 rnd(seed);
constexpr size_t bufferSize = 1U << 16;
uint32_t buffer[bufferSize];
while (size) {
size_t n = std::min(size, bufferSize);
for (size_t i = 0; i < n; ++i) {
buffer[i] = rnd();
}
size_t written = folly::writeFull(fd, buffer, sizeof(uint32_t) * n);
CHECK_EQ(written, sizeof(uint32_t) * n);
size -= n;
}
CHECK_EQ(::fdatasync(fd), 0);
// the file was opened with O_EXCL so we need to close to be able
// to open it again
tmpFile.close();
return tmpFile;
}
} // namespace test
} // namespace folly
/*
* Copyright (c) Facebook, Inc. and its affiliates.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <glog/logging.h>
#include <folly/experimental/TestUtil.h>
#include <folly/experimental/io/FsUtil.h>
namespace folly {
namespace test {
class TempFileUtil {
public:
// Returns a temporary file that is NOT kept open
// but is deleted on destruction
// Generate random-looking but reproduceable data.
static TemporaryFile getTempFile(size_t size);
private:
TempFileUtil() = delete;
};
} // namespace test
} // namespace folly
......@@ -15,9 +15,12 @@
*/
#include <sys/eventfd.h>
#include <numeric>
#include <folly/FileUtil.h>
#include <folly/String.h>
#include <folly/experimental/io/IoUringBackend.h>
#include <folly/experimental/io/test/IoTestTempFileUtil.h>
#include <folly/init/Init.h>
#include <folly/io/async/AsyncUDPServerSocket.h>
#include <folly/io/async/AsyncUDPSocket.h>
......@@ -532,6 +535,184 @@ TEST(IoUringBackend, RegisteredFds) {
}
}
TEST(IoUringBackend, FileReadWrite) {
static constexpr size_t kBackendCapacity = 2048;
static constexpr size_t kBackendMaxSubmit = 32;
static constexpr size_t kBackendMaxGet = 32;
std::unique_ptr<folly::IoUringBackend> backend;
try {
folly::PollIoBackend::Options options;
options.setCapacity(kBackendCapacity)
.setMaxSubmit(kBackendMaxSubmit)
.setMaxGet(kBackendMaxGet)
.setUseRegisteredFds(false);
backend = std::make_unique<folly::IoUringBackend>(options);
} catch (const folly::IoUringBackend::NotAvailable&) {
}
SKIP_IF(!backend) << "Backend not available";
static constexpr size_t kNumBlocks = 512;
static constexpr size_t kBlockSize = 4096;
static constexpr size_t kFileSize = kNumBlocks * kBlockSize;
auto tempFile = folly::test::TempFileUtil::getTempFile(kFileSize);
int fd = ::open(tempFile.path().c_str(), O_DIRECT | O_RDWR);
SKIP_IF(fd == -1) << "Tempfile can't be opened with O_DIRECT: "
<< folly::errnoStr(errno);
SCOPE_EXIT {
::close(fd);
};
folly::EventBase evb(std::move(backend));
auto* backendPtr = dynamic_cast<folly::IoUringBackend*>(evb.getBackend());
CHECK(!!backendPtr);
size_t num = 0;
std::string writeData(kBlockSize, 'A'), readData(kBlockSize, 'Z');
std::vector<std::string> writeDataVec(kNumBlocks, writeData),
readDataVec(kNumBlocks, readData);
CHECK_NE(readData, writeData);
for (size_t i = 0; i < kNumBlocks; i++) {
folly::IoUringBackend::FileOpCallback writeCb = [&, i](int res) {
CHECK_EQ(res, writeDataVec[i].size());
folly::IoUringBackend::FileOpCallback readCb = [&, i](int res) {
CHECK_EQ(res, readDataVec[i].size());
CHECK_EQ(readDataVec[i], writeDataVec[i]);
++num;
};
backendPtr->queueRead(
fd,
readDataVec[i].data(),
readDataVec[i].size(),
i * kBlockSize,
std::move(readCb));
};
backendPtr->queueWrite(
fd,
writeDataVec[i].data(),
writeDataVec[i].size(),
i * kBlockSize,
std::move(writeCb));
}
evb.loop();
EXPECT_EQ(num, kNumBlocks);
}
TEST(IoUringBackend, FileReadvWritev) {
static constexpr size_t kBackendCapacity = 2048;
static constexpr size_t kBackendMaxSubmit = 32;
static constexpr size_t kBackendMaxGet = 32;
std::unique_ptr<folly::IoUringBackend> backend;
try {
folly::PollIoBackend::Options options;
options.setCapacity(kBackendCapacity)
.setMaxSubmit(kBackendMaxSubmit)
.setMaxGet(kBackendMaxGet)
.setUseRegisteredFds(false);
backend = std::make_unique<folly::IoUringBackend>(options);
} catch (const folly::IoUringBackend::NotAvailable&) {
}
SKIP_IF(!backend) << "Backend not available";
static constexpr size_t kNumBlocks = 512;
static constexpr size_t kNumIov = 4;
static constexpr size_t kIovSize = 1024;
static constexpr size_t kBlockSize = kNumIov * kIovSize;
static constexpr size_t kFileSize = kNumBlocks * kBlockSize;
auto tempFile = folly::test::TempFileUtil::getTempFile(kFileSize);
int fd = ::open(tempFile.path().c_str(), O_DIRECT | O_RDWR);
SKIP_IF(fd == -1) << "Tempfile can't be opened with O_DIRECT: "
<< folly::errnoStr(errno);
SCOPE_EXIT {
::close(fd);
};
folly::EventBase evb(std::move(backend));
auto* backendPtr = dynamic_cast<folly::IoUringBackend*>(evb.getBackend());
CHECK(!!backendPtr);
size_t num = 0;
std::string writeData(kIovSize, 'A'), readData(kIovSize, 'Z');
std::vector<std::string> writeDataVec(kNumIov, writeData),
readDataVec(kNumIov, readData);
std::vector<std::vector<std::string>> writeDataVecVec(
kNumBlocks, writeDataVec),
readDataVecVec(kNumBlocks, readDataVec);
CHECK(readDataVec != writeDataVec);
std::vector<std::vector<struct iovec>> readDataIov, writeDataIov;
std::vector<size_t> lenVec;
readDataIov.reserve(kNumBlocks);
writeDataIov.reserve(kNumBlocks);
lenVec.reserve(kNumBlocks);
for (size_t i = 0; i < kNumBlocks; i++) {
size_t len = 0;
std::vector<struct iovec> readIov, writeIov;
readIov.reserve(kNumIov);
writeIov.reserve(kNumIov);
for (size_t j = 0; j < kNumIov; j++) {
struct iovec riov {
readDataVecVec[i][j].data(), readDataVecVec[i][j].size()
};
readIov.push_back(riov);
struct iovec wiov {
writeDataVecVec[i][j].data(), writeDataVecVec[i][j].size()
};
writeIov.push_back(wiov);
len += riov.iov_len;
}
readDataIov.emplace_back(std::move(readIov));
writeDataIov.emplace_back(std::move(writeIov));
lenVec.emplace_back(len);
}
for (size_t i = 0; i < kNumBlocks; i++) {
folly::IoUringBackend::FileOpCallback writeCb = [&, i](int res) {
CHECK_EQ(res, lenVec[i]);
folly::IoUringBackend::FileOpCallback readCb = [&, i](int res) {
CHECK_EQ(res, lenVec[i]);
CHECK(readDataVecVec[i] == writeDataVecVec[i]);
if (++num == kNumBlocks) {
evb.terminateLoopSoon();
}
};
backendPtr->queueReadv(
fd, readDataIov[i], i * kBlockSize, std::move(readCb));
};
backendPtr->queueWritev(
fd, writeDataIov[i], i * kBlockSize, std::move(writeCb));
}
evb.loopForever();
EXPECT_EQ(num, kNumBlocks);
}
namespace folly {
namespace test {
static constexpr size_t kCapacity = 16 * 1024;
......
......@@ -22,7 +22,9 @@ using folly::IoUring;
namespace folly {
namespace test {
namespace async_base_test_lib_detail {
INSTANTIATE_TYPED_TEST_CASE_P(AsyncTest, AsyncTest, IoUring);
} // namespace async_base_test_lib_detail
} // namespace test
} // 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