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"

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

1. 获取任务列表
2. 遍历这些任务以获取其子任务
3. 递归

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

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 中提高效率？我有几个想法：

1. 也许我需要创建一个新的 Monad 来表示这个池化资源访问？
2. 我是否可以急切地计算整个列表（当然，因为有些请求只有在其他请求的结果返回后才能知道）？
3. 我需要显式使用线程吗？

Answer 1

代替

mapM (rFetchTaskGraph token) subtasks

使用

mapConcurrently (rFetchTaskGraph token) subtasks

其中mapConcurrently 来自async 库。

但是，在发出并发 HTTP 请求时，应小心限制它们，以免使远程服务器不堪重负——或被它禁止。一种简单的节流方法是使用semaphore 对rFetchTaskGraph 的所有调用进行门控，如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)

更全面的解决方案将涉及线程池和/或concurrentqueues。

编辑：我认为之前的代码在实践中可能会导致死锁，因为临界区的范围太大。这样的事情应该会更好：

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 请求。