@@ -41,17 +41,17 @@ module Scheduling =
4141
4242 /// Batch of work as passed from the Submitter to the Scheduler comprising messages with their associated checkpointing/completion callback
4343[<NoComparison; NoEquality>]
44- type Batch < 'M > = { partitionId : int ; messages: 'M []; onCompletion: unit -> unit }
44+ type Batch < 'S , ' M> = { source : 'S ; messages: 'M []; onCompletion: unit -> unit }
4545
4646 /// Thread-safe/lock-free batch-level processing state
4747/// - referenced [ indirectly, see ` mkDispatcher ` ] among all task invocations for a given batch
4848/// - scheduler loop continuously inspects oldest active instance per partition in order to infer attainment of terminal (completed or faulted) state
4949[<NoComparison; NoEquality>]
50- type WipBatch < 'M > =
50+ type WipBatch < 'S , ' M> =
5151 { mutable elapsedMs : int64 // accumulated processing time for stats
5252 mutable remaining : int // number of outstanding completions; 0 => batch is eligible for completion
5353 mutable faults : ConcurrentStack < exn > // exceptions, order is not relevant and use is infrequent hence ConcurrentStack
54- batch: Batch < 'M > }
54+ batch: Batch < 'S , ' M> }
5555
5656 member private __.RecordOk ( duration : TimeSpan ) =
5757 // need to record stats first as remaining = 0 is used as completion gate
@@ -61,7 +61,7 @@ module Scheduling =
6161 __. faults.Push exn
6262
6363 /// Prepares an initial set of shared state for a batch of tasks, together with the Async<unit > computations that will feed their results into it
64- static member Create ( batch : Batch < 'M >, handle ) : WipBatch < 'M > * seq < Async < unit >> =
64+ static member Create ( batch : Batch < 'S , ' M>, handle ) : WipBatch < 'S , 'M > * seq < Async < unit >> =
6565 let x = { elapsedMs = 0 L; remaining = batch.messages.Length; faults = ConcurrentStack(); batch = batch }
6666 x, seq {
6767 for item in batch.messages -> async {
@@ -84,13 +84,13 @@ module Scheduling =
8484/// - replenishing the Dispatcher
8585/// - determining when WipBatches attain terminal state in order to triggering completion callbacks at the earliest possible opportunity
8686/// - triggering abend of the processing should any dispatched tasks start to fault
87- type PartitionedSchedulingEngine < 'M >( log : ILogger , handle , tryDispatch : ( Async <unit> ) -> bool , statsInterval , ? logExternalStats ) =
87+ type PartitionedSchedulingEngine < 'S , 'M when 'S : equality >( log : ILogger , handle , tryDispatch : ( Async <unit> ) -> bool , statsInterval , ? logExternalStats ) =
8888 // Submitters dictate batch commencement order by supply batches in a fair order; should never be empty if there is work in the system
89- let incoming = ConcurrentQueue< Batch< 'M>>()
89+ let incoming = ConcurrentQueue< Batch< 'S , ' M>>()
9090 // Prepared work items ready to feed to Dispatcher (only created on demand in order to ensure we maximize overall progress and fairness)
9191 let waiting = Queue< Async< unit>>( 1024 )
9292 // Index of batches that have yet to attain terminal state (can be >1 per partition)
93- let active = Dictionary< int (* partitionId*) , Queue< WipBatch< 'M>>>()
93+ let active = Dictionary< 'S (* partitionId*) , Queue< WipBatch< 'S , 'M>>>()
9494 (* accumulators for periodically emitted statistics info *)
9595 let mutable cycles , processingDuration = 0 , TimeSpan.Zero
9696 let startedBatches , completedBatches , startedItems , completedItems = PartitionStats(), PartitionStats(), PartitionStats(), PartitionStats()
@@ -130,7 +130,7 @@ module Scheduling =
130130 abend ( AggregateException( exns))
131131 | Some ( Completed batchProcessingDuration) -> // call completion function asap
132132 let partitionId , markCompleted , itemCount =
133- let { batch = { partitionId = p ; onCompletion = f ; messages = msgs } } = queue.Dequeue()
133+ let { batch = { source = p ; onCompletion = f ; messages = msgs } } = queue.Dequeue()
134134 p, f, msgs.LongLength
135135 completedBatches.Record partitionId
136136 completedItems.Record( partitionId, itemCount)
@@ -144,7 +144,7 @@ module Scheduling =
144144let tryPrepareNext () =
145145 match incoming.TryDequeue() with
146146 | false , _ -> false
147- | true , ({ partitionId = pid; messages = msgs} as batch) ->
147+ | true , ({ source = pid; messages = msgs} as batch) ->
148148 startedBatches.Record( pid)
149149 startedItems.Record( pid, msgs.LongLength)
150150 let wipBatch , runners = WipBatch.Create( batch, handle)
@@ -180,34 +180,34 @@ module Scheduling =
180180 Thread.Sleep 1 } // not Async.Sleep, we like this context and/or cache state if nobody else needs it
181181
182182 /// Feeds a batch of work into the queue; the caller is expected to ensure sumbissions are timely to avoid starvation, but throttled to ensure fair ordering
183- member __.Submit ( batches : Batch < 'M >) =
183+ member __.Submit ( batches : Batch < 'S , ' M>) =
184184 incoming.Enqueue batches
185185
186186type ParallelIngester < 'Item > =
187187 static member Start ( log , partitionId , maxRead , submit , ? statsInterval , ? sleepInterval ) =
188188 let makeBatch onCompletion ( items : 'Item seq ) =
189189 let items = Array.ofSeq items
190- let batch : Submission.SubmissionBatch < 'Item > = { partitionId = partitionId; onCompletion = onCompletion; messages = items }
190+ let batch : Submission.SubmissionBatch < _ , 'Item > = { source = partitionId; onCompletion = onCompletion; messages = items }
191191 batch,( items.Length, items.Length)
192- Ingestion.Ingester< 'Item seq, Submission.SubmissionBatch< 'Item>>. Start( log, maxRead, makeBatch, submit, ?statsInterval= statsInterval, ?sleepInterval= sleepInterval)
192+ Ingestion.Ingester< 'Item seq, Submission.SubmissionBatch<_, 'Item>>. Start( log, maxRead, makeBatch, submit, ?statsInterval= statsInterval, ?sleepInterval= sleepInterval)
193193
194194type ParallelProjector =
195195 static member Start ( log : ILogger , maxReadAhead , maxDop , handle , ? statsInterval , ? maxSubmissionsPerPartition , ? logExternalStats )
196196 : ProjectorPipeline < _ > =
197197
198198 let statsInterval = defaultArg statsInterval ( TimeSpan.FromMinutes 5. )
199199 let dispatcher = Scheduling.Dispatcher maxDop
200- let scheduler = Scheduling.PartitionedSchedulingEngine< 'Item>( log, handle, dispatcher.TryAdd, statsInterval, ?logExternalStats= logExternalStats)
200+ let scheduler = Scheduling.PartitionedSchedulingEngine<_, 'Item>( log, handle, dispatcher.TryAdd, statsInterval, ?logExternalStats= logExternalStats)
201201 let maxSubmissionsPerPartition = defaultArg maxSubmissionsPerPartition 5
202202
203- let mapBatch onCompletion ( x : Submission.SubmissionBatch < 'Item >) : Scheduling.Batch < 'Item > =
203+ let mapBatch onCompletion ( x : Submission.SubmissionBatch < _ , 'Item >) : Scheduling.Batch < _ , 'Item > =
204204 let onCompletion () = x.onCompletion(); onCompletion()
205- { partitionId = x.partitionId ; onCompletion = onCompletion; messages = x.messages}
205+ { source = x.source ; onCompletion = onCompletion; messages = x.messages}
206206
207- let submitBatch ( x : Scheduling.Batch < 'Item >) : int =
207+ let submitBatch ( x : Scheduling.Batch < _ , 'Item >) : int =
208208 scheduler.Submit x
209209 0
210210
211- let submitter = Submission.SubmissionEngine<_, _>( log, maxSubmissionsPerPartition, mapBatch, submitBatch, statsInterval)
211+ let submitter = Submission.SubmissionEngine<_, _, _ >(, maxSubmissionsPerPartition, mapBatch, submitBatch, statsInterval)
212212 let startIngester ( rangeLog , partitionId ) = ParallelIngester< 'Item>. Start( rangeLog, partitionId, maxReadAhead, submitter.Ingest)
213213 ProjectorPipeline.Start( log, dispatcher.Pump(), scheduler.Pump, submitter.Pump(), startIngester)
0 commit comments