folly/experimental/io/AsyncIO.h

   1 /*
   2  * Copyright 2017 Facebook, Inc.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *   http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 #pragma once
  18
  19 #include <sys/types.h>
  20 #include <libaio.h>
  21
  22 #include <atomic>
  23 #include <cstdint>
  24 #include <deque>
  25 #include <functional>
  26 #include <iosfwd>
  27 #include <mutex>
  28 #include <utility>
  29 #include <vector>
  30
  31 #include <boost/noncopyable.hpp>
  32
  33 #include <folly/Portability.h>
  34 #include <folly/Range.h>
  35 #include <folly/portability/SysUio.h>
  36
  37 namespace folly {
  38
  39 /**
  40  * An AsyncIOOp represents a pending operation.  You may set a notification
  41  * callback or you may use this class's methods directly.
  42  *
  43  * The op must remain allocated until it is completed or canceled.
  44  */
  45 class AsyncIOOp : private boost::noncopyable {
  46   friend class AsyncIO;
  47   friend std::ostream& operator<<(std::ostream& stream, const AsyncIOOp& o);
  48
  49  public:
  50   typedef std::function<void(AsyncIOOp*)> NotificationCallback;
  51
  52   explicit AsyncIOOp(NotificationCallback cb = NotificationCallback());
  53   ~AsyncIOOp();
  54
  55   enum class State {
  56     UNINITIALIZED,
  57     INITIALIZED,
  58     PENDING,
  59     COMPLETED,
  60     CANCELED,
  61   };
  62
  63   /**
  64    * Initiate a read request.
  65    */
  66   void pread(int fd, void* buf, size_t size, off_t start);
  67   void pread(int fd, Range<unsigned char*> range, off_t start);
  68   void preadv(int fd, const iovec* iov, int iovcnt, off_t start);
  69
  70   /**
  71    * Initiate a write request.
  72    */
  73   void pwrite(int fd, const void* buf, size_t size, off_t start);
  74   void pwrite(int fd, Range<const unsigned char*> range, off_t start);
  75   void pwritev(int fd, const iovec* iov, int iovcnt, off_t start);
  76
  77   /**
  78    * Return the current operation state.
  79    */
  80   State state() const { return state_; }
  81
  82   /**
  83    * Reset the operation for reuse.  It is an error to call reset() on
  84    * an Op that is still pending.
  85    */
  86   void reset(NotificationCallback cb = NotificationCallback());
  87
  88   void setNotificationCallback(NotificationCallback cb) { cb_ = std::move(cb); }
  89   const NotificationCallback& notificationCallback() const { return cb_; }
  90
  91   /**
  92    * Retrieve the result of this operation.  Returns >=0 on success,
  93    * -errno on failure (that is, using the Linux kernel error reporting
  94    * conventions).  Use checkKernelError (folly/Exception.h) on the result to
  95    * throw a std::system_error in case of error instead.
  96    *
  97    * It is an error to call this if the Op hasn't completed.
  98    */
  99   ssize_t result() const;
 100
 101  private:
 102   void init();
 103   void start();
 104   void complete(ssize_t result);
 105   void cancel();
 106
 107   NotificationCallback cb_;
 108   iocb iocb_;
 109   State state_;
 110   ssize_t result_;
 111 };
 112
 113 std::ostream& operator<<(std::ostream& stream, const AsyncIOOp& o);
 114 std::ostream& operator<<(std::ostream& stream, AsyncIOOp::State state);
 115
 116 /**
 117  * C++ interface around Linux Async IO.
 118  */
 119 class AsyncIO : private boost::noncopyable {
 120  public:
 121   typedef AsyncIOOp Op;
 122
 123   enum PollMode {
 124     NOT_POLLABLE,
 125     POLLABLE,
 126   };
 127
 128   /**
 129    * Create an AsyncIO context capable of holding at most 'capacity' pending
 130    * requests at the same time.  As requests complete, others can be scheduled,
 131    * as long as this limit is not exceeded.
 132    *
 133    * Note: the maximum number of allowed concurrent requests is controlled
 134    * by the fs.aio-max-nr sysctl, the default value is usually 64K.
 135    *
 136    * If pollMode is POLLABLE, pollFd() will return a file descriptor that
 137    * can be passed to poll / epoll / select and will become readable when
 138    * any IOs on this AsyncIO have completed.  If you do this, you must use
 139    * pollCompleted() instead of wait() -- do not read from the pollFd()
 140    * file descriptor directly.
 141    *
 142    * You may use the same AsyncIO object from multiple threads, as long as
 143    * there is only one concurrent caller of wait() / pollCompleted() / cancel()
 144    * (perhaps by always calling it from the same thread, or by providing
 145    * appropriate mutual exclusion).  In this case, pending() returns a snapshot
 146    * of the current number of pending requests.
 147    */
 148   explicit AsyncIO(size_t capacity, PollMode pollMode = NOT_POLLABLE);
 149   ~AsyncIO();
 150
 151   /**
 152    * Wait for at least minRequests to complete.  Returns the requests that
 153    * have completed; the returned range is valid until the next call to
 154    * wait().  minRequests may be 0 to not block.
 155    */
 156   Range<Op**> wait(size_t minRequests);
 157
 158   /**
 159    * Cancel all pending requests and return them; the returned range is
 160    * valid until the next call to cancel().
 161    */
 162   Range<Op**> cancel();
 163
 164   /**
 165    * Return the number of pending requests.
 166    */
 167   size_t pending() const { return pending_; }
 168
 169   /**
 170    * Return the maximum number of requests that can be kept outstanding
 171    * at any one time.
 172    */
 173   size_t capacity() const { return capacity_; }
 174
 175   /**
 176    * Return the accumulative number of submitted I/O, since this object
 177    * has been created.
 178    */
 179   size_t totalSubmits() const { return submitted_; }
 180
 181   /**
 182    * If POLLABLE, return a file descriptor that can be passed to poll / epoll
 183    * and will become readable when any async IO operations have completed.
 184    * If NOT_POLLABLE, return -1.
 185    */
 186   int pollFd() const { return pollFd_; }
 187
 188   /**
 189    * If POLLABLE, call instead of wait after the file descriptor returned
 190    * by pollFd() became readable.  The returned range is valid until the next
 191    * call to pollCompleted().
 192    */
 193   Range<Op**> pollCompleted();
 194
 195   /**
 196    * Submit an op for execution.
 197    */
 198   void submit(Op* op);
 199
 200  private:
 201   void decrementPending();
 202   void initializeContext();
 203
 204   enum class WaitType { COMPLETE, CANCEL };
 205   Range<AsyncIO::Op**> doWait(
 206       WaitType type,
 207       size_t minRequests,
 208       size_t maxRequests,
 209       std::vector<Op*>& result);
 210
 211   io_context_t ctx_{nullptr};
 212   std::atomic<bool> ctxSet_{false};
 213   std::mutex initMutex_;
 214
 215   std::atomic<size_t> pending_{0};
 216   std::atomic<size_t> submitted_{0};
 217   const size_t capacity_;
 218   int pollFd_{-1};
 219   std::vector<Op*> completed_;
 220   std::vector<Op*> canceled_;
 221 };
 222
 223 /**
 224  * Wrapper around AsyncIO that allows you to schedule more requests than
 225  * the AsyncIO's object capacity.  Other requests are queued and processed
 226  * in a FIFO order.
 227  */
 228 class AsyncIOQueue {
 229  public:
 230   /**
 231    * Create a queue, using the given AsyncIO object.
 232    * The AsyncIO object may not be used by anything else until the
 233    * queue is destroyed.
 234    */
 235   explicit AsyncIOQueue(AsyncIO* asyncIO);
 236   ~AsyncIOQueue();
 237
 238   size_t queued() const { return queue_.size(); }
 239
 240   /**
 241    * Submit an op to the AsyncIO queue.  The op will be queued until
 242    * the AsyncIO object has room.
 243    */
 244   void submit(AsyncIOOp* op);
 245
 246   /**
 247    * Submit a delayed op to the AsyncIO queue; this allows you to postpone
 248    * creation of the Op (which may require allocating memory, etc) until
 249    * the AsyncIO object has room.
 250    */
 251   typedef std::function<AsyncIOOp*()> OpFactory;
 252   void submit(OpFactory op);
 253
 254  private:
 255   void onCompleted(AsyncIOOp* op);
 256   void maybeDequeue();
 257
 258   AsyncIO* asyncIO_;
 259
 260   std::deque<OpFactory> queue_;
 261 };
 262
 263 }  // namespace folly