make sure we rethrow exceptions in async tasks

[zhengbuqian]

in many places where folly::Future is used, we only wait but not try to get the value. if the async task thrown an exception wait() will unblock but not rethrow it, causing the async thread to crash.

now using folly::collect().get() to make sure we always wait for all futures and rethrow exceptions so callers may catch and handle them.

/kind improvement

[sre-ci-robot]

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: zhengbuqian

The full list of commands accepted by this bot can be found here.

The pull request process is described here

Needs approval from an approver in each of these files:

• OWNERS [zhengbuqian]

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[alexanderguzhva]

No.

Basically, we want to wait for all the futures to finish and only then to process the exception. Alternatively, we could have something that stops further futures from getting spawned if one of them experienced a problem, but this is not really needed.

I've tried the following snippet:

int main() {
  auto pool = ThreadPool::GetGlobalSearchThreadPool(); // say, it can run 4 threads concurrently
  std::vector<folly::Future<folly::Unit>> futures;

  for (size_t i = 0; i < 10; ++i) {
    futures.emplace_back(pool->push([&, id = i]() {
      if (id == 0)
        throw std::runtime_error("foo");

      using namespace std::chrono_literals;
      std::this_thread::sleep_for(2000ms);

      printf("%zd ", id);
    }));
  }

  try {
    folly::collect(futures).get();
  } catch (std::exception &e) {
    std::cout << " caught " << e.what() << std::endl;
  }
  printf(" done\n");
}

And this code may produce the following:

1 2 3  caught foo
 done
4 5 6 7 8 9

or

1 2 3 4 5  caught foo
6  done
7 8 9 

This is wrong, because it means that some futures continue working after the synchronization code folly::collect(futures).get(); finishes.

What we'd like ideally to have is

1 2 3 4 5 6 7 8 9 
caught foo
done

The original snippet

  try {
    for (auto &f : futures)
      f.wait();
    for (auto &f : futures)
      f.result().value();
  } catch (std::exception &e) {
    std::cout << " caught " << e.what() << std::endl;
  }

produces

1 2 3 4 5 6 7 8 9  caught foo
 done

which is a desired behavior.

[alexanderguzhva]

/hold

[zhengbuqian]

thanks @alexanderguzhva for the clarification! I should have used collectAll instead of collect, I thought collect waits for all futures.

[alexanderguzhva]

collectAll works, but possible memory leaks need to be addressed. Please change T* x = new T[N]; into std::vector<T> x(N); or std::unique_ptr<T[]> x(N); whenever possible.

[alexanderguzhva]

a memory leak in case of exception

[zhengbuqian]

TryDiskANNCall catches all exceptions and returns non success Status, thus memory leak won't happen here.

[alexanderguzhva]

memory leaks in case of exception

    if (samples != nullptr)
      delete[] samples;
    if (pq_code != nullptr)
      delete[] pq_code;

[alexanderguzhva]

a memory leak in case of exception

delete[] stats;

[alexanderguzhva]

memory leaks in case of exception

    delete[] distances;
    delete[] center;

[zhengbuqian]

thanks Alex!

[zhengbuqian]

TryDiskANNCall catches all exceptions and returns non success Status, thus memory leak won't happen here.

[alexanderguzhva]

/lgtm

[alexanderguzhva]

/unhold