haskell - Haskell 中的并发 HTTP 请求

Question

我有一组函数，旨在从 Asana API 构建子任务树。为此，我有一个名为“Asana.hs”的相当简单的模块，其中最重要的两个功能是使用 Network.HTTP.Simple 的功能。执行请求:

getTasksForProject :: String -> String -> IO [Task]
getTasksForProject token projectId = getFromAsana token $ "projects/" ++ projectId ++ "/tasks"

getSubtasks :: String -> String -> IO [Task]
getSubtasks token taskId = getFromAsana token $ "tasks/" ++ taskId ++ "/subtasks"

问题是当我想构建我必须完成的所有任务的图表时:

获取任务列表

遍历这些任务以获得它们的子任务

递归

例如，我有这些函数来构建节点和边的“图”:

type TaskGraph = ([Task], [Edge])

merge :: TaskGraph -> TaskGraph -> TaskGraph
merge (aTasks, aEdges) (bTasks, bEdges) = (aTasks ++ bTasks, aEdges ++ bEdges)

makeEdge :: Relation -> Task -> Task -> Edge
makeEdge rel parent child = Edge rel (taskId parent) (taskId child)

rFetchTaskGraph :: String -> Task -> IO TaskGraph
rFetchTaskGraph token task = do
  subtasks <- getSubtasks token $ taskId task
  let edges = map (makeEdge Subtask task) subtasks
  foldr merge ([task], edges) <$> mapM (rFetchTaskGraph token) subtasks

这非常慢，因为据我所知，它按顺序发出每个 HTTP 请求。如果我用 Javascript 之类的东西执行此操作，Promises 将允许我急切地执行所有计算，但将请求排队，因此仅在请求完成时才解析相关的 Promise，但将并行性集中到某种连接池管理器中。
如何在 Haskell 中提高效率？我有几个想法:

也许我需要创建一个新的 Monad 来表示这个池化资源访问？

我是否可以急切地计算整个列表(当然，因为有些请求只有在其他请求的结果返回后才能知道)？

我需要明确使用线程吗？

Answer 1

代替

mapM (rFetchTaskGraph token) subtasks

采用

mapConcurrently (rFetchTaskGraph token) subtasks

在哪里 mapConcurrently 来自 async图书馆。
但是，在发出并发 HTTP 请求时，应小心限制它们，以免使远程服务器不堪重负——或被它禁止。一种简单的节流方法是对 rFetchTaskGraph 的所有调用进行门控。使用 semaphore ，如 this SO answer 中所述.
因为 rFetchTaskGraph是递归的，它应该接受信号量作为参数，以便将其传递给它的子调用:

rFetchTaskGraph :: QSem -> String -> Task -> IO TaskGraph
rFetchTaskGraph sem token task = 
    bracket_ 
      (waitQSem sem) 
      (signalQSem sem)
      (do
        subtasks <- getSubtasks token $ taskId task
        let edges = map (makeEdge Subtask task) subtasks
        foldr merge ([task], edges) <$> mapConcurrently (rFetchTaskGraph sem token) subtasks)

更全面的解决方案将涉及线程池和/或 concurrent queues .
编辑:我认为前面的代码在实践中可能会导致死锁，因为临界区的范围太大。这样的事情应该会更好:

rFetchTaskGraph sem token task = do
       subtasks <- bracket_ (waitQSem sem) (signalQSem sem) $ getSubtasks token $ taskId task
       let edges = map (makeEdge Subtask task) subtasks
       foldr merge ([task], edges) <$> mapConcurrently (rFetchTaskGraph sem token) subtasks 

也就是说，将临界区仅限于实际的 HTTP 请求。