Commit 584b74cf authored by Dan Melnic's avatar Dan Melnic Committed by Facebook Github Bot

Add support for registered fds

Summary: Add support for registered fds

Reviewed By: kevin-vigor

Differential Revision: D19313387

fbshipit-source-id: 19527958b3986992cdafbad2c4c4902c887a9e9f
parent dafd4506
...@@ -23,8 +23,73 @@ ...@@ -23,8 +23,73 @@
#include <glog/logging.h> #include <glog/logging.h>
namespace folly { namespace folly {
IoUringBackend::IoUringBackend(size_t capacity, size_t maxSubmit, size_t maxGet) IoUringBackend::FdRegistry::FdRegistry(struct io_uring& ioRing, size_t n)
: PollIoBackend(capacity, maxSubmit, maxGet) { : ioRing_(ioRing), files_(n, -1), inUse_(n), records_(n) {}
int IoUringBackend::FdRegistry::init() {
if (inUse_) {
int ret = ::io_uring_register_files(&ioRing_, files_.data(), inUse_);
if (!ret) {
// build and set the free list head if we succeed
for (size_t i = 0; i < records_.size(); i++) {
records_[i].idx_ = i;
free_.push_front(records_[i]);
}
} else {
LOG(ERROR) << "io_uring_register_files(" << inUse_ << ") "
<< "failed errno = " << errno << ":\""
<< folly::errnoStr(errno) << "\" " << this;
}
return ret;
}
return 0;
}
IoUringBackend::FdRegistrationRecord* IoUringBackend::FdRegistry::alloc(
int fd) {
if (FOLLY_UNLIKELY(free_.empty())) {
return nullptr;
}
auto& ret = free_.front();
// now we have an idx
if (::io_uring_register_files_update(&ioRing_, ret.idx_, &fd, 1) != 1) {
return nullptr;
}
ret.fd_ = fd;
ret.count_ = 1;
free_.pop_front();
return &ret;
}
bool IoUringBackend::FdRegistry::free(
IoUringBackend::FdRegistrationRecord* record) {
if (record && (--record->count_ == 0)) {
record->fd_ = -1;
int ret = ::io_uring_register_files_update(
&ioRing_, record->idx_, &record->fd_, 1);
// we add it to the free list anyway here
free_.push_front(*record);
return (ret == 1);
}
return false;
}
IoUringBackend::IoUringBackend(
size_t capacity,
size_t maxSubmit,
size_t maxGet,
bool useRegisteredFds)
: PollIoBackend(capacity, maxSubmit, maxGet),
fdRegistry_(ioRing_, useRegisteredFds ? capacity : 0) {
::memset(&ioRing_, 0, sizeof(ioRing_)); ::memset(&ioRing_, 0, sizeof(ioRing_));
::memset(&params_, 0, sizeof(params_)); ::memset(&params_, 0, sizeof(params_));
...@@ -62,6 +127,15 @@ IoUringBackend::IoUringBackend(size_t capacity, size_t maxSubmit, size_t maxGet) ...@@ -62,6 +127,15 @@ IoUringBackend::IoUringBackend(size_t capacity, size_t maxSubmit, size_t maxGet)
entries_[numEntries_ - 1].backendCb_ = PollIoBackend::processPollIoCb; entries_[numEntries_ - 1].backendCb_ = PollIoBackend::processPollIoCb;
freeHead_ = &entries_[1]; freeHead_ = &entries_[1];
// we need to call the init before adding the timer fd
// so we avoid a deadlock - waiting for the queue to be drained
if (useRegisteredFds) {
// now init the file registry
// if this fails, we still continue since we
// can run without registered fds
fdRegistry_.init();
}
// add the timer fd // add the timer fd
if (!addTimerFd()) { if (!addTimerFd()) {
cleanup(); cleanup();
...@@ -206,7 +280,11 @@ size_t IoUringBackend::submitList( ...@@ -206,7 +280,11 @@ size_t IoUringBackend::submitList(
CHECK(sqe); // this should not happen CHECK(sqe); // this should not happen
auto* ev = entry->event_->getEvent(); auto* ev = entry->event_->getEvent();
entry->prepPollAdd(sqe, ev->ev_fd, getPollFlags(ev->ev_events)); entry->prepPollAdd(
sqe,
ev->ev_fd,
getPollFlags(ev->ev_events),
(ev->ev_events & EV_PERSIST) != 0);
i++; i++;
if (ioCbs.empty()) { if (ioCbs.empty()) {
int num = (waitForEvents == WaitForEventsMode::WAIT) int num = (waitForEvents == WaitForEventsMode::WAIT)
......
...@@ -35,14 +35,43 @@ class IoUringBackend : public PollIoBackend { ...@@ -35,14 +35,43 @@ class IoUringBackend : public PollIoBackend {
explicit IoUringBackend( explicit IoUringBackend(
size_t capacity, size_t capacity,
size_t maxSubmit = 128, size_t maxSubmit = 128,
size_t maxGet = static_cast<size_t>(-1)); size_t maxGet = static_cast<size_t>(-1),
bool useRegisteredFds = false);
~IoUringBackend() override; ~IoUringBackend() override;
// returns true if the current Linux kernel version // returns true if the current Linux kernel version
// supports the io_uring backend // supports the io_uring backend
static bool isAvailable(); static bool isAvailable();
// from PollIoBackend
FdRegistrationRecord* registerFd(int fd) override {
return fdRegistry_.alloc(fd);
}
bool unregisterFd(FdRegistrationRecord* rec) override {
return fdRegistry_.free(rec);
}
protected: protected:
struct FdRegistry {
FdRegistry() = delete;
FdRegistry(struct io_uring& ioRing, size_t n);
FdRegistrationRecord* alloc(int fd);
bool free(FdRegistrationRecord* record);
int init();
size_t update();
struct io_uring& ioRing_;
std::vector<int> files_;
size_t inUse_;
std::vector<FdRegistrationRecord> records_;
boost::intrusive::
slist<FdRegistrationRecord, boost::intrusive::cache_last<false>>
free_;
};
// from PollIoBackend // from PollIoBackend
void* allocSubmissionEntry() override; void* allocSubmissionEntry() override;
int getActiveEvents(WaitForEventsMode waitForEvents) override; int getActiveEvents(WaitForEventsMode waitForEvents) override;
...@@ -58,10 +87,20 @@ class IoUringBackend : public PollIoBackend { ...@@ -58,10 +87,20 @@ class IoUringBackend : public PollIoBackend {
: PollIoBackend::IoCb(backend, poolAlloc) {} : PollIoBackend::IoCb(backend, poolAlloc) {}
~IoSqe() override = default; ~IoSqe() override = default;
void prepPollAdd(void* entry, int fd, uint32_t events) override { void prepPollAdd(void* entry, int fd, uint32_t events, bool registerFd)
override {
CHECK(entry); CHECK(entry);
struct io_uring_sqe* sqe = reinterpret_cast<struct io_uring_sqe*>(entry); struct io_uring_sqe* sqe = reinterpret_cast<struct io_uring_sqe*>(entry);
::io_uring_prep_poll_add(sqe, fd, events); if (registerFd && !fdRecord_) {
fdRecord_ = backend_->registerFd(fd);
}
if (fdRecord_) {
::io_uring_prep_poll_add(sqe, fdRecord_->idx_, events);
sqe->flags |= IOSQE_FIXED_FILE;
} else {
::io_uring_prep_poll_add(sqe, fd, events);
}
::io_uring_sqe_set_data(sqe, this); ::io_uring_sqe_set_data(sqe, this);
} }
...@@ -93,5 +132,7 @@ class IoUringBackend : public PollIoBackend { ...@@ -93,5 +132,7 @@ class IoUringBackend : public PollIoBackend {
uint32_t sqRingMask_{0}; uint32_t sqRingMask_{0};
uint32_t cqRingMask_{0}; uint32_t cqRingMask_{0};
FdRegistry fdRegistry_;
}; };
} // namespace folly } // namespace folly
...@@ -57,7 +57,7 @@ PollIoBackend::~PollIoBackend() { ...@@ -57,7 +57,7 @@ PollIoBackend::~PollIoBackend() {
bool PollIoBackend::addTimerFd() { bool PollIoBackend::addTimerFd() {
auto* entry = allocSubmissionEntry(); // this can be nullptr auto* entry = allocSubmissionEntry(); // this can be nullptr
timerEntry_->prepPollAdd(entry, timerFd_, POLLIN); timerEntry_->prepPollAdd(entry, timerFd_, POLLIN, true /*registerFd*/);
return (1 == submitOne(timerEntry_)); return (1 == submitOne(timerEntry_));
} }
...@@ -196,6 +196,12 @@ PollIoBackend::IoCb* PollIoBackend::allocIoCb() { ...@@ -196,6 +196,12 @@ PollIoBackend::IoCb* PollIoBackend::allocIoCb() {
void PollIoBackend::releaseIoCb(PollIoBackend::IoCb* aioIoCb) { void PollIoBackend::releaseIoCb(PollIoBackend::IoCb* aioIoCb) {
numIoCbInUse_--; numIoCbInUse_--;
// unregister the file descriptor record
if (aioIoCb->fdRecord_) {
unregisterFd(aioIoCb->fdRecord_);
aioIoCb->fdRecord_ = nullptr;
}
if (FOLLY_LIKELY(aioIoCb->poolAlloc_)) { if (FOLLY_LIKELY(aioIoCb->poolAlloc_)) {
aioIoCb->event_ = nullptr; aioIoCb->event_ = nullptr;
aioIoCb->next_ = freeHead_; aioIoCb->next_ = freeHead_;
...@@ -356,7 +362,11 @@ int PollIoBackend::eb_event_add(Event& event, const struct timeval* timeout) { ...@@ -356,7 +362,11 @@ int PollIoBackend::eb_event_add(Event& event, const struct timeval* timeout) {
return 0; return 0;
} else { } else {
auto* entry = allocSubmissionEntry(); // this can be nullptr auto* entry = allocSubmissionEntry(); // this can be nullptr
iocb->prepPollAdd(entry, ev->ev_fd, getPollFlags(ev->ev_events)); iocb->prepPollAdd(
entry,
ev->ev_fd,
getPollFlags(ev->ev_events),
(ev->ev_events & EV_PERSIST) != 0);
int ret = submitOne(iocb); int ret = submitOne(iocb);
if (ret == 1) { if (ret == 1) {
if (~event_ref_flags(ev) & EVLIST_INTERNAL) { if (~event_ref_flags(ev) & EVLIST_INTERNAL) {
......
...@@ -24,6 +24,7 @@ ...@@ -24,6 +24,7 @@
#include <vector> #include <vector>
#include <boost/intrusive/list.hpp> #include <boost/intrusive/list.hpp>
#include <boost/intrusive/slist.hpp>
#include <folly/CPortability.h> #include <folly/CPortability.h>
#include <folly/CppAttributes.h> #include <folly/CppAttributes.h>
...@@ -47,6 +48,21 @@ class PollIoBackend : public EventBaseBackendBase { ...@@ -47,6 +48,21 @@ class PollIoBackend : public EventBaseBackendBase {
int eb_event_add(Event& event, const struct timeval* timeout) override; int eb_event_add(Event& event, const struct timeval* timeout) override;
int eb_event_del(Event& event) override; int eb_event_del(Event& event) override;
struct FdRegistrationRecord : public boost::intrusive::slist_base_hook<
boost::intrusive::cache_last<false>> {
int count_{0};
int fd_{-1};
size_t idx_{0};
};
virtual FdRegistrationRecord* registerFd(int /*fd*/) {
return nullptr;
}
virtual bool unregisterFd(FdRegistrationRecord* /*rec*/) {
return false;
}
protected: protected:
enum class WaitForEventsMode { WAIT, DONT_WAIT }; enum class WaitForEventsMode { WAIT, DONT_WAIT };
struct IoCb; struct IoCb;
...@@ -65,6 +81,7 @@ class PollIoBackend : public EventBaseBackendBase { ...@@ -65,6 +81,7 @@ class PollIoBackend : public EventBaseBackendBase {
const bool poolAlloc_; const bool poolAlloc_;
IoCb* next_{nullptr}; // this is for the free list IoCb* next_{nullptr}; // this is for the free list
Event* event_{nullptr}; Event* event_{nullptr};
FdRegistrationRecord* fdRecord_{nullptr};
size_t useCount_{0}; size_t useCount_{0};
FOLLY_ALWAYS_INLINE void resetEvent() { FOLLY_ALWAYS_INLINE void resetEvent() {
...@@ -76,7 +93,8 @@ class PollIoBackend : public EventBaseBackendBase { ...@@ -76,7 +93,8 @@ class PollIoBackend : public EventBaseBackendBase {
} }
} }
virtual void prepPollAdd(void* entry, int fd, uint32_t events) = 0; virtual void
prepPollAdd(void* entry, int fd, uint32_t events, bool registerFd) = 0;
}; };
using IoCbList = using IoCbList =
......
...@@ -141,27 +141,106 @@ TEST(IoUringBackend, OverflowPersist) { ...@@ -141,27 +141,106 @@ TEST(IoUringBackend, OverflowPersist) {
testOverflow(true, true); testOverflow(true, true);
} }
TEST(IoUringBackend, RegisteredFds) {
static constexpr size_t kBackendCapacity = 64;
static constexpr size_t kBackendMaxSubmit = 32;
static constexpr size_t kBackendMaxGet = 32;
std::unique_ptr<folly::IoUringBackend> backendReg;
std::unique_ptr<folly::IoUringBackend> backendNoReg;
try {
backendReg = std::make_unique<folly::IoUringBackend>(
kBackendCapacity,
kBackendMaxSubmit,
kBackendMaxGet,
true /*useRegisteredFds*/);
backendNoReg = std::make_unique<folly::IoUringBackend>(
kBackendCapacity,
kBackendMaxSubmit,
kBackendMaxGet,
false /*useRegisteredFds*/);
} catch (const folly::IoUringBackend::NotAvailable&) {
}
SKIP_IF(!backendReg) << "Backend not available";
SKIP_IF(!backendNoReg) << "Backend not available";
int eventFd = ::eventfd(0, EFD_CLOEXEC | EFD_SEMAPHORE | EFD_NONBLOCK);
CHECK_GT(eventFd, 0);
SCOPE_EXIT {
::close(eventFd);
};
// verify for useRegisteredFds = false we get a nullptr FdRegistrationRecord
auto* record = backendNoReg->registerFd(eventFd);
CHECK(!record);
std::vector<folly::IoUringBackend::FdRegistrationRecord*> records;
// we use kBackendCapacity -1 since we can have the timerFd
// already using one fd
records.reserve(kBackendCapacity - 1);
for (size_t i = 0; i < kBackendCapacity - 1; i++) {
record = backendReg->registerFd(eventFd);
CHECK(record);
records.emplace_back(record);
}
// try to allocate one more and check if we get a nullptr
record = backendReg->registerFd(eventFd);
CHECK(!record);
// deallocate and allocate again
for (size_t i = 0; i < records.size(); i++) {
CHECK(backendReg->unregisterFd(records[i]));
record = backendReg->registerFd(eventFd);
CHECK(record);
records[i] = record;
}
}
namespace folly { namespace folly {
namespace test { namespace test {
static constexpr size_t kCapacity = 16 * 1024; static constexpr size_t kCapacity = 16 * 1024;
static constexpr size_t kMaxSubmit = 128; static constexpr size_t kMaxSubmit = 128;
static constexpr size_t kMaxGet = static_cast<size_t>(-1);
struct IoUringBackendProvider { struct IoUringBackendProvider {
static std::unique_ptr<folly::EventBaseBackendBase> getBackend() { static std::unique_ptr<folly::EventBaseBackendBase> getBackend() {
try { try {
return std::make_unique<folly::IoUringBackend>(kCapacity, kMaxSubmit); return std::make_unique<folly::IoUringBackend>(
kCapacity, kMaxSubmit, kMaxGet, false /* useRegisteredFds */);
} catch (const IoUringBackend::NotAvailable&) { } catch (const IoUringBackend::NotAvailable&) {
return nullptr; return nullptr;
} }
} }
}; };
struct IoUringRegFdBackendProvider {
static std::unique_ptr<folly::EventBaseBackendBase> getBackend() {
try {
return std::make_unique<folly::IoUringBackend>(
kCapacity, kMaxSubmit, kMaxGet, true /* useRegisteredFds */);
} catch (const IoUringBackend::NotAvailable&) {
return nullptr;
}
}
};
// Instantiate the non registered fd tests
INSTANTIATE_TYPED_TEST_CASE_P(IoUring, EventBaseTest, IoUringBackendProvider);
INSTANTIATE_TYPED_TEST_CASE_P(IoUring, EventBaseTest1, IoUringBackendProvider);
// Instantiate the registered fd tests
INSTANTIATE_TYPED_TEST_CASE_P( INSTANTIATE_TYPED_TEST_CASE_P(
IoUringRegFd,
EventBaseTest, EventBaseTest,
EventBaseTest, IoUringRegFdBackendProvider);
IoUringBackendProvider);
INSTANTIATE_TYPED_TEST_CASE_P( INSTANTIATE_TYPED_TEST_CASE_P(
IoUringRegFd,
EventBaseTest1, EventBaseTest1,
EventBaseTest1, IoUringRegFdBackendProvider);
IoUringBackendProvider);
} // namespace test } // namespace test
} // namespace folly } // 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 <sys/eventfd.h>
#include <folly/Benchmark.h>
#include <folly/FileUtil.h>
#include <folly/experimental/io/IoUringBackend.h>
#include <folly/io/async/EventBase.h>
#include <folly/io/async/EventHandler.h>
#include <folly/portability/GFlags.h>
using namespace folly;
namespace {
class EventFD : public EventHandler {
public:
EventFD(uint64_t num, uint64_t& total, bool persist, EventBase* eventBase)
: EventFD(total, createFd(num), persist, eventBase) {}
~EventFD() override {
unregisterHandler();
if (fd_ > 0) {
changeHandlerFD(NetworkSocket());
::close(fd_);
fd_ = -1;
}
}
// from folly::EventHandler
void handlerReady(uint16_t /*events*/) noexcept override {
// we do not read to leave the fd signalled
if (!persist_) {
registerHandler(folly::EventHandler::READ);
}
if (total_ > 0) {
--total_;
if (total_ == 0) {
evb_->terminateLoopSoon();
}
}
}
private:
static int createFd(uint64_t num) {
// we want a semaphore
int fd = ::eventfd(0, EFD_CLOEXEC | EFD_SEMAPHORE | EFD_NONBLOCK);
CHECK_GT(fd, 0);
CHECK_EQ(writeNoInt(fd, &num, sizeof(num)), sizeof(num));
return fd;
}
EventFD(uint64_t& total, int fd, bool persist, EventBase* eventBase)
: EventHandler(eventBase, NetworkSocket::fromFd(fd)),
total_(total),
fd_(fd),
persist_(persist),
evb_(eventBase) {
if (persist_) {
registerHandler(folly::EventHandler::READ | folly::EventHandler::PERSIST);
} else {
registerHandler(folly::EventHandler::READ);
}
}
uint64_t& total_;
int fd_{-1};
bool persist_;
EventBase* evb_;
};
// static constexpr size_t kAsyncIoEvents = 2048;
class BackendEventBase : public EventBase {
public:
explicit BackendEventBase(bool useRegisteredFds, size_t capacity = 32 * 1024)
: EventBase(getBackend(useRegisteredFds, capacity), false) {}
private:
static std::unique_ptr<folly::EventBaseBackendBase> getBackend(
bool useRegisteredFds,
size_t capacity) {
return std::make_unique<IoUringBackend>(
capacity, 256, 128, useRegisteredFds);
}
};
void runTest(
unsigned int iters,
bool persist,
bool useRegisteredFds,
size_t numEvents) {
BenchmarkSuspender suspender;
uint64_t total = iters * numEvents;
BackendEventBase evb(useRegisteredFds);
std::vector<std::unique_ptr<EventFD>> eventsVec;
eventsVec.reserve(numEvents);
for (size_t i = 0; i < numEvents; i++) {
eventsVec.emplace_back(std::make_unique<EventFD>(1, total, persist, &evb));
}
suspender.dismiss();
evb.loopForever();
suspender.rehire();
}
} // namespace
BENCHMARK_DRAW_LINE();
BENCHMARK_NAMED_PARAM(runTest, io_uring_persist_1, true, false, 1)
BENCHMARK_RELATIVE_NAMED_PARAM(runTest, io_uring_persist_reg_1, true, true, 1)
BENCHMARK_DRAW_LINE();
BENCHMARK_NAMED_PARAM(runTest, io_uring_persist_64, true, false, 64)
BENCHMARK_RELATIVE_NAMED_PARAM(runTest, io_uring_persist_reg_64, true, true, 64)
BENCHMARK_DRAW_LINE();
BENCHMARK_NAMED_PARAM(runTest, io_uring_persist_128, true, false, 128)
BENCHMARK_RELATIVE_NAMED_PARAM(
runTest,
io_uring_persist_reg_128,
true,
true,
128)
BENCHMARK_DRAW_LINE();
// add a non persistent benchamrk too
// so we can see the useRegisteredFds flag
// does not have any effect
BENCHMARK_DRAW_LINE();
BENCHMARK_NAMED_PARAM(runTest, io_uring_no_persist_128, false, false, 128)
BENCHMARK_RELATIVE_NAMED_PARAM(
runTest,
io_uring_no_persist_reg_128,
false,
true,
128)
BENCHMARK_DRAW_LINE();
int main(int argc, char** argv) {
gflags::ParseCommandLineFlags(&argc, &argv, true);
runBenchmarks();
}
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