Method readResults

bolt/exec/tests/DriverTest.cpp:179–241

Opens a cursor and reads data. Takes action 'operation' every 'numRows' rows of data. Increments the 'counter' for each successfully read row.

  // Opens a cursor and reads data. Takes action 'operation' every 'numRows'
  // rows of data. Increments the 'counter' for each successfully read row.
  void readResults(
      CursorParameters& params,
      ResultOperation operation,
      int32_t numRows,
      int32_t* counter,
      int32_t threadId = 0) {
    auto cursor = std::make_unique<RowCursor>(params);
    {
      std::lock_guard<std::mutex> l(mutex_);
      tasks_.push_back(cursor->task());
      // To be realized either after 1s wall time or when the corresponding Task
      // is no longer running.
      auto& executor = folly::QueuedImmediateExecutor::instance();
      auto future = tasks_.back()
                        ->taskCompletionFuture()
                        .within(std::chrono::microseconds(1'000'000))
                        .via(&executor);
      stateFutures_.emplace(threadId, std::move(future));

      EXPECT_FALSE(stateFutures_.at(threadId).isReady());
    }
    bool paused = false;
    for (;;) {
      if (operation == ResultOperation::kPause && paused) {
        if (!cursor->hasNext()) {
          paused = false;
          Task::resume(cursor->task());
        }
      }
      if (!cursor->next()) {
        break;
      }
      ++*counter;
      if (*counter % numRows == 0) {
        if (operation == ResultOperation::kDrop) {
          return;
        }
        if (operation == ResultOperation::kReadSlow) {
          std::this_thread::sleep_for(std::chrono::milliseconds(100));
          // If this is an EXPECT this is flaky when running on a
          // noisy test cloud.
          LOG(INFO) << "Task::toString() while probably blocked: "
                    << tasks_[0]->toString();
        } else if (operation == ResultOperation::kCancel) {
          cancelFuture_ = cursor->task()->requestCancel();
        } else if (operation == ResultOperation::kTerminate) {
          cancelFuture_ = cursor->task()->requestAbort();
        } else if (operation == ResultOperation::kYield) {
          if (*counter % 2 == 0) {
            auto time = getCurrentTimeMicro();
            cursor->task()->yieldIfDue(time - 10);
          } else {
            cursor->task()->requestYield();
          }
        } else if (operation == ResultOperation::kPause) {
          auto& executor = folly::QueuedImmediateExecutor::instance();
          auto future = cursor->task()->requestPause().via(&executor);