folly::AsyncIO Class Reference

#include <AsyncIO.h>

Inheritance diagram for folly::AsyncIO:

Public Types
enum	PollMode { NOT_POLLABLE, POLLABLE }
typedef AsyncIOOp	Op

Public Member Functions
	AsyncIO (size_t capacity, PollMode pollMode=NOT_POLLABLE)
	~AsyncIO ()
Range< Op ** >	wait (size_t minRequests)
Range< Op ** >	cancel ()
size_t	pending () const
size_t	capacity () const
size_t	totalSubmits () const
int	pollFd () const
Range< Op ** >	pollCompleted ()
void	submit (Op *op)

Private Types
enum	WaitType { WaitType::COMPLETE, WaitType::CANCEL }

Private Member Functions
void	decrementPending ()
void	initializeContext ()
Range< AsyncIO::Op ** >	doWait (WaitType type, size_t minRequests, size_t maxRequests, std::vector< Op * > &result)

Private Attributes
io_context_t	ctx_ {nullptr}
std::atomic< bool >	ctxSet_ {false}
std::mutex	initMutex_
std::atomic< size_t >	pending_ {0}
std::atomic< size_t >	submitted_ {0}
const size_t	capacity_
int	pollFd_ {-1}
std::vector< Op * >	completed_
std::vector< Op * >	canceled_

Detailed Description

C++ interface around Linux Async IO.

Definition at line 125 of file AsyncIO.h.

Member Typedef Documentation

typedef AsyncIOOp folly::AsyncIO::Op

Definition at line 127 of file AsyncIO.h.

Member Enumeration Documentation

enum folly::AsyncIO::PollMode

Enumerator
NOT_POLLABLE
POLLABLE

Definition at line 129 of file AsyncIO.h.

                {
    NOT_POLLABLE,
    POLLABLE,
  };

enum folly::AsyncIO::WaitType

strongprivate

Enumerator
COMPLETE
CANCEL

Definition at line 218 of file AsyncIO.h.

{ COMPLETE, CANCEL };

Constructor & Destructor Documentation

folly::AsyncIO::AsyncIO ( size_t  capacity, PollMode  pollMode = NOT_POLLABLE  )

explicit

Create an AsyncIO context capable of holding at most 'capacity' pending requests at the same time. As requests complete, others can be scheduled, as long as this limit is not exceeded.

Note: the maximum number of allowed concurrent requests is controlled by the fs.aio-max-nr sysctl, the default value is usually 64K.

If pollMode is POLLABLE, pollFd() will return a file descriptor that can be passed to poll / epoll / select and will become readable when any IOs on this AsyncIO have completed. If you do this, you must use pollCompleted() instead of wait() – do not read from the pollFd() file descriptor directly.

You may use the same AsyncIO object from multiple threads, as long as there is only one concurrent caller of wait() / pollCompleted() / cancel() (perhaps by always calling it from the same thread, or by providing appropriate mutual exclusion). In this case, pending() returns a snapshot of the current number of pending requests.

Definition at line 110 of file AsyncIO.cpp.

References capacity_, folly::checkUnixError(), completed_, EFD_NONBLOCK, eventfd, POLLABLE, and pollFd_.

                                                   : capacity_(capacity) {
  CHECK_GT(capacity_, 0);
  completed_.reserve(capacity_);
  if (pollMode == POLLABLE) {
    pollFd_ = eventfd(0, EFD_NONBLOCK);
    checkUnixError(pollFd_, "AsyncIO: eventfd creation failed");
  }
}

folly::AsyncIO::~AsyncIO

(

)

Definition at line 119 of file AsyncIO.cpp.

References folly::netops::close(), ctx_, folly::errnoStr(), pending_, and pollFd_.

                  {
  CHECK_EQ(pending_, 0);
  if (ctx_) {
    int rc = io_queue_release(ctx_);
    CHECK_EQ(rc, 0) << "io_queue_release: " << errnoStr(-rc);
  }
  if (pollFd_ != -1) {
    CHECK_ERR(close(pollFd_));
  }
}

Member Function Documentation

Range< AsyncIO::Op ** > folly::AsyncIO::cancel

(

)

Cancel all pending requests and return them; the returned range is valid until the next call to cancel().

Definition at line 197 of file AsyncIO.cpp.

References CANCEL, canceled_, ctx_, doWait(), and pending_.

Referenced by TEST().

                                   {
  CHECK(ctx_);
  auto p = pending_.load(std::memory_order_acquire);
  return doWait(WaitType::CANCEL, p, p, canceled_);
}

size_t folly::AsyncIO::capacity ( ) const

inline

Return the maximum number of requests that can be kept outstanding at any one time.

Definition at line 181 of file AsyncIO.h.

Referenced by folly::AsyncIOQueue::maybeDequeue().

                          {
    return capacity_;
  }

void folly::AsyncIO::decrementPending ( )

private

Definition at line 130 of file AsyncIO.cpp.

References pending_.

Referenced by doWait(), and submit().

                               {
  auto p = pending_.fetch_add(-1, std::memory_order_acq_rel);
  DCHECK_GE(p, 1);
}

Range< AsyncIO::Op ** > folly::AsyncIO::doWait ( WaitType  type, size_t  minRequests, size_t  maxRequests, std::vector< Op * > &  result  )

private

Definition at line 226 of file AsyncIO.cpp.

References folly::AsyncIOOp::cancel(), CANCEL, folly::AsyncIOOp::complete(), COMPLETE, count, ctx_, decrementPending(), folly::errnoStr(), i, folly::AsyncIOOp::iocb_, and folly::range().

Referenced by cancel(), pollCompleted(), and wait().

                            {
  io_event events[maxRequests];

  // Unfortunately, Linux AIO doesn't implement io_cancel, so even for
  // WaitType::CANCEL we have to wait for IO completion.
  size_t count = 0;
  do {
    int ret;
    do {
      // GOTCHA: io_getevents() may returns less than min_nr results if
      // interrupted after some events have been read (if before, -EINTR
      // is returned).
      ret = io_getevents(
          ctx_,
          minRequests - count,
          maxRequests - count,
          events + count,
          /* timeout */ nullptr); // wait forever
    } while (ret == -EINTR);
    // Check as may not be able to recover without leaking events.
    CHECK_GE(ret, 0) << "AsyncIO: io_getevents failed with error "
                     << errnoStr(-ret);
    count += ret;
  } while (count < minRequests);
  DCHECK_LE(count, maxRequests);

  result.clear();
  for (size_t i = 0; i < count; ++i) {
    DCHECK(events[i].obj);
    Op* op = boost::intrusive::get_parent_from_member(
        events[i].obj, &AsyncIOOp::iocb_);
    decrementPending();
    switch (type) {
      case WaitType::COMPLETE:
        op->complete(events[i].res);
        break;
      case WaitType::CANCEL:
        op->cancel();
        break;
    }
    result.push_back(op);
  }

  return range(result);
}

void folly::AsyncIO::initializeContext ( )

private

Definition at line 135 of file AsyncIO.cpp.

References capacity_, folly::checkKernelError(), ctx_, ctxSet_, initMutex_, and folly::lock().

Referenced by submit().

                                {
  if (!ctxSet_.load(std::memory_order_acquire)) {
    std::lock_guard<std::mutex> lock(initMutex_);
    if (!ctxSet_.load(std::memory_order_relaxed)) {
      int rc = io_queue_init(capacity_, &ctx_);
      // returns negative errno
      if (rc == -EAGAIN) {
        long aio_nr, aio_max;
        std::unique_ptr<FILE, int (*)(FILE*)> fp(
            fopen("/proc/sys/fs/aio-nr", "r"), fclose);
        PCHECK(fp);
        CHECK_EQ(fscanf(fp.get(), "%ld", &aio_nr), 1);

        std::unique_ptr<FILE, int (*)(FILE*)> aio_max_fp(
            fopen("/proc/sys/fs/aio-max-nr", "r"), fclose);
        PCHECK(aio_max_fp);
        CHECK_EQ(fscanf(aio_max_fp.get(), "%ld", &aio_max), 1);

        LOG(ERROR) << "No resources for requested capacity of " << capacity_;
        LOG(ERROR) << "aio_nr " << aio_nr << ", aio_max_nr " << aio_max;
      }

      checkKernelError(rc, "AsyncIO: io_queue_init failed");
      DCHECK(ctx_);
      ctxSet_.store(true, std::memory_order_release);
    }
  }
}

size_t folly::AsyncIO::pending ( ) const

inline

Return the number of pending requests.

Definition at line 173 of file AsyncIO.h.

Referenced by folly::AsyncIOQueue::maybeDequeue(), TEST(), and folly::AsyncIOQueue::~AsyncIOQueue().

                         {
    return pending_;
  }

Range< AsyncIO::Op ** > folly::AsyncIO::pollCompleted

(

)

If POLLABLE, call instead of wait after the file descriptor returned by pollFd() became readable. The returned range is valid until the next call to pollCompleted().

Definition at line 203 of file AsyncIO.cpp.

References folly::checkUnixError(), COMPLETE, completed_, ctx_, doWait(), pending_, pollFd_, fizz::detail::read(), uint64_t, and UNLIKELY.

                                          {
  CHECK(ctx_);
  CHECK_NE(pollFd_, -1) << "pollCompleted() only allowed on pollable object";
  uint64_t numEvents;
  // This sets the eventFd counter to 0, see
  // http://www.kernel.org/doc/man-pages/online/pages/man2/eventfd.2.html
  ssize_t rc;
  do {
    rc = ::read(pollFd_, &numEvents, 8);
  } while (rc == -1 && errno == EINTR);
  if (UNLIKELY(rc == -1 && errno == EAGAIN)) {
    return Range<Op**>(); // nothing completed
  }
  checkUnixError(rc, "AsyncIO: read from event fd failed");
  DCHECK_EQ(rc, 8);

  DCHECK_GT(numEvents, 0);
  DCHECK_LE(numEvents, pending_);

  // Don't reap more than numEvents, as we've just reset the counter to 0.
  return doWait(WaitType::COMPLETE, numEvents, numEvents, completed_);
}

int folly::AsyncIO::pollFd ( ) const

inline

If POLLABLE, return a file descriptor that can be passed to poll / epoll and will become readable when any async IO operations have completed. If NOT_POLLABLE, return -1.

Definition at line 198 of file AsyncIO.h.

References submit.

                     {
    return pollFd_;
  }

void folly::AsyncIO::submit

(

Op *

op

)

Submit an op for execution.

Definition at line 164 of file AsyncIO.cpp.

References capacity_, ctx_, decrementPending(), initializeContext(), folly::AsyncIOOp::INITIALIZED, folly::AsyncIOOp::iocb_, pending_, pollFd_, folly::AsyncIOOp::start(), folly::AsyncIOOp::state(), submitted_, and folly::throwSystemErrorExplicit().

Referenced by folly::AsyncIOQueue::maybeDequeue(), and TEST().

                           {
  CHECK_EQ(op->state(), Op::State::INITIALIZED);
  initializeContext(); // on demand

  // We can increment past capacity, but we'll clean up after ourselves.
  auto p = pending_.fetch_add(1, std::memory_order_acq_rel);
  if (p >= capacity_) {
    decrementPending();
    throw std::range_error("AsyncIO: too many pending requests");
  }
  iocb* cb = &op->iocb_;
  cb->data = nullptr; // unused
  if (pollFd_ != -1) {
    io_set_eventfd(cb, pollFd_);
  }
  int rc = io_submit(ctx_, 1, &cb);
  if (rc < 0) {
    decrementPending();
    throwSystemErrorExplicit(-rc, "AsyncIO: io_submit failed");
  }
  submitted_++;
  DCHECK_EQ(rc, 1);
  op->start();
}

size_t folly::AsyncIO::totalSubmits ( ) const

inline

Return the accumulative number of submitted I/O, since this object has been created.

Definition at line 189 of file AsyncIO.h.

                              {
    return submitted_;
  }

Range< AsyncIO::Op ** > folly::AsyncIO::wait

(

size_t

minRequests

)

Wait for at least minRequests to complete. Returns the requests that have completed; the returned range is valid until the next call to wait(). minRequests may be 0 to not block.

Definition at line 189 of file AsyncIO.cpp.

References COMPLETE, completed_, ctx_, doWait(), pending_, and pollFd_.

Referenced by TEST().

                                                   {
  CHECK(ctx_);
  CHECK_EQ(pollFd_, -1) << "wait() only allowed on non-pollable object";
  auto p = pending_.load(std::memory_order_acquire);
  CHECK_LE(minRequests, p);
  return doWait(WaitType::COMPLETE, minRequests, p, completed_);
}

Member Data Documentation

std::vector<Op*> folly::AsyncIO::canceled_

private

Definition at line 234 of file AsyncIO.h.

Referenced by cancel().

const size_t folly::AsyncIO::capacity_

private

Definition at line 231 of file AsyncIO.h.

Referenced by AsyncIO(), initializeContext(), and submit().

std::vector<Op*> folly::AsyncIO::completed_

private

Definition at line 233 of file AsyncIO.h.

Referenced by AsyncIO(), pollCompleted(), and wait().

io_context_t folly::AsyncIO::ctx_ {nullptr}

private

Definition at line 225 of file AsyncIO.h.

Referenced by cancel(), doWait(), initializeContext(), pollCompleted(), submit(), wait(), and ~AsyncIO().

std::atomic<bool> folly::AsyncIO::ctxSet_ {false}

private

Definition at line 226 of file AsyncIO.h.

Referenced by initializeContext().

std::mutex folly::AsyncIO::initMutex_

private

Definition at line 227 of file AsyncIO.h.

Referenced by initializeContext().

std::atomic<size_t> folly::AsyncIO::pending_ {0}

private

Definition at line 229 of file AsyncIO.h.

Referenced by cancel(), decrementPending(), pollCompleted(), submit(), wait(), and ~AsyncIO().

int folly::AsyncIO::pollFd_ {-1}

private

Definition at line 232 of file AsyncIO.h.

Referenced by AsyncIO(), pollCompleted(), submit(), wait(), and ~AsyncIO().

std::atomic<size_t> folly::AsyncIO::submitted_ {0}

private

Definition at line 230 of file AsyncIO.h.

Referenced by submit().

---

The documentation for this class was generated from the following files:

• proxygen/folly/folly/experimental/io/AsyncIO.h
• proxygen/folly/folly/experimental/io/AsyncIO.cpp