AsyncIO.h

Go to the documentation of this file.

/*
 * Copyright 2013-present Facebook, Inc.
 *
 * 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.
 */

#pragma once

#include <sys/types.h>

#include <atomic>
#include <cstdint>
#include <deque>
#include <functional>
#include <iosfwd>
#include <mutex>
#include <utility>
#include <vector>

#include <boost/noncopyable.hpp>
#include <libaio.h>

#include <folly/Portability.h>
#include <folly/Range.h>
#include <folly/portability/SysUio.h>

namespace folly {

class AsyncIOOp : private boost::noncopyable {
  friend class AsyncIO;
  friend std::ostream& operator<<(std::ostream& stream, const AsyncIOOp& o);

 public:
  typedef std::function<void(AsyncIOOp*)> NotificationCallback;

  explicit AsyncIOOp(NotificationCallback cb = NotificationCallback());
  ~AsyncIOOp();

  enum class State {
    UNINITIALIZED,
    INITIALIZED,
    PENDING,
    COMPLETED,
    CANCELED,
  };

  void pread(int fd, void* buf, size_t size, off_t start);
  void pread(int fd, Range<unsigned char*> range, off_t start);
  void preadv(int fd, const iovec* iov, int iovcnt, off_t start);

  void pwrite(int fd, const void* buf, size_t size, off_t start);
  void pwrite(int fd, Range<const unsigned char*> range, off_t start);
  void pwritev(int fd, const iovec* iov, int iovcnt, off_t start);

  State state() const {
    return state_;
  }

  void reset(NotificationCallback cb = NotificationCallback());

  void setNotificationCallback(NotificationCallback cb) {
    cb_ = std::move(cb);
  }
  const NotificationCallback& notificationCallback() const {
    return cb_;
  }

  ssize_t result() const;

 private:
  void init();
  void start();
  void complete(ssize_t result);
  void cancel();

  NotificationCallback cb_;
  iocb iocb_;
  State state_;
  ssize_t result_;
};

std::ostream& operator<<(std::ostream& stream, const AsyncIOOp& o);
std::ostream& operator<<(std::ostream& stream, AsyncIOOp::State state);

class AsyncIO : private boost::noncopyable {
 public:
  typedef AsyncIOOp Op;

  enum PollMode {
    NOT_POLLABLE,
    POLLABLE,
  };

  explicit AsyncIO(size_t capacity, PollMode pollMode = NOT_POLLABLE);
  ~AsyncIO();

  Range<Op**> wait(size_t minRequests);

  Range<Op**> cancel();

  size_t pending() const {
    return pending_;
  }

  size_t capacity() const {
    return capacity_;
  }

  size_t totalSubmits() const {
    return submitted_;
  }

  int pollFd() const {
    return pollFd_;
  }

  Range<Op**> pollCompleted();

  void submit(Op* op);

 private:
  void decrementPending();
  void initializeContext();

  enum class WaitType { COMPLETE, CANCEL };
  Range<AsyncIO::Op**> doWait(
      WaitType type,
      size_t minRequests,
      size_t maxRequests,
      std::vector<Op*>& result);

  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_;
};

class AsyncIOQueue {
 public:
  explicit AsyncIOQueue(AsyncIO* asyncIO);
  ~AsyncIOQueue();

  size_t queued() const {
    return queue_.size();
  }

  void submit(AsyncIOOp* op);

  typedef std::function<AsyncIOOp*()> OpFactory;
  void submit(OpFactory op);

 private:
  void onCompleted(AsyncIOOp* op);
  void maybeDequeue();

  AsyncIO* asyncIO_;

  std::deque<OpFactory> queue_;
};

} // namespace folly