Function TestTaskGroupCancel

cpp/src/arrow/util/task_group_test.cc:118–159

}

void TestTaskGroupCancel(std::shared_ptr<TaskGroup> task_group, StopSource* stop_source) {
  const int NSUCCESSES = 2;
  const int NCANCELS = 20;

  std::atomic<int> count(0);

  auto task_group_was_ok = false;
  task_group->Append([&]() -> Status {
    for (int i = 0; i < NSUCCESSES; ++i) {
      task_group->Append([&]() {
        count++;
        return Status::OK();
      });
    }
    task_group_was_ok = task_group->ok();
    for (int i = 0; i < NCANCELS; ++i) {
      task_group->Append([&]() {
        SleepFor(1e-2);
        stop_source->RequestStop();
        count++;
        return Status::OK();
      });
    }

    return Status::OK();
  });

  // Cancellation is propagated
  ASSERT_RAISES(Cancelled, task_group->Finish());
  ASSERT_TRUE(task_group_was_ok);
  ASSERT_FALSE(task_group->ok());
  if (task_group->parallelism() == 1) {
    // Serial: exactly three successes
    ASSERT_EQ(count.load(), NSUCCESSES + 1);
  } else {
    // Parallel: at least three successes
    ASSERT_GE(count.load(), NSUCCESSES + 1);
    ASSERT_LE(count.load(), NSUCCESSES * task_group->parallelism());
  }
  // Finish() is idempotent
  ASSERT_RAISES(Cancelled, task_group->Finish());
}

class CopyCountingTask {
 public: