return range from AsyncIO::cancel(), fix test
[folly.git] / 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