1

I have some computation intensive tasks, which are now only running on 1 core, so 1/8th of my machine capacity. At the end of each task, I write a log in a file.

What would be the most graceful way to handle this IO using parallel tasks ?

Having my write be itself async ?

Sending messages to an agent who'd process the write sequentially ?

[<Fact>]
let boom () = 
    let tasks = [1 .. 10]
                |> Seq.map (fun components ->  async {  //do compute intensive stuff
                                                        use writer = new StreamWriter("samefile")
                                                        writer.WriteLine "toto" }
           )
    tasks |> Async.Parallel  |> Async.RunSynchronously

Edit

I ended up doing this, and replacing the new Stream in my async to be code by synchronous call to the agent.

[<Fact>]
let pasBoom () = 
    let tasks = [2 .. 2 .. 17]
                |> Seq.map (fun components ->  async {  //do compute intensive stuff
                                                        //use writer = new StreamWriter("samefile")
                                                        use writerhanlde = repoFileHandle.PostAndReply(fun replyChannel -> GetFile(@"samefile", replyChannel))  
                                                        printfn "%A" (writerhanlde.getWriter().ToString())
                                                        writerhanlde.getWriter().WriteLine "toto" }
           )
    tasks |> Async.Parallel  |> Async.RunSynchronously

and the agent (there might be bugs please be careful, I just need something quick myself)

type IDisposableWriter = 
    inherit IDisposable
    abstract getWriter : unit -> StreamWriter


type StreamMessage = | GetFile of string * AsyncReplyChannel<IDisposableWriter>

let repoFileHandle =
    let writerCount = new Dictionary<string, int>()
    let writerRepo  = new Dictionary<string, StreamWriter> ()

    Agent.Start(fun inbox ->
        async { while true do
                    let! msg = inbox.Receive()
                    match msg with
                    | GetFile(filename, reply) -> 
                        if not (writerRepo.ContainsKey(filename)) then
                            writerRepo.[filename]  <- new StreamWriter(filename,true)
                            writerCount.[filename] <- 0
                        writerCount.[filename] <- writerCount.[filename] + 1

                        let obj = {new IDisposableWriter with 
                                    member this.getWriter () = writerRepo.[filename] 
                                    member IDisposable.Dispose() =  
                                        writerCount.[filename] <- writerCount.[filename] - 1                                                                        
                                        if writerCount.[filename] = 0 then
                                            writerRepo.[filename].Dispose()
                                            writerRepo.Remove(filename) |> ignore
                                }
                        reply.Reply(obj) })

and to avoid concurrent write 

  type WriteToStreamMessage = | WriteToStream of string * string

  let fileWriterAgent =
        Agent.Start(fun inbox ->
            async { while true do
                        let! msg = inbox.Receive()
                        match msg with
                        | WriteToStream(filename, content) -> 
                            use writerhanlde = repoFileHandle.PostAndReply(fun replyChannel -> GetFile(filename, replyChannel))
                            writerhanlde.getWriter().WriteLine content
    })
4

1 回答 1

1

您可以更改计算以返回要记录的消息而不是将其写入文件吗？然后你可以PSeq在 PowerPack 中使用，它是 TPL 上的一个瘦包装器：

open Microsoft.FSharp.Collections

let work n = sprintf "running task %d" n
let msgs = PSeq.init 10 work |> PSeq.toList
use writer = System.IO.StreamWriter(@"C:\out.log")
msgs |> List.iter writer.WriteLine
于 2013-03-05T15:49:25.567 回答