Implementare CCR Interleave Arbiter in F#

Question

Voglio implementare il concetto di porto del framework CCR in F# (poiché il CCR non è ufficialmente supportato per .NET 4.0). So che si può usare il Mailboxprocessor Classe in F# per farlo. Funziona perfettamente per semplice Ricevere Arbitri ma ho bisogno del concetto di Interleave Arbiter, cioè voglio controllare quali messaggi vengono elaborati esclusivamente e quali vengono elaborati contemporaneamente. Finora non ho idea di implementarlo in F# e sarei grato per il tuo aiuto.

Answer 1

Non ho molta familiarità con il CCR, ma cercherò di rispondere - la mia comprensione è quella arbitro interleave si comporta un po 'come ReaderWriterLock. Cioè, è possibile specificare alcune operazioni che possono essere eseguite in parallelo (letture) e alcune operazioni esclusive (scrive).

Il seguente agente è un modo per implementarlo (non testato, ma i controlli del tipo :-)). L'agente espone due operazioni destinate all'uso pubblico. L'ultimo è interno:

type Message<'T> =
  | PerformReadOperation of ('T -> Async<unit>)
  | PerformWriteOperation of ('T -> Async<'T>)
  | ReadOperationCompleted

• Inviando l'agente PerformReadOperation, gli stai dando un'operazione che dovrebbe essere eseguita (una volta) usando lo stato e possibilmente in parallelo con altre operazioni di lettura.

• Inviando l'agente PerformWriteOperation, gli stai dando un'operazione che calcola un nuovo stato e deve essere eseguito dopo che tutte le operazioni di lettura sono complete. (Se stavi lavorando con uno stato immutabile, ciò renderebbe le cose più semplici - non dovresti aspettare fino al completamento dei lettori! Ma l'implementazione di seguito implementa l'attesa).

L'agente inizia con uno stato iniziale:

let initial = // initial state

E il resto dell'agente è implementato utilizzando due loop:

let interleaver = MailboxProcessor.Start(fun mbox ->

  // Asynchronously wait until all read operations complete
  let rec waitUntilReadsComplete reads = 
    if reads = 0 then async { return () }
    else mbox.Scan(fun msg ->
      match msg with
      | ReadOperationCompleted -> Some(waitUntilReadsComplete (reads - 1))
      | _ -> None)

  let rec readingLoop state reads = async {
    let! msg = mbox.Receive()
    match msg with
    | ReadOperationCompleted ->
        // Some read operation completed - decrement counter
        return! readingLoop state (reads - 1) 
    | PerformWriteOperation(op) ->
        do! waitUntilReadsComplete reads
        let! newState = op state
        return! readingLoop newState 0
    | PerformReadOperation(op) ->
        // Start the operation in background & increment counter
        async { do! op state
                mbox.Post(ReadOperationCompleted) }
        |> Async.Start
        return! readingLoop state (reads + 1) }
  readingLoop initial 0)

Answer 2

Solo per aggiungere una soluzione suggerita da Tomas, nel caso in cui non si desideri esporre il messaggio "ReadOPeration Completed" al consumatore della casella di posta (poiché questo messaggio è interno e nell'implementazione corrente può essere inviata da qualsiasi consumatore della casella di posta) Una casella di posta separata può essere creata all'interno della funzione Processore della casella di posta principale che accetterà due messaggi: readoperation completeted e waitforeadCompleted (questa verrà utilizzata con postandasyncreply dalla casella di posta principale) poiché la risposta a questo messaggio arriverà solo quando tutta la lettura Le operazioni sono completate. Anche il conteggio di "lettura" rappresentato da "letture" verrà spostato in questa nuova casella di posta interna poiché quello stato sarà incapsulato da questa casella di posta interna.