Как правильно использовать TryScan в F#

Я не думаю, что метод TryScan поможет вам в этом случае. Это позволяет вам указать тайм-аут, который будет использоваться во время ожидания сообщений. Как только какое-то сообщение будет получено, оно начнет обрабатывать сообщение (игнорируя тайм-аут).

Например, если вы хотите дождаться какого-то определенного сообщения, но выполнять какие-то другие проверки каждую секунду (во время ожидания), вы можете написать:

let loop () = async {
  let! res = mbox.TryScan(function
    | ImportantMessage -> Some(async { 
          // process message 
          return 0
        })
    | _ -> None)
  match res with
  | None -> 
       // perform some check & continue waiting
       return! loop ()
  | Some n -> 
       // ImportantMessage was received and processed 
}

Что можно сделать, чтобы избежать блокировки процессора почтового ящика при обработке сообщения UpdateState? Обработчик почтового ящика (логически) однопоточный — вы, вероятно, не хотите отменять обработку UpdateState сообщения, поэтому лучший вариант — запустить его обработку в фоновом режиме и дождаться завершения обработки. Код, который обрабатывает UpdateState, может затем отправить какое-то сообщение обратно в почтовый ящик (например, UpdateStateCompleted).

Вот набросок, как это может выглядеть:

let rec loop (state) = async {
  let! msg = mbox.Receive()
  match msg with
  | GetState(repl) -> 
      repl.Reply(state)
      return! scanning state
  | UpdateState -> 
      async { 
        // complex calculation (runs in parallel)
        mbox.Post(UpdateStateCompleted newState) }
      |> Async.Start
  | UpdateStateCompleted newState ->
      // Received new state from background workflow
      return! loop newState }

Теперь, когда фоновая задача выполняется параллельно, вам нужно быть осторожным с изменяемым состоянием. Кроме того, если вы отправляете UpdateState сообщений быстрее, чем успеваете их обработать, у вас будут проблемы. Это можно исправить, например, игнорированием или постановкой в ​​очередь запросов, когда вы уже обрабатываете предыдущий.

НЕ ИСПОЛЬЗУЙТЕ TRYSCAN!!!

К сожалению, функция TryScan в текущей версии F# не работает по двум причинам. Во-первых, все дело в том, чтобы указать тайм-аут, но реализация фактически не соблюдает его. В частности, нерелевантные сообщения сбрасывают таймер. Во-вторых, как и в случае с другой функцией Scan, очередь сообщений проверяется под замком, который предотвращает отправку сообщений любым другим потокам на время сканирования, которое может занимать сколь угодно долгое время. Следовательно, сама функция TryScan имеет тенденцию блокировать параллельные системы и даже может создавать тупиковые ситуации, поскольку код вызывающей стороны оценивается внутри блокировки (например, отправка аргумента функции в Scan или TryScan может привести к зависанию агента, когда код под блокировкой блокирует ожидание). приобрести замок, под которым он уже находится).

Я использовал TryScan в раннем прототипе моего производственного кода, и это вызвало множество проблем. Тем не менее, мне удалось спроектировать вокруг него, и получившаяся архитектура была на самом деле лучше. По сути, я с нетерпением Receive все сообщения и фильтрую, используя свою собственную локальную очередь.

Как упомянул Томас, MailboxProcessor является однопоточным. Вам понадобится еще один MailboxProcessor для запуска обновлений в отдельном потоке от получателя состояния.

#nowarn "40"

type Msg = 
    | GetState of AsyncReplyChannel<int> 
    | UpdateState

let runner_UpdateState = MailboxProcessor.Start(fun mbox ->
    let rec loop = async {
        let! state = mbox.Receive()
        printfn "U start processing %d" !state
        // something very time consuming here...
        do! Async.Sleep 100
        printfn "U done processing %d" !state
        state := !state + 1
        do! loop
    }
    loop
)

let mbox = MailboxProcessor.Start(fun mbox ->
    // we need a mutiple state if another thread can change it at any time
    let state = ref 0

    let rec loop = async {
        let! msg = mbox.Receive()

        match msg with
        | UpdateState -> runner_UpdateState.Post state
        | GetState chnl -> chnl.Reply !state

        return! loop 
    }
    loop)

[
    async { 
        for i in 1..10 do 
            mbox.Post UpdateState
            do! Async.Sleep 200
    };
    async { 
        // wait some time so that several `UpdateState` messages are fired
        do! Async.Sleep 1000

        for i in 1..20 do 
            printfn "G %d" (mbox.PostAndReply GetState)
            do! Async.Sleep 50
    }
] 
|> Async.Parallel 
|> Async.RunSynchronously
|> ignore

System.Console.ReadLine() |> ignore

выход:

U start processing 0
U done processing 0
U start processing 1
U done processing 1
U start processing 2
U done processing 2
U start processing 3
U done processing 3
U start processing 4
U done processing 4
G 5
U start processing 5
G 5
U done processing 5
G 5
G 6
U start processing 6
G 6
G 6
U done processing 6
G 7
U start processing 7
G 7
G 7
U done processing 7
G 8
G U start processing 8
8
G 8
U done processing 8
G 9
G 9
U start processing 9
G 9
U done processing 9
G 9
G 10
G 10
G 10
G 10

Вы также можете использовать ThreadPool.

open System.Threading

type Msg = 
    | GetState of AsyncReplyChannel<int> 
    | SetState of int
    | UpdateState

let mbox = MailboxProcessor.Start(fun mbox ->
    let rec loop state = async {
        let! msg = mbox.Receive()

        match msg with
        | UpdateState -> 
            ThreadPool.QueueUserWorkItem((fun obj -> 
                let state = obj :?> int

                printfn "U start processing %d" state
                Async.Sleep 100 |> Async.RunSynchronously
                printfn "U done processing %d" state
                mbox.Post(SetState(state + 1))

                ), state)
            |> ignore
        | GetState chnl -> 
            chnl.Reply state
        | SetState newState ->
            return! loop newState
        return! loop state
    }
    loop 0)

[
    async { 
        for i in 1..10 do 
            mbox.Post UpdateState
            do! Async.Sleep 200
    };
    async { 
        // wait some time so that several `UpdateState` messages are fired
        do! Async.Sleep 1000

        for i in 1..20 do 
            printfn "G %d" (mbox.PostAndReply GetState)
            do! Async.Sleep 50
    }
] 
|> Async.Parallel 
|> Async.RunSynchronously
|> ignore

System.Console.ReadLine() |> игнорировать