F#异步工作流/任务与自由单子结合

Question

我正在尝试使用自由Monad模式构建消息处理管道，我的代码如下：

module PipeMonad =
type PipeInstruction<'msgIn, 'msgOut, 'a> =
    | HandleAsync of 'msgIn * (Async<'msgOut> -> 'a)
    | SendOutAsync of 'msgOut * (Async -> 'a)

let private mapInstruction f = function
    | HandleAsync (x, next) -> HandleAsync (x, next >> f)
    | SendOutAsync (x, next) -> SendOutAsync (x, next >> f)

type PipeProgram<'msgIn, 'msgOut, 'a> =
    | Act of PipeInstruction<'msgIn, 'msgOut, PipeProgram<'msgIn, 'msgOut, 'a>>
    | Stop of 'a

let rec bind f = function
    | Act x -> x |> mapInstruction (bind f) |> Act
    | Stop x -> f x

type PipeBuilder() =
    member __.Bind (x, f) = bind f x
    member __.Return x = Stop x
    member __.Zero () = Stop ()
    member __.ReturnFrom x = x

let pipe = PipeBuilder()
let handleAsync msgIn = Act (HandleAsync (msgIn, Stop))
let sendOutAsync msgOut = Act (SendOutAsync (msgOut, Stop))

这篇文章是我根据这篇文章编写的。

然而对我来说，很重要的一点是将这些方法异步化（最好使用Task，但Async也可以），但是当我为我的pipeline创建一个构建器时，我不知道如何使用它——我该如何等待一个Task<'msgOut>或者Async<'msgOut>，使其可以发送并等待这个“发送”任务？

现在我有以下这段代码：

let pipeline log msgIn =
    pipe {
        let! msgOut = handleAsync msgIn
        let result = async {
            let! msgOut = msgOut
            log msgOut
            return sendOutAsync msgOut
        }
        return result
    }

这段代码返回 PipeProgram<'b, 'a, Async<PipeProgram<'c, 'a, Async>>>。

- kagetoki

---

你能分享一下你已经写的完整代码（例如 pipe 计算构建器），这样我们就可以尝试运行它吗？ - Tomas Petricek

@TomasPetricek 完成 - kagetoki

在我看来，自由单子的整个重点在于不暴露像Async这样的效果，因此我认为它们不应该在PipeInstruction类型中使用。解释器是添加效果的地方。 - Grundoon

Answer 1

首先，我认为在F#中使用自由单子非常接近反模式。这是一个非常抽象的构造，与惯用的F#风格并不太匹配 - 但这是个人偏好的问题，如果您（和您的团队）发现这种编写代码的方式易于阅读和理解，那么您当然可以朝着这个方向前进。

出于好奇心，我花了一些时间来尝试您的示例 - 尽管我还没有完全弄清楚如何完全修复您的示例，但我希望以下内容可以帮助您朝正确的方向前进。总结一下，我认为您需要将“Async”集成到您的“PipeProgram”中，以便管道程序本身是异步的。

type PipeInstruction<'msgIn, 'msgOut, 'a> =
    | HandleAsync of 'msgIn * (Async<'msgOut> -> 'a)
    | SendOutAsync of 'msgOut * (Async<unit> -> 'a)
    | Continue of 'a 

type PipeProgram<'msgIn, 'msgOut, 'a> =
    | Act of Async<PipeInstruction<'msgIn, 'msgOut, PipeProgram<'msgIn, 'msgOut, 'a>>>
    | Stop of Async<'a>

请注意，我不得不添加Continue以使我的函数通过类型检查，但我认为这可能是一种错误的hack，你可能需要删除它。有了这些定义，您可以执行以下操作：

let private mapInstruction f = function
    | HandleAsync (x, next) -> HandleAsync (x, next >> f)
    | SendOutAsync (x, next) -> SendOutAsync (x, next >> f)
    | Continue v -> Continue v

let rec bind (f:'a -> PipeProgram<_, _, _>) = function
    | Act x -> 
        let w = async { 
          let! x = x 
          return mapInstruction (bind f) x }
        Act w
    | Stop x -> 
        let w = async {
          let! x = x
          let pg = f x
          return Continue pg
        }
        Act w

type PipeBuilder() =
    member __.Bind (x, f) = bind f x
    member __.Return x = Stop x
    member __.Zero () = Stop (async.Return())
    member __.ReturnFrom x = x

let pipe = PipeBuilder()
let handleAsync msgIn = Act (async.Return(HandleAsync (msgIn, Stop)))
let sendOutAsync msgOut = Act (async.Return(SendOutAsync (msgOut, Stop)))

let pipeline log msgIn =
    pipe {
        let! msgOut = handleAsync msgIn
        log msgOut
        return! sendOutAsync msgOut
    }

pipeline ignore 0 

现在，您只需要普通的PipeProgram<int, unit, unit>，您应该能够通过具有递归异步函数的命令来进行评估。

- Tomas Petricek

Answer 2

在我看来，Free Monad 的全部意义在于不暴露像 Async 这样的效果，因此我认为它们不应该在 PipeInstruction 类型中使用。解释器是添加效果的地方。

此外，Free Monad 只在 Haskell 中有意义，你只需要定义一个 functor，然后就可以自动获得其余的实现。在 F# 中，你还必须编写其余的代码，因此与传统的解释器模式相比，使用 Free 没有太大的优势。你链接的 TurtleProgram 代码只是一个实验-我不建议在真正的代码中使用 Free。

最后，如果你已经知道要使用的效果，并且不会有多个解释，那么使用这种方法就没有意义。只有当好处超过复杂性时才有意义。

无论如何，如果你想编写解释器版本（而不是 Free），这是我会做的：

首先，定义指令没有任何效果。

/// The abstract instruction set
module PipeProgram =

    type PipeInstruction<'msgIn, 'msgOut,'state> =
        | Handle of 'msgIn * ('msgOut -> PipeInstruction<'msgIn, 'msgOut,'state>)
        | SendOut of 'msgOut * (unit -> PipeInstruction<'msgIn, 'msgOut,'state>)
        | Stop of 'state

然后，您可以为此编写计算表达式：

/// A computation expression for a PipeProgram
module PipeProgramCE =
    open PipeProgram

    let rec bind f instruction =
        match instruction with
        | Handle (x,next) ->  Handle (x, (next >> bind f))
        | SendOut (x, next) -> SendOut (x, (next >> bind f))
        | Stop x -> f x

    type PipeBuilder() =
        member __.Bind (x, f) = bind f x
        member __.Return x = Stop x
        member __.Zero () = Stop ()
        member __.ReturnFrom x = x

let pipe = PipeProgramCE.PipeBuilder()

然后你可以开始编写计算表达式。这将有助于在你开始解释器之前清楚设计。

// helper functions for CE
let stop x = PipeProgram.Stop x
let handle x = PipeProgram.Handle (x,stop)
let sendOut x  = PipeProgram.SendOut (x, stop)

let exampleProgram : PipeProgram.PipeInstruction<string,string,string> = pipe {
    let! msgOut1 = handle "In1"
    do! sendOut msgOut1
    let! msgOut2 = handle "In2"
    do! sendOut msgOut2
    return msgOut2
    }

一旦你描述了这些指令，就可以编写解释器。如果你不需要编写多个解释器，那么也许根本不需要这样做。

以下是非异步版本的解释器（也就是所谓的“Id单子”）：

module PipeInterpreterSync =
    open PipeProgram

    let handle msgIn =
        printfn "In: %A"  msgIn
        let msgOut = System.Console.ReadLine()
        msgOut

    let sendOut msgOut =
        printfn "Out: %A"  msgOut
        ()

    let rec interpret instruction =
        match instruction with
        | Handle (x, next) ->
            let result = handle x
            result |> next |> interpret
        | SendOut (x, next) ->
            let result = sendOut x
            result |> next |> interpret
        | Stop x ->
            x

这里是异步版本：

module PipeInterpreterAsync =
    open PipeProgram

    /// Implementation of "handle" uses async/IO
    let handleAsync msgIn = async {
        printfn "In: %A"  msgIn
        let msgOut = System.Console.ReadLine()
        return msgOut
        }

    /// Implementation of "sendOut" uses async/IO
    let sendOutAsync msgOut = async {
        printfn "Out: %A"  msgOut
        return ()
        }

    let rec interpret instruction =
        match instruction with
        | Handle (x, next) -> async {
            let! result = handleAsync x
            return! result |> next |> interpret
            }
        | SendOut (x, next) -> async {
            do! sendOutAsync x
            return! () |> next |> interpret
            }
        | Stop x -> x

- Grundoon

---

非常感谢您的出色回答。由于我对函数式模式还不熟悉，您的回答对我整理思路和提高理解非常有帮助。 - kagetoki

Grundoon，当你说不建议在实际代码中使用Free时，是否意味着你对传统的解释器模式没有问题？ - Chechy Levas

是的，如果情况需要，使用解释器是可以的。在 F# 中，使用 Free 作为额外的抽象层没有意义，我个人认为。 - Grundoon