AsyncIO::cancel

[folly.git] / folly / experimental / io / AsyncIO.h

diff --git a/folly/experimental/io/AsyncIO.h b/folly/experimental/io/AsyncIO.h
index faa5e27fc4b4e075d36f023bba1e5d72cae15dcc..6702c53fda07ab6423c53cd3fbe80b19a3b934fe 100644 (file)
--- a/folly/experimental/io/AsyncIO.h
+++ b/folly/experimental/io/AsyncIO.h
@@ -40,25 +40,24 @@ namespace folly {
  * An AsyncIOOp represents a pending operation.  You may set a notification
  * callback or you may use this class's methods directly.
  *
- * The op must remain allocated until completion.
+ * The op must remain allocated until it is completed or canceled.
  */
 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();
 
-  // There would be a cancel() method here if Linux AIO actually implemented
-  // it.  But let's not get your hopes up.
-
   enum class State {
     UNINITIALIZED,
     INITIALIZED,
     PENDING,
-    COMPLETED
+    COMPLETED,
+    CANCELED,
   };
 
   /**
@@ -95,8 +94,7 @@ class AsyncIOOp : private boost::noncopyable {
    * conventions).  Use checkKernelError (folly/Exception.h) on the result to
    * throw a std::system_error in case of error instead.
    *
-   * It is an error to call this if the Op hasn't yet started or is still
-   * pending.
+   * It is an error to call this if the Op hasn't completed.
    */
   ssize_t result() const;
 
@@ -104,6 +102,7 @@ class AsyncIOOp : private boost::noncopyable {
   void init();
   void start();
   void complete(ssize_t result);
+  void cancel();
 
   NotificationCallback cb_;
   iocb iocb_;
@@ -123,7 +122,7 @@ class AsyncIO : private boost::noncopyable {
 
   enum PollMode {
     NOT_POLLABLE,
-    POLLABLE
+    POLLABLE,
   };
 
   /**
@@ -141,12 +140,12 @@ class AsyncIO : private boost::noncopyable {
    * 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() (perhaps
-   * by always calling it from the same thread, or by providing appropriate
-   * mutual exclusion)  In this case, pending() returns a snapshot
+   * 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.
    */
-  explicit AsyncIO(size_t capacity, PollMode pollMode=NOT_POLLABLE);
+  explicit AsyncIO(size_t capacity, PollMode pollMode = NOT_POLLABLE);
   ~AsyncIO();
 
   /**
@@ -156,6 +155,11 @@ class AsyncIO : private boost::noncopyable {
    */
   Range<Op**> wait(size_t minRequests);
 
+  /**
+   * Cancel all pending requests and return their number.
+   */
+  size_t cancel();
+
   /**
    * Return the number of pending requests.
    */
@@ -196,16 +200,21 @@ class AsyncIO : private boost::noncopyable {
   void decrementPending();
   void initializeContext();
 
-  Range<Op**> doWait(size_t minRequests, size_t maxRequests);
+  enum class WaitType { COMPLETE, CANCEL };
+  void doWait(
+      WaitType type,
+      size_t minRequests,
+      size_t maxRequests,
+      std::vector<Op*>* result);
 
-  io_context_t ctx_;
-  std::atomic<bool> ctxSet_;
+  io_context_t ctx_{nullptr};
+  std::atomic<bool> ctxSet_{false};
   std::mutex initMutex_;
 
-  std::atomic<size_t> pending_;
-  std::atomic<size_t> submitted_;
+  std::atomic<size_t> pending_{0};
+  std::atomic<size_t> submitted_{0};
   const size_t capacity_;
-  int pollFd_;
+  int pollFd_{-1};
   std::vector<Op*> completed_;
 };