Haskell 等效于 C# 5 async/await

Question

我刚刚阅读了有关在 C# 5.0 中使用 await 和 async 关键字处理异步函数的新方法。来自C# reference on await的示例：

private async Task SumPageSizesAsync()
{
    // To use the HttpClient type in desktop apps, you must include a using directive and add a 
    // reference for the System.Net.Http namespace.
    HttpClient client = new HttpClient();
    // . . .
    Task<byte[]> getContentsTask = client.GetByteArrayAsync(url);
    byte[] urlContents = await getContentsTask;

    // Equivalently, now that you see how it works, you can write the same thing in a single line.
    //byte[] urlContents = await client.GetByteArrayAsync(url);
    // . . .
}

Task<byte[]> 表示将生成 byte[] 类型值的异步任务的未来。在Task 上使用关键字await 基本上会将函数的其余部分放在一个延续中，该延续将在任务完成时调用。任何使用await 的函数都必须使用关键字async 并具有Task<a> 类型，如果它会返回类型a。

所以线条

byte[] urlContents = await getContentsTask;
// Do something with urlContents

会翻译成类似的东西

Task newTask = getContentsTask.registerContinuation(
               byte[] urlContents => {
                 // Do something with urlContents
               });
return newTask;

---

这感觉很像 Monad（-transformer？）。感觉应该 与 CPS monad 有一些关系，但可能不是。

这是我编写相应 Haskell 类型的尝试

-- The monad that async functions should run in
instance Monad Async
-- The same as the the C# keyword
await         :: Async (Task a) -> Async a
-- Returns the current Task, should wrap what corresponds to
-- a async method in C#.
asyncFunction :: Async a -> Async (Task a)
-- Corresponds to the method Task.Run()
taskRun       :: a -> Task a

以及上面例子的粗略翻译

instance MonadIO Async -- Needed for this example

sumPageSizesAsync :: Async (Task ()) 
sumPageSizesAsync = asyncFunction $ do
    client <- liftIO newHttpClient
    -- client :: HttpClient
    -- ...
    getContentsTask <- getByteArrayAsync client url
    -- getContentsTask :: Task [byte]
    urlContents <- await getContentsTask
    -- urlContents :: [byte]

    -- ...

这会是 Haskell 中的对应类型吗？是否有任何 Haskell 库以这种方式（或类似方式）实现处理异步函数/动作的方式？

另外：你能用 CPS-transformer 构建这个吗？

编辑

是的，Control.Concurrent.Async 模块确实解决了类似的问题（并且具有类似的界面），但以完全不同的方式解决了问题。我猜Control.Monad.Task 会更接近。我正在寻找的（我认为）是 Futures 的一元接口在幕后使用延续传递样式。

Answer 1

这是一个建立在 async 库之上的 Task monad：

import Control.Concurrent.Async (async, wait)

newtype Task a = Task { fork :: IO (IO a) }

newTask :: IO a -> Task a
newTask io = Task $ do
    w <- async io
    return (wait w)

instance Monad Task where
    return a = Task $ return (return a)
    m >>= f  = newTask $ do
        aFut <- fork m
        a    <- aFut
        bFut <- fork (f a)
        bFut

请注意，我没有为此检查过单子定律，所以它可能不正确。

这是定义在后台运行的原始任务的方式：

import Control.Concurrent (threadDelay)

test1 :: Task Int
test1 = newTask $ do
    threadDelay 1000000  -- Wait 1 second
    putStrLn "Hello,"
    return 1

test2 :: Task Int
test2 = newTask $ do
    threadDelay 1000000
    putStrLn " world!"
    return 2

然后您可以使用do 表示法组合Tasks，这会创建一个准备运行的新延迟任务：

test3 :: Task Int
test3 = do
    n1 <- test1
    n2 <- test2
    return (n1 + n2)

运行 fork test3 将生成 Task 并返回一个未来，您可以随时调用它来要求结果，如有必要，阻塞直到完成。

为了证明它有效，我将做两个简单的测试。首先，我将 fork test3 而不要求它的未来，只是为了确保它正确生成复合线程：

main = do
    fork test3
    getLine -- wait without demanding the future

这可以正常工作：

$ ./task
Hello,
 world!
<Enter>
$

现在我们可以测试当我们要求结果时会发生什么：

main = do
    fut <- fork test3
    n   <- fut  -- block until 'test3' is done
    print n

...这也有效：

$ ./task
Hello,
 world!
3
$

Answer 2

monad-par 库提供了spawn 和get 函数，可用于创建类似Future 的计算。您可以将 Par monad 用于将并行运行的纯代码，或者将 ParIO 用于具有副作用的代码。

特别是，我认为您的代码示例可以翻译成：

import Control.Monad.Par.IO

sumPageSizesAsync :: URL -> IO ByteString
sumPageSizesAsync url = runParIO $ do
  task <- spawn $ liftIO $ do client <- newHttpClient
                              return $ getContents url
  urlContents <- get task

如您所见，spawn 负责创建并行运行的代码，并返回一个IVar，稍后get 可以查询它以检索其答案。我认为这两个函数的行为非常匹配async 和await。

有关更多信息，我建议您阅读Haskell 中的并行和并发编程的Par monad chapter。