c# - 根据调度程序将 async-await C# 代码转换为 F#

Question

我想知道这是否是一个过于宽泛的问题，但最近我让自己遇到了一段代码，我想确定如何从 C# 转换为正确的 F#。旅程从here (1)开始(TPL-F# 交互的原始问题)，并继续 here (2) (我正在考虑将一些示例代码转换为 F#)。

示例代码太长，这里无法重现，但有趣的功能是 ActivateAsync , RefreshHubs和 AddHub .特别有趣的地方是

AddHub签名为 private async Task AddHub(string address) .

RefreshHubs电话AddHub在循环中收集 tasks 的列表，然后它在最后等待 await Task.WhenAll(tasks)因此返回值匹配其签名 private async Task RefreshHubs(object _) .

RefreshHubs由 ActivateAsync 调用正如 await RefreshHubs(null)然后最后有一个电话await base.ActivateAsync()匹配函数签名 public override async Task ActivateAsync() .

问题:

将此类函数签名正确转换为 F# 的正确转换是什么，该 F# 仍然保持界面和功能并尊重默认的自定义调度程序？而且我也不太确定这种“F# 中的异步/等待”。至于如何“机械地”做到这一点。 :)

原因是在链接“here (1)”中似乎存在问题(我尚未验证这一点)，因为 F# 异步操作不尊重(奥尔良)运行时设置的自定义协作调度程序。另外，它说 here TPL 操作逃脱调度程序并进入任务池，因此禁止使用它们。

我能想到的一种处理方法是使用 F# 函数，如下所示

//Sorry for the inconvenience of shorterned code, for context see the link "here (1)"...
override this.ActivateAsync() =
    this.RegisterTimer(new Func<obj, Task>(this.FlushQueue), null, TimeSpan.FromMilliseconds(100.0), TimeSpan.FromMilliseconds(100.0)) |> ignore

    if RoleEnvironment.IsAvailable then
        this.RefreshHubs(null) |> Async.awaitPlainTask |> Async.RunSynchronously
    else
        this.AddHub("http://localhost:48777/") |> Async.awaitPlainTask |> Async.RunSynchronously

    //Return value comes from here.
    base.ActivateAsync()

member private this.RefreshHubs(_) =
    //Code omitted, in case mor context is needed, take a look at the link "here (2)", sorry for the inconvinience...
    //The return value is Task.
    //In the C# version the AddHub provided tasks are collected and then the
    //on the last line there is return await Task.WhenAll(newHubAdditionTasks) 
    newHubs |> Array.map(fun i -> this.AddHub(i)) |> Task.WhenAll

member private this.AddHub(address) =
    //Code omitted, in case mor context is needed, take a look at the link "here (2)", sorry for the inconvinience...
    //In the C# version:
    //...
    //hubs.Add(address, new Tuple<HubConnection, IHubProxy>(hubConnection, hub))
    //} 
    //so this is "void" and could perhaps be Async<void> in F#... 
    //The return value is Task.
    hubConnection.Start() |> Async.awaitTaskVoid |> Async.RunSynchronously
    TaskDone.Done

startAsPlainTask功能来自 萨沙理发师 来自 here .另一个有趣的选项可能是 here作为

module Async =
    let AwaitTaskVoid : (Task -> Async<unit>) =
        Async.AwaitIAsyncResult >> Async.Ignore

<编辑:我刚刚注意到 Task.WhenAll也需要等待。但正确的方法是什么？呃，该 sleep 了(一个糟糕的双关语)......

<编辑2:在 here (1) (TPL-F# 交互的原始问题)在 Codeplex 中提到 F# 使用同步上下文将工作推送到线程，而 TPL 没有。现在，这是一个合理的解释，我觉得(尽管无论自定义调度程序如何，我仍然在正确翻译这些片段时遇到问题)。一些有趣的附加信息可以从

How to get a Task that uses SynchronizationContext? And how are SynchronizationContext used anyway?

Await, SynchronizationContext, and Console Apps其中一个例子SingleThreadSynchronizationContext提供看起来像排队要执行的工作。也许这应该被使用？

我想我需要提到Hopac在这种情况下，作为一个有趣的切线，还提到我在接下来的 50 多个小时左右无法联系，以防我所有的交叉发布失控。

<编辑 3 :Daniel和 svick在评论中给出很好的建议以使用自定义任务构建器。 Daniel 提供了一个链接，该链接已在 FSharpx 中定义。 .

查看源代码我看到带有参数的接口(interface)定义为

type TaskBuilder(?continuationOptions, ?scheduler, ?cancellationToken) =
    let contOptions = defaultArg continuationOptions TaskContinuationOptions.None
    let scheduler = defaultArg scheduler TaskScheduler.Default
    let cancellationToken = defaultArg cancellationToken CancellationToken.None

如果在奥尔良使用它，它看起来像 TaskScheduler应该是 TaskScheduler.Current根据文档 here

Orleans has it's own task scheduler which provides the single threaded execution model used within grains. It's important that when running tasks the Orleans scheduler is used, and not the .NET thread pool.

Should your grain code require a subtask to be created, you should use Task.Factory.StartNew:

await Task.Factory.StartNew(() =>{ /* logic */ });

This technique will use the current task scheduler, which will be the Orleans scheduler.

You should avoid using Task.Run, which always uses the .NET thread pool, and therefore will not run in the single-threaded execution model.

看起来 TaskScheduler.Current 之间有细微的差别和 TaskScheduler.Default .也许这需要提出一个问题，即在哪些示例情况下会出现不希望有的差异。正如奥尔良文档指出的那样，不要使用 Task.Run而是指向 Task.Factory.StartNew , 我想知道是否应该定义 TaskCreationOptions.DenyAttachChild正如 等权威机构所推荐的那样斯蒂芬·图布 在 Task.Run vs Task.Factory.StartNew和 斯蒂芬·克利里 在 StartNew is Dangerous .嗯，看起来像 .Default将是 .DenyAttachChilld除非我弄错了。

而且，由于Task.Run有问题即Task.Factory.CreateNew关于自定义调度程序，我想知道是否可以通过使用自定义 TaskFactory 来消除这个特定问题。如 Task Scheduler (Task.Factory) and controlling the number of threads 中所述和 How to: Create a Task Scheduler That Limits Concurrency .

嗯，这已经变成了一个相当长的“思考”。我想知道我应该如何关闭这个？也许如果 斯维克和 丹尼尔可以将他们的评论作为答案，我会赞成并接受 斯维克的 ?

Answer 1

您可以使用 TaskBuilder在 FSharpx 中并传入 TaskScheduler.Current .这是我尝试翻译 RefreshHubs .请注意 Task<unit>用于代替 Task .

let RefreshHubs _ =
    let task = TaskBuilder(scheduler = TaskScheduler.Current)
    task {
        let addresses = 
            RoleEnvironment.Roles.["GPSTracker.Web"].Instances
            |> Seq.map (fun instance ->
                let endpoint = instance.InstanceEndpoints.["InternalSignalR"]
                sprintf "http://%O" endpoint.IPEndpoint
            )
            |> Seq.toList

        let newHubs = addresses |> List.filter (not << hubs.ContainsKey)
        let deadHubs = hubs.Keys |> Seq.filter (fun x -> 
            not (List.exists ((=) x) addresses))

        // remove dead hubs
        deadHubs |> Seq.iter (hubs.Remove >> ignore)

        // add new hubs
        let! _ = Task.WhenAll [| for hub in newHubs -> AddHub hub |]
        return ()
    }