folly/experimental/io/AsyncIO.h

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