7 people like it.
Like the snippet!
Throttling agent
Implements a simple agent that lets you throttle the degree of parallelism by limiting the number of work items that are processed in parallel.
1:
2:
3:
4:
5:
6:
7:
8:
9:
10:
11:
12:
13:
14:
15:
16:
17:
18:
19:
20:
21:
22:
23:
24:
25:
26:
27:
28:
29:
30:
|
// We can ask the agent to enqueue a new work item;
// and the agent sends itself a completed notification
type ThrottlingMessage =
| Enqueue of Async<unit>
| Completed
let throttlingAgent limit = MailboxProcessor.Start(fun inbox -> async {
// The agent body is not executing in parallel,
// so we can safely use mutable queue & counter
let queue = System.Collections.Generic.Queue<_>()
let running = ref 0
while true do
// Enqueue new work items or decrement the counter
// of how many tasks are running in the background
let! msg = inbox.Receive()
match msg with
| Completed -> decr running
| Enqueue w -> queue.Enqueue(w)
// If we have less than limit & there is some work to
// do, then start the work in the background!
while running.Value < limit && queue.Count > 0 do
let work = queue.Dequeue()
incr running
do!
// When the work completes, send 'Completed'
// back to the agent to free a slot
async { do! work
inbox.Post(Completed) }
|> Async.StartChild
|> Async.Ignore })
|
1:
2:
3:
4:
5:
6:
7:
8:
9:
|
// To use the throttling agent, call it with a specified limit
// and then add items using the 'Enqueue' message!
let w = throttlingAgent 5
for i in 0 .. 20 do
async { printfn "Starting %d" i
do! Async.Sleep(1000)
printfn "Done %d" i }
|> Enqueue
|> w.Post
|
union case ThrottlingMessage.Enqueue: Async<unit> -> ThrottlingMessage
Multiple items
type Async
static member AsBeginEnd : computation:('Arg -> Async<'T>) -> ('Arg * AsyncCallback * obj -> IAsyncResult) * (IAsyncResult -> 'T) * (IAsyncResult -> unit)
static member AwaitEvent : event:IEvent<'Del,'T> * ?cancelAction:(unit -> unit) -> Async<'T> (requires delegate and 'Del :> Delegate)
static member AwaitIAsyncResult : iar:IAsyncResult * ?millisecondsTimeout:int -> Async<bool>
static member AwaitTask : task:Task<'T> -> Async<'T>
static member AwaitWaitHandle : waitHandle:WaitHandle * ?millisecondsTimeout:int -> Async<bool>
static member CancelDefaultToken : unit -> unit
static member Catch : computation:Async<'T> -> Async<Choice<'T,exn>>
static member FromBeginEnd : beginAction:(AsyncCallback * obj -> IAsyncResult) * endAction:(IAsyncResult -> 'T) * ?cancelAction:(unit -> unit) -> Async<'T>
static member FromBeginEnd : arg:'Arg1 * beginAction:('Arg1 * AsyncCallback * obj -> IAsyncResult) * endAction:(IAsyncResult -> 'T) * ?cancelAction:(unit -> unit) -> Async<'T>
static member FromBeginEnd : arg1:'Arg1 * arg2:'Arg2 * beginAction:('Arg1 * 'Arg2 * AsyncCallback * obj -> IAsyncResult) * endAction:(IAsyncResult -> 'T) * ?cancelAction:(unit -> unit) -> Async<'T>
static member FromBeginEnd : arg1:'Arg1 * arg2:'Arg2 * arg3:'Arg3 * beginAction:('Arg1 * 'Arg2 * 'Arg3 * AsyncCallback * obj -> IAsyncResult) * endAction:(IAsyncResult -> 'T) * ?cancelAction:(unit -> unit) -> Async<'T>
static member FromContinuations : callback:(('T -> unit) * (exn -> unit) * (OperationCanceledException -> unit) -> unit) -> Async<'T>
static member Ignore : computation:Async<'T> -> Async<unit>
static member OnCancel : interruption:(unit -> unit) -> Async<IDisposable>
static member Parallel : computations:seq<Async<'T>> -> Async<'T []>
static member RunSynchronously : computation:Async<'T> * ?timeout:int * ?cancellationToken:CancellationToken -> 'T
static member Sleep : millisecondsDueTime:int -> Async<unit>
static member Start : computation:Async<unit> * ?cancellationToken:CancellationToken -> unit
static member StartAsTask : computation:Async<'T> * ?taskCreationOptions:TaskCreationOptions * ?cancellationToken:CancellationToken -> Task<'T>
static member StartChild : computation:Async<'T> * ?millisecondsTimeout:int -> Async<Async<'T>>
static member StartChildAsTask : computation:Async<'T> * ?taskCreationOptions:TaskCreationOptions -> Async<Task<'T>>
static member StartImmediate : computation:Async<unit> * ?cancellationToken:CancellationToken -> unit
static member StartWithContinuations : computation:Async<'T> * continuation:('T -> unit) * exceptionContinuation:(exn -> unit) * cancellationContinuation:(OperationCanceledException -> unit) * ?cancellationToken:CancellationToken -> unit
static member SwitchToContext : syncContext:SynchronizationContext -> Async<unit>
static member SwitchToNewThread : unit -> Async<unit>
static member SwitchToThreadPool : unit -> Async<unit>
static member TryCancelled : computation:Async<'T> * compensation:(OperationCanceledException -> unit) -> Async<'T>
static member CancellationToken : Async<CancellationToken>
static member DefaultCancellationToken : CancellationToken
Full name: Microsoft.FSharp.Control.Async
--------------------
type Async<'T>
Full name: Microsoft.FSharp.Control.Async<_>
type unit = Unit
Full name: Microsoft.FSharp.Core.unit
union case ThrottlingMessage.Completed: ThrottlingMessage
val throttlingAgent : limit:int -> MailboxProcessor<ThrottlingMessage>
Full name: Script.throttlingAgent
val limit : int
Multiple items
type MailboxProcessor<'Msg> =
interface IDisposable
new : body:(MailboxProcessor<'Msg> -> Async<unit>) * ?cancellationToken:CancellationToken -> MailboxProcessor<'Msg>
member Post : message:'Msg -> unit
member PostAndAsyncReply : buildMessage:(AsyncReplyChannel<'Reply> -> 'Msg) * ?timeout:int -> Async<'Reply>
member PostAndReply : buildMessage:(AsyncReplyChannel<'Reply> -> 'Msg) * ?timeout:int -> 'Reply
member PostAndTryAsyncReply : buildMessage:(AsyncReplyChannel<'Reply> -> 'Msg) * ?timeout:int -> Async<'Reply option>
member Receive : ?timeout:int -> Async<'Msg>
member Scan : scanner:('Msg -> Async<'T> option) * ?timeout:int -> Async<'T>
member Start : unit -> unit
member TryPostAndReply : buildMessage:(AsyncReplyChannel<'Reply> -> 'Msg) * ?timeout:int -> 'Reply option
...
Full name: Microsoft.FSharp.Control.MailboxProcessor<_>
--------------------
new : body:(MailboxProcessor<'Msg> -> Async<unit>) * ?cancellationToken:System.Threading.CancellationToken -> MailboxProcessor<'Msg>
static member MailboxProcessor.Start : body:(MailboxProcessor<'Msg> -> Async<unit>) * ?cancellationToken:System.Threading.CancellationToken -> MailboxProcessor<'Msg>
val inbox : MailboxProcessor<ThrottlingMessage>
val async : AsyncBuilder
Full name: Microsoft.FSharp.Core.ExtraTopLevelOperators.async
val queue : System.Collections.Generic.Queue<Async<unit>>
namespace System
namespace System.Collections
namespace System.Collections.Generic
Multiple items
type Queue<'T> =
new : unit -> Queue<'T> + 2 overloads
member Clear : unit -> unit
member Contains : item:'T -> bool
member CopyTo : array:'T[] * arrayIndex:int -> unit
member Count : int
member Dequeue : unit -> 'T
member Enqueue : item:'T -> unit
member GetEnumerator : unit -> Enumerator<'T>
member Peek : unit -> 'T
member ToArray : unit -> 'T[]
...
nested type Enumerator
Full name: System.Collections.Generic.Queue<_>
--------------------
System.Collections.Generic.Queue() : unit
System.Collections.Generic.Queue(capacity: int) : unit
System.Collections.Generic.Queue(collection: System.Collections.Generic.IEnumerable<'T>) : unit
val running : int ref
Multiple items
val ref : value:'T -> 'T ref
Full name: Microsoft.FSharp.Core.Operators.ref
--------------------
type 'T ref = Ref<'T>
Full name: Microsoft.FSharp.Core.ref<_>
val msg : ThrottlingMessage
member MailboxProcessor.Receive : ?timeout:int -> Async<'Msg>
val decr : cell:int ref -> unit
Full name: Microsoft.FSharp.Core.Operators.decr
val w : Async<unit>
System.Collections.Generic.Queue.Enqueue(item: Async<unit>) : unit
property Ref.Value: int
property System.Collections.Generic.Queue.Count: int
val work : Async<unit>
System.Collections.Generic.Queue.Dequeue() : Async<unit>
val incr : cell:int ref -> unit
Full name: Microsoft.FSharp.Core.Operators.incr
member MailboxProcessor.Post : message:'Msg -> unit
static member Async.StartChild : computation:Async<'T> * ?millisecondsTimeout:int -> Async<Async<'T>>
static member Async.Ignore : computation:Async<'T> -> Async<unit>
val w : MailboxProcessor<ThrottlingMessage>
Full name: Script.w
val i : int32
val printfn : format:Printf.TextWriterFormat<'T> -> 'T
Full name: Microsoft.FSharp.Core.ExtraTopLevelOperators.printfn
static member Async.Sleep : millisecondsDueTime:int -> Async<unit>
More information