在 Haskell 中使用工作池并行运行命令调用超时

Question

我必须编写一个命令行工具，将一些组件粘合在一起进行实验，并且需要帮助来设计符合我要求的代码。

在顶层，我必须处理每个生成的样本——在运行时以及内存消耗中——使用函数“System.Process.readProcessWithExitCode”对另一个程序进行昂贵的调用。因此，您可以想象有一个（昂贵的）函数“genSample :: IO a”，并且您需要该函数的 n 个返回值。

我的要求是： 1. 假设 p 是处理器的数量，那么最多 p 个样本（即对 genSample 的调用）应该并行计算。 2. 应该可以设置一个超时来中止样本的生成。 3. 如果所有样本的计算都超时，则应该停止在 genSample-call 中启动的进程

我当前的解决方案满足要求 1 和 2。对于第三个解决方案，我目前通过执行 killall-command 来帮助自己。这对我来说似乎是一个肮脏的黑客。也许有人有更好的主意？

这是我当前解决方案的核心部分：

import qualified Control.Monad.Par.Class as ParIO
import qualified Control.Monad.Par.IO as ParIO
…
-- | @parRepeatM i n a@ performs action @a@ @n@ times in parallel with timeout @t@
parRepeatM :: ParIO.NFData a =>
              Integer -- ^ timeout in seconds
           -> Integer -- ^ number of duplicates (here: number of req. samples)
           -> IO a    -- ^ action to perform (here: genSample)
           -> IO (Maybe [a])
parRepeatM t n a = timeout t $ ParIO.runParIO $ do
  let tasks = genericReplicate n $ liftIO a -- :: [ParIO a]
  ivars <- mapM ParIO.spawn tasks
  mapM ParIO.get ivars

目前的一个核心问题是，在由于超时而中止后，在 genSample 中调用的命令会继续执行——在最坏的情况下，直到整个 haskell-gluing-program 结束。

Answer 1

在 Haskell 中，取消通常通过异步异常处理。 timeout 似乎就是这么用的。

因此，我们可以尝试在执行外部进程的代码中安装异常处理程序。每当出现异常（异步或非异步）时，处理程序将调用terminateProcess。因为terminateProcess 需要引用 进程句柄，我们将不得不使用createProcess 而不是更高级别的readProcessWithExitCode。

首先，一些导入和辅助功能（我使用的是async 包）：

{-# LANGUAGE ScopedTypeVariables #-}

import Control.Applicative
import Control.Exception 
import Control.Concurrent (threadDelay, MVar, newEmptyMVar, putMVar, takeMVar)
import Control.Concurrent.Async (race_, Concurrently(..), waitEither, withAsync)
import System.Process
import System.Exit
import System.IO
import qualified Data.ByteString as B

-- Executes two actions concurrently and returns the one that finishes first.
-- If an asynchronous exception is thrown, the second action is terminated
-- first.
race' :: IO a -> IO a -> IO a
race' left right =
    withAsync left $ \a ->
    withAsync right $ \b ->
    fmap (either id id) (waitEither a b)

-- terminate external process on exception, ignore if already dead.
terminateCarefully :: ProcessHandle -> IO ()
terminateCarefully pHandle = 
    catch (terminateProcess pHandle) (\(e::IOException) -> return ())

此函数启动一个外部进程并返回其标准输出和退出代码，如果线程被取消则终止该进程：

safeExec :: CreateProcess -> IO (B.ByteString, ExitCode)
safeExec cp = 
    bracketOnError 
        (createProcess cp {std_out = CreatePipe})
        (\(_,_        ,_,pHandle) -> terminateCarefully pHandle)  
        (\(_,Just hOut,_,pHandle) -> do
            -- Workaround for a Windows issue.
            latch <- newEmptyMVar
            race' 
               (do -- IO actions are uninterruptible on Windows :(
                  takeMVar latch 
                  contents <- B.hGetContents hOut 
                  ec <- waitForProcess pHandle
                  pure (contents,ec))
               -- Dummy interruptible action that   
               -- receives asynchronous exceptions first
               -- and helps to end the other action.
               (onException 
                   (do 
                      putMVar latch () 
                      -- runs forever unless interrupted
                      runConcurrently empty)
                   (terminateCarefully pHandle))) 

关于实现：

• bracketOnError用于确保在发生异常时终止外部进程。

• 在 Windows 中，I/O 操作（如从 Handle 读取）是不可中断的（请参阅 https://ghc.haskell.org/trac/ghc/ticket/7353）。这意味着它们不受异步异常的影响。作为一种解决方法，我创建了一个“虚拟”线程，它永远等待（runConcurrently empty）并且可以被异常中断。当它被中断时，它会终止外部进程，导致伴随线程中的读取完成，使得伴随线程再次容易受到异步异常的影响。

• “锁存器”用于防止在安装内部异常处理程序之前对句柄进行任何不可中断的操作。

这有点令人费解，但它似乎有效，至少经过以下测试：

main :: IO ()
main = do
    race_ (safeExec $ proc "calc" []) 
          (threadDelay (3*10^6))

calc 应用程序在三秒后被终止。这是整个gist。

还要记住：

在 Windows 上，如果进程是由 createProcess 用 shell 创建的 shell 命令，或由 runCommand 或 runInteractiveCommand 创建的，则 terminateProcess 只会终止 shell，而不是命令本身。