Function ASSERT_OK_AND_ASSIGN

cpp/src/arrow/util/thread_pool_test.cc:856–1050

    // Reference passing
    std::string s = "foo";
    ASSERT_OK_AND_ASSIGN(auto fut,
                         pool->Submit(inplace_add<std::string>, std::ref(s), "bar"));
    ASSERT_OK_AND_EQ("foobar", fut.result());
    ASSERT_EQ(s, "foobar");
  }
  {
    // `void` return type
    ASSERT_OK_AND_ASSIGN(auto fut, pool->Submit(SleepFor, 0.001));
    ASSERT_OK(fut.status());
  }
}

TEST_F(TestThreadPool, SubmitWithStopToken) {
  auto pool = this->MakeThreadPool(3);
  {
    StopSource stop_source;
    ASSERT_OK_AND_ASSIGN(Future<int> fut,
                         pool->Submit(stop_source.token(), add<int>, 4, 5));
    Result<int> res = fut.result();
    ASSERT_OK_AND_EQ(9, res);
  }
}

TEST_F(TestThreadPool, SubmitWithStopTokenCancelled) {
  auto pool = this->MakeThreadPool(3);
  {
    const int n_futures = 100;
    StopSource stop_source;
    StopToken stop_token = stop_source.token();
    std::vector<Future<int>> futures;
    for (int i = 0; i < n_futures; ++i) {
      ASSERT_OK_AND_ASSIGN(
          auto fut, pool->Submit(stop_token, slow_add<int>, 0.01 /*seconds*/, i, 1));
      futures.push_back(std::move(fut));
    }
    SleepFor(0.05);  // Let some work finish
    stop_source.RequestStop();
    int n_success = 0;
    int n_cancelled = 0;
    for (int i = 0; i < n_futures; ++i) {
      Result<int> res = futures[i].result();
      if (res.ok()) {
        ASSERT_EQ(i + 1, *res);
        ++n_success;
      } else {
        ASSERT_RAISES(Cancelled, res);
        ++n_cancelled;
      }
    }
    ASSERT_GT(n_success, 0);
    ASSERT_GT(n_cancelled, 0);
  }
}

// Test fork safety on Unix

#if !(defined(_WIN32) || defined(ARROW_VALGRIND) || defined(ADDRESS_SANITIZER) || \
      defined(THREAD_SANITIZER))