haskell - Cloud Haskell 中的 ManagedProcess 如何工作？

Question

我正在关注这个 tutorial并查看 source code 中的测试用例.我的代码使用 SimplePool.hs在源代码中创建了以下文件:(片段)

sampleTask :: (TimeInterval, String) -> Process String
sampleTask (t, s) = sleep t >> return s 

$(remotable ['sampleTask])

jobTest :: MVar (AsyncResult (Either String String)) -> Process ()
jobTest result = do
  pid <- startTestPool 1 -- start the pool of workers here only one worker
  job <- return $ ($(mkClosure 'sampleTask) (seconds 2, "foobar"))
  -- callAsync put job into pool
  p <- callAsync pid job 
  a <- wait p
  setResult result a
  where
    setResult :: MVar a -> a -> Process ()
    setResult mvar x = liftIO $ putMVar mvar x

startTestPool :: Int -> Process ProcessId
startTestPool s = spawnLocal $ do
  _ <- runPool s
  return ()

runPool :: Int -> Process (Either (InitResult (Pool String)) TerminateReason)
runPool s =
  -- setting a to String
  let s' = poolServer :: ProcessDefinition (Pool String)
  in simplePool s s'

myRemoteTable :: RemoteTable
myRemoteTable = Control.Distributed.Process.Platform.__remoteTable initRemoteTable

main :: IO ()
main = do
  Right (transport, _) <- createTransportExposeInternals
                                    "127.0.0.1" "9901" defaultTCPParameters
  localNode       <- newLocalNode transport myRemoteTable
  result          <- newEmptyMVar
  pid             <- forkProcess localNode $ jobTest result
  ans             <- takeMVar result
  putStrLn $ show pid
  putStrLn $ show ans

运行后出现此错误:

AsyncFailed (DiedException "exit-from=pid://127.0.0.1:9901:0:6")

如果我错了请纠正我，我假设作业没有正确运行，一定是从属进程有问题。 p <- callAsync pid job这行代码我认为是将任务传递给从属进程执行的地方。我查看了 library找到 callAsync 的定义. callAsyncUsing中的关键行是sendTo sid (CallMessage msg (Pid wpid))函数将任务传递给 poolServer。

acceptTask 中的 SimplePool.hs这条线 asyncHandle <- async proc是我认为他们产生一个新进程来执行任务的地方。所以我认为异步进程可能没有完成运行导致调用者过早终止？或者可能是进程没有正确生成？知道调试这个的最佳方法是什么吗？另外，有人可以指出正确的方向以找出如何使 poolSever 跨越不同的节点/不同的计算机(使用 Control.Distributed.Process.Platform.Async.AsyncChan 吗？)？

Answer 1

我稍微修改了您的代码，此代码段包含导入，因此可以编译。确保您使用的是最新的 SimplePool module ，因为您的代码正在使用我找不到的 simplePool，并且您对 runPool 的使用不明确。

{-# LANGUAGE TemplateHaskell #-}

import Control.Concurrent.MVar
import Control.Exception (SomeException)
import Control.Distributed.Process hiding (call)
import Control.Distributed.Process.Closure
import Control.Distributed.Process.Node
import Control.Distributed.Process.Platform hiding (__remoteTable)
import Control.Distributed.Process.Platform.Async
import Control.Distributed.Process.Platform.ManagedProcess
import Control.Distributed.Process.Platform.Test
import Control.Distributed.Process.Platform.Time
import Control.Distributed.Process.Platform.Timer
import Control.Distributed.Process.Serializable()

import Network.Transport
import Network.Transport.TCP

import Data.Binary
import Data.Typeable (Typeable)

import SimplePool hiding (runPool)
import qualified SimplePool (runPool)

sampleTask :: (TimeInterval, String) -> Process String
sampleTask (t, s) = sleep t >> return s

$(remotable ['sampleTask])

jobTest :: MVar (AsyncResult (Either String String)) -> Process ()
jobTest result = do
  pid <- startTestPool 1 -- start the pool of workers here only one worker
  let job = $(mkClosure 'sampleTask) (seconds 2, "foobar")
  -- callAsync put job into pool
  p <- callAsync pid job
  a <- wait p
  setResult result a
  where
    setResult :: MVar a -> a -> Process ()
    setResult mvar x = liftIO $ putMVar mvar x

startTestPool :: Int -> Process ProcessId
startTestPool s = spawnLocal $ do
  _ <- runPool s
  return ()

runPool :: Int -> Process (Either (InitResult (Pool String)) TerminateReason)
runPool = SimplePool.runPool

myRemoteTable :: RemoteTable
myRemoteTable = Main.__remoteTable initRemoteTable

main :: IO ()
main = do

  Right (transport, _) <- createTransportExposeInternals
                                "127.0.0.1" "9901" defaultTCPParameters
  localNode       <- newLocalNode transport myRemoteTable
  result          <- newEmptyMVar
  pid             <- forkProcess localNode $ jobTest result
  ans             <- takeMVar result
  print pid >> print ans

运行这段可编译代码:

$ ./Example 
pid://127.0.0.1:9901:0:3
AsyncDone (Right "foobar")