haskell - 并发线程管理器

Question

我有以下并发线程管理器的实现

newtype Query = Query String

type ThreadWorker = (Query, ThreadStatus)
data ThreadStatus = Running | Finished | Threw IOException

newtype ThreadManager = Manager (MVar (M.Map ThreadId (MVar ThreadWorker)

我想编写 manageWorkers::ThreadManager -> IO () 来遍历 Map 并查看 ThreadStatus 的 >ThreadWorker.

如果ThreadWorker完成，那么它就会从ThreadManager中删除。如果抛出异常，则应该对其进行处理(出于示例目的，打印到 stdout 就可以了)并且应该 fork 一个新线程来处理查询(假设存在函数 runQuery::Query -> IO a) 并添加到 ThreadManager 中，否则线程仍在运行，应该保持不变。

我的第一次实现尝试是:

manageWorkers :: ThreadManager -> IO ()
manageWorkers (Manager mgr) =
    modifyMVar mgr $ \m -> do
        m' <- M.traverseWithKey manageWorker m
        return (m', ())
where manageWorker :: ThreadId -> MVar ThreadWorker -> IO (MVar ThreadWorker)
      manageWorker tid wkr = tryTakeMVar wkr >>= \mwkr ->
          case mwkr of
               Just (_, Finished) -> undefined -- need to delete this finished ThreadWorker
               Just (q, Threw e ) -> do
                   putStrLn ("[ERROR] " ++ show e)
                   tid' <- forkIO $ runQuery q
                   undefined -- need to add new ThreadWorker
               Just r -> newMVar r
               _ -> newEmptyMVar

但后来我陷入困境，在 manageWorker 中似乎无法从 ThreadManager 删除或添加。我不确定是否可以从类似于遍历的函数中执行我想要的操作。

是否可以使用我的 ThreadManager 实现此 manageWorkers 函数，或者是否有更好的抽象？

编辑:根据 ThomasM.DuBuisson 使用折叠的建议，我现在有以下内容

manageWorkers (Manager mgr) =
    modifyMVar mgr $ \m ->
    return (M.foldrWithKey manageWorker M.empty m, ())
where manageWorker :: ThreadId -> MVar ThreadWorker -> M.Map ThreadId (MVar ThreadWorker)
                      -> IO (M.Map ThreadId (MVar ThreadWorker))
      manageWorker tid wkr ts = tryTakeMVar wkr >>= \mwkr ->
          case mwkr of
              Just (q, Threw e) -> do
                  putStrLn ("[ERROR] " ++ show e)
                  wkr' <- newEmptyMVar
                  tid' <- forkIO $ runQuery q
                  return $ M.insert tid' wkr' ts
              Just (_, Running) -> return $ M.insert tid wkr
              _ -> return ts

唯一的问题是，显然manageWorker的签名不适用于M.foldrWithKey。我需要一个 M.foldrWithKeyM::Monad m => (k -> a -> b -> m b) -> b -> M.Map k a -> m b 但这样的东西不存在，而我自己创作时遇到了困难。

显然我可以使用unsafePerformIO来转义IO monad并满足编译器的要求，但我只会将其作为最后的手段。在这种情况下使用 unsafePerformIO 有意义吗？

Answer 1

在我看来，你的问题与并发无关。
您只想遍历一个 map ，同时从 map 中删除一些键或在 map 中插入一些键并执行一些IO操作。
问题是 traverseWithKey 无法从 map 中删除键或在 map 中插入键，foldrWithKey 无法执行 IO 操作。
您需要一个 M.foldrWithKeyM::Monad m => (k -> a -> b -> m b) -> b -> M.Map k a -> m b
确实，这样的事情并不存在。但是，如果您查看 foldrWithKey 的文档，其中指出:

foldrWithKey f z == foldr (uncurry f) z . toAscList. 

如果我们将foldr替换为foldM，我们可以猜测可以组成这样的M.foldrWithKeyM。

以下是我的解决方案。

manageWorkers :: ThreadManager -> IO ()
manageWorkers (Manager mgr) = 
    modifyMVar mgr $ \m -> do
        m' <- foldM manageWorker m (M.toList m)
        return (m', ())
  where manageWorker :: M.Map ThreadId (MVar ThreadWorker) -> (ThreadId, MVar ThreadWorker) -> IO (M.Map ThreadId (MVar ThreadWorker))
        manageWorker ts (tid, wkr) = tryTakeMVar wkr >>= \mwkr ->
            case mwkr of
                 Just (_, Finished) -> return $ M.delete tid ts -- need to delete this finished ThreadWorker
                 Just (q, Threw e ) -> do
                       putStrLn ("[ERROR] " ++ show e)
                       wkr' <- newEmptyMVar
                       tid' <- forkIO $ runQuery q
                       return $ M.insert tid' wkr' ts -- need to add new ThreadWorker
                 _ -> return ts