fix raciness in AsyncBase::submit

Summary:
AsyncBase::submit() is racy: it calls submitOne() and then later calls op->start(). But
op can be completed between return from submit and call to start(). In this case a CHECK
will fail somewhere, either because we see op as COMPLETED in start(), or because we
see op as INITALIZED in complete().

Call start() to put ops into PENDING state *before* issuing them; add "unstart" method
to put them back into INITIALIZED state when submit fails.

In passing, fix bug: if submit submits fewer ops than expected, decrement pending count
properly.

Differential Revision: D24094495

fbshipit-source-id: 2508dbd7ba5d306ceba34d62471d957f6cb20ff5

folly/experimental/io/AsyncBase.cpp

@@ -52,6 +52,11 @@ void AsyncBaseOp::start() {
   state_ = State::PENDING;
 }
 
+void AsyncBaseOp::unstart() {
+  DCHECK_EQ(state_, State::PENDING);
+  state_ = State::INITIALIZED;
+}
+
 void AsyncBaseOp::complete(ssize_t result) {
   DCHECK_EQ(state_, State::PENDING);
   state_ = State::COMPLETED;
@@ -117,20 +122,24 @@ void AsyncBase::submit(Op* op) {
     throw std::range_error("AsyncBase: too many pending requests");
   }
 
+  op->start();
   int rc = submitOne(op);
 
-  if (rc < 0) {
+  if (rc <= 0) {
+    op->unstart();
     decrementPending();
+    if (rc < 0) {
       throwSystemErrorExplicit(-rc, "AsyncBase: io_submit failed");
     }
-  submitted_++;
+  }
+  submitted_ += rc;
   DCHECK_EQ(rc, 1);
-  op->start();
 }
 
 int AsyncBase::submit(Range<Op**> ops) {
   for (auto& op : ops) {
     CHECK_EQ(op->state(), Op::State::INITIALIZED);
+    op->start();
   }
   initializeContext(); // on demand
 
@@ -147,11 +156,14 @@ int AsyncBase::submit(Range<Op**> ops) {
     decrementPending(ops.size());
     throwSystemErrorExplicit(-rc, "AsyncBase: io_submit failed");
   }
+  // Any ops that did not get submitted go back to INITIALIZED state
+  // and are removed from pending count.
+  for (size_t i = rc; i < ops.size(); i++) {
+    ops[i]->unstart();
+    decrementPending(1);
+  }
   submitted_ += rc;
   DCHECK_LE(rc, ops.size());
-  for (int i = 0; i < rc; i++) {
-    ops[i]->start();
-  }
 
   return rc;
 }

folly/experimental/io/AsyncBase.h

@@ -148,6 +148,7 @@ class AsyncBaseOp {
  protected:
   void init();
   void start();
+  void unstart();
   void complete(ssize_t result);
   void cancel();