将消息从异步线程发布到 F# 中的主线程

Question

订阅了一个发送日志消息的可观察对象。一些日志消息来自其他线程，因为它们位于 F# 异步块中。我需要能够从主线程中写出消息。

这里是the code，它目前过滤掉了许多日志消息，因为它们不在主线程上：

member x.RegisterTrace() =
    Logging.verbose <- x.Verbose
    let id = Threading.Thread.CurrentThread.ManagedThreadId
    Logging.subscribe (fun trace ->
        if id = Threading.Thread.CurrentThread.ManagedThreadId then
            match trace.Level with
            | TraceLevel.Warning -> x.WriteWarning trace.Text
            | TraceLevel.Error -> x.WriteWarning trace.Text
            | _ -> x.WriteObject trace.Text
        else
            Diagnostics.Debug.Write(sprintf "not on main PS thread: %A" trace)
    )

我有多种使用System.Threading.SynchronizationContent.Current、.SetSynchronizationConent、.Send、.Post的形式。我还涉足System.Threading.Tasks.TaskScheduler.FromCurrentSynchronizationContext。我也试过Async.SwitchToContext。无论我做什么，System.Threading.Thread.CurrentThread.ManagedThreadId 最终都会变得不同，PowerShell 会抱怨。我是不是搞错了？

这是正在进行的工作 pull request 以及有关 the problem 的更多详细信息。

更新时间：太平洋标准时间 2015 年 6 月 16 日星期二上午 11:45

@RCH 谢谢，但使用Async.SwitchToContext 设置SynchronizationContext 似乎不起作用。这是我做Paket-Restore -Force时的代码和调试输出：

member x.RegisterTrace() =
    let a = Thread.CurrentThread.ManagedThreadId
    Logging.verbose <- x.Verbose
    let ctx = SynchronizationContext.Current
    Logging.subscribe (fun trace ->
        let b = Thread.CurrentThread.ManagedThreadId
        async {
            let c = Thread.CurrentThread.ManagedThreadId
            do! Async.SwitchToContext ctx
            let d = Thread.CurrentThread.ManagedThreadId
            Debug.WriteLine (sprintf "%d %d %d %d %s" a b c d trace.Text)
        } |> Async.Start
    )

一位工作专家推荐了另一种解决方案，我将尝试在订阅时传递上下文。

更新时间：太平洋标准时间 2015 年 6 月 16 日星期二下午 5:30

我得到了创建IObservable.SubscribeOn 的帮助，该IObservable.SubscribeOn 允许传入SynchrnonizationContext。不幸的是，它也不能解决问题，但可能是解决方案的一部分。可能需要像 SingleThreadSynchrnonizationContext 这样的自定义 SynchronizationContext。我很乐意帮助制作一个，但在此之前，我将尝试 System.Reactive 的 Observable.ObserveOn(Scheduler.CurrentThread)。

更新时间：太平洋标准时间 2015 年 6 月 16 日星期二晚上 8:30

我也无法让 Rx 工作。 Scheduler.CurrentThread doesn't behave 我希望的方式。然后我尝试了these changes 并且没有调用回调。

member x.RegisterTrace() =
    Logging.verbose <- x.Verbose
    let a = Threading.Thread.CurrentThread.ManagedThreadId
    let ctx = match SynchronizationContext.Current with null -> SynchronizationContext() | sc -> sc
    let sch = SynchronizationContextScheduler ctx
    Logging.event.Publish.ObserveOn sch
    |> Observable.subscribe (fun trace ->
        let b = Threading.Thread.CurrentThread.ManagedThreadId
        Debug.WriteLine(sprintf "%d %d %s" a b trace.Text)

可能需要自定义 SynchronizationContext。 ：/

Answer 1

对于 UI 应用程序（表单、WPF、F# 交互式...），从您的试验中选择 SynchronizationContext.Current 和 Async.SwitchToContext 以及从 Paket 中借用一些代码就足够了。

然而，对于控制台应用程序，没有 SynchronizationContext，因此没有线程可以Posted 到线程，因此它们最终会出现在线程池中。 MSDN Blogs.

仅适用于 UI 应用程序的解决方案：

拥有

module Logging

open System.Diagnostics

type Trace = { Level: TraceLevel; Text: string }

let public event = Event<Trace>()

let subscribe callback = Observable.subscribe callback event.Publish

和

[<AutoOpen>]
module CmdletExt

open System.Diagnostics
open System.Threading

type PSCmdletStandalone() =
    member x.RegisterTrace() =
        let syncContext = SynchronizationContext.Current
        Logging.subscribe (fun trace ->
            async {
                do! Async.SwitchToContext syncContext
                let threadId = Thread.CurrentThread.ManagedThreadId
                match trace.Level with
                | TraceLevel.Warning -> printfn "WARN (on %i): %s" threadId trace.Text
                | TraceLevel.Error -> printfn "ERROR (on %i): %s" threadId trace.Text
                | _ -> printfn "(on %i): %s" threadId trace.Text
            } |> Async.Start // or Async.StartImmediate
        )

然后，在主线程上注册

#load "Logging.fs"
#load "SO.fs"

open System.Diagnostics
open System.Threading
open System

(new PSCmdletStandalone()).RegisterTrace()

printfn "Main: %i" Thread.CurrentThread.ManagedThreadId

for i in Enum.GetValues(typeof<TraceLevel>) do
    async {
        let workerId = Thread.CurrentThread.ManagedThreadId
        do Logging.event.Trigger { Level = unbox i; Text = sprintf "From %i" workerId }
    } |> Async.Start

收益例如

Main: 1
WARN (on 1): From 23
(on 1): From 25
ERROR (on 1): From 22
(on 1): From 13
(on 1): From 14

Answer 2

我最终创建了一个 EventSink，它有一个回调队列，通过 Drain() 在主 PowerShell 线程上执行。我将主要计算放在另一个线程上。 pull request 有完整代码和更多细节。