fix: Fix the async direct coalesce load memory leak in case of early task failure in pyspark

[xiaoxmeng]

Summary:
Here is the race condition can cause memory leak in race between async direct coalesce load and early task failure:
T1. file reader triggers stripe load which does coalesce prefetch
T2. coalesce prefetch kicks off and pass the cancellation check
T3. coalesce load does the memory allocations and do read from storage
T4. before table scan do on-demand load, the task fails or prefetch data is skipped by filtering and task finishes.
T5. task destruction frees the memory pool hits the memory leak check failure.
T6. if we disable the memory leak check failure (with memory leak metric reporting in Meta production), then
the buffer free will throw with bad memory pool pointer.

Verified the fix with unit test that reproduce the race condition. This is exposed by pyspark use case.

Differential Revision: D71529929

[facebook-github-bot]

This pull request was exported from Phabricator. Differential Revision: D71529929

[netlify]

✅ Deploy Preview for meta-velox canceled.

Name	Link
🔨 Latest commit	ebd682c
🔍 Latest deploy log	https://app.netlify.com/sites/meta-velox/deploys/67dbba86f0e6c600085bfe1d

[facebook-github-bot]

This pull request was exported from Phabricator. Differential Revision: D71529929

[tanjialiang]

Thanks for the quick fix

[facebook-github-bot]

This pull request has been merged in 48b9803.

[conbench-facebook]

Conbench analyzed the 1 benchmark run on commit 48b98030.

There were no benchmark performance regressions. 🎉

The full Conbench report has more details.

[fzhedu]

@xiaoxmeng
why not let the main thread wait all IO threads to be finished? one possible case happen that the main thread ~memoryManager, but the scan pool is still held by the IO threads, causing core dump.

  ~DirectBufferedInput() override {
    std::vector<folly::Future<bool>> futures;
    for (auto& load : coalescedLoads_) {
      load->cancel();
      futures.push_back(load->getCompletionFuture());
    }
    if (!futures.empty()) {
      LOG(WARNING) << "wait " << futures.size() << " loads to be completed";
      folly::collectAll(futures).wait();
    }

[j7nhai]

@xiaoxmeng why not let the main thread wait all IO threads to be finished? one possible case happen that the main thread ~memoryManager, but the scan pool is still held by the IO threads, causing core dump.

  ~DirectBufferedInput() override {
    std::vector<folly::Future<bool>> futures;
    for (auto& load : coalescedLoads_) {
      load->cancel();
      futures.push_back(load->getCompletionFuture());
    }
    if (!futures.empty()) {
      LOG(WARNING) << "wait " << futures.size() << " loads to be completed";
      folly::collectAll(futures).wait();
    }

Actually, the core dump is caused by memory leak. refer to my PR fix: #14722