F# 异步工作流/任务与免费 monad 相结合

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>>>

Answer 1

在我的理解中，自由monad的全部意义在于你不会像Async那样公开效果，所以我认为它们不应该在PipeInstruction 类型中使用。解释器是添加效果的地方。

此外，Free Monad 实际上只在 Haskell 中有意义，您需要做的就是定义一个函子，然后您会自动获得实现的其余部分。在 F# 中，您还必须编写其余代码，因此与更传统的解释器模式相比，使用 Free 没有多大好处。 你链接到的TurtleProgram代码只是一个实验 - 我不建议使用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 monad")：

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

Answer 2

首先，我认为在 F# 中使用自由 monads 非常接近于反模式。这是一个非常抽象的结构，不适合惯用的 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才能使我的函数进行类型检查，但我认为这可能是一个错误的黑客，您可能需要远程处理它。使用这些定义，您可以执行以下操作：

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>您应该能够通过使用作用于命令的递归异步函数来评估这些。