可靠地等待多个异步？答案

【问题标题】：Waiting on multiple async's reliably?可靠地等待多个异步？
【发布时间】：2020-01-21 20:55:24
【问题描述】：

我的代码需要触发多个线程并跟踪哪些已完成，哪些仍在运行。我打算使用waitAny 或waitAnyCatch，但被documentation 中的以下内容抛弃了

如果多个 Async 完成或已经完成，则返回的值对应于列表中第一个完成的 Async。

如果真的确实如此，如何可靠地跟踪正在运行/退出的线程？

这是我的简化代码：

chan <- newChan
currentThreadsRef <- newIORef []

-- read jobs from a channel, and run them in parallel asyncs/threads,
-- while adding all threads references to currentThreadsRef
async $ do
  jobArgs <- readChan chan
  jobAsync <- async $ runJob jobArgs
  atomicallyModifyIORef' currentThreadsRef $ \x -> (jobAsync:x, ())

-- wait for jobs to be finished, and remove the thread refernece
-- from currentThreadsRef 
waitForAllJobs currentJobsRef = do
  (readIORef currentJobsRef) >>= \case
    [] -> logDebug "All jobs exited"
    currentJobs -> do
      (exitedJob, jobResult) <- waitAnyCatch currentJobs
      atomicallyModifyIORef currentJobsRef $ \x -> (filter (/= exitedjob) x, ())
      logDebug $ "Job completed with result=" <> show result
      waitForAllJobs currentJobsRef

PS：虽然从我上面的简化代码中可能不明显，但我不能简单地在输入数据上使用mapConcurrently 是有原因的。实际上，async-pool 似乎很适合我的用例，但即便如此，waitAny 也存在同样的问题。

【问题讨论】：

我不明白这个问题。 waitAny 返回已完成的其中一个（您还希望它做什么？），然后将其从列表中删除。然后你递归。只要在运行waitForAllJobs 时没有其他人在currentJobsRef 上写信，我就找不到任何不可靠的原因。（注：我对async不是很熟悉）
"PS：虽然从我上面的简化代码中可能看不出来，但我不能简单地在输入数据上使用mapConcurrently 是有原因的。"你能解释一下这是什么原因吗？
@chi 是的，你是对的。我没有意识到传递给waithAny 的任何Async 是否已经退出，它会立即报告这一点，并且下一次迭代将一次接一个。
@JosephSible-ReinstateMonica 有一个生产者正在生产昂贵的工作，而生产和运行工作都是昂贵的。我们不能同时产生所有作业，也不能同时执行它们。使用标准 async api 可能的唯一解决方案是批量生成作业并在小批量上运行 mapConcurrently。但是，即使这样也是低效的，因为它会等待批次中的每个作业完成，然后再开始下一个批次。

标签： haskell asynchronous concurrency

【解决方案1】：

这是一个启动 1000 个异步的程序，全部设置为在一秒钟内终止，并在一个循环中等待它们。使用ghc -O2 -threaded 编译并使用+RTS -N 运行，它运行大约1.5 秒，并且没有一个异步“丢失”：

import Control.Concurrent
import Control.Concurrent.Async
import qualified Data.Set as Set

main :: IO ()
main = do
  let n = 1000 :: Int
  asyncs0 <- mapM (\i -> async (threadDelay 1000000 >> return i)) [1..n]
  let loop :: Set.Set (Async Int) -> IO ()
      loop asyncs | null asyncs = return ()
                  | otherwise = do
                      (a, _i) <- waitAny (Set.toList asyncs)
                      loop (Set.delete a asyncs)
  loop (Set.fromList asyncs0)

因此，正如评论中提到的那样，文档指的是提供的列表中第一个完成的异步是将被“返回”的，但如果多个异步已经完成，那么其他的不是t“被遗忘”。您只需要从列表中删除返回的 async 并重新轮询，您最终会得到它们。

因此，使用waitAny 等待多个异步应该不会有任何问题。

【讨论】：