Module Async_kernel.Async_kernel_scheduler
The Async scheduler is responsible for running Async jobs. It maintains the queue of jobs that need to run. A "cycle" consists of running some (possibly all) jobs in the queue, along with some other bookkeeping, like advancing Async's clock to the current time.
type 'a with_options= ?monitor:Monitor.t -> ?priority:Priority.t -> 'a
val current_execution_context : unit -> Execution_context.tval within_context : Execution_context.t -> (unit -> 'a) -> ('a, unit) Core_kernel.Result.twithin_context context frunsf ()right now with the specified execution context. Iffraises, then the exception is sent to the monitor ofcontext, andError ()is returned.
val within' : ((unit -> 'a Deferred.t) -> 'a Deferred.t) with_optionswithin' f ~monitor ~priorityrunsf ()right now, with the specified block group, monitor, and priority set as specified. They will be reset to their original values whenfreturns. Iffraises, then the result ofwithin'will never become determined, but the exception will end up in the specified monitor.
val within : ((unit -> unit) -> unit) with_optionswithinis likewithin', but doesn't require the thunk to return a deferred.
val within_v : ((unit -> 'a) -> 'a option) with_optionswithin_vis likewithin, but allows a value to be returned byf.
val with_local : 'a Core_kernel.Univ_map.Key.t -> 'a option -> f:(unit -> 'b) -> 'bwith_local key value ~f, when run in the current execution context,e, runsfright now in a new execution context,e', that is identical toeexcept thatfind_local key = value. As usual,e'will be in effect in asynchronous computations started byf. Whenwith_localreturns, the execution context is restored toe.
val find_local : 'a Core_kernel.Univ_map.Key.t -> 'a optionfind_local keyreturns the value associated tokeyin the current execution context.
val schedule' : ((unit -> 'a Deferred.t) -> 'a Deferred.t) with_optionsJust like
within', but instead of running the thunk right now, adds it to the Async queue to be run with other Async jobs.
val schedule : ((unit -> unit) -> unit) with_optionsJust like
schedule', but doesn't require the thunk to return a deferred.
val enqueue_job : Execution_context.t -> ('a -> unit) -> 'a -> uniteneque_job execution_context.t f aenqueues into the scheduler's job queue a job that will runf ainexecution_context.
val thread_safe_enqueue_job : Execution_context.t -> ('a -> unit) -> 'a -> unitthread_safe_enqueue_jobis likeenqueue_job, except it is for use from a thread that doesn't hold the Async lock.
val preserve_execution_context : ('a -> unit) -> ('a -> unit) Core_kernel.Staged.tpreserve_execution_context t fsaves the current execution context and returns a functiongsuch thatg arunsf ain the saved execution context.g abecomes determined whenf abecomes determined.
val preserve_execution_context' : ('a -> 'b Deferred.t) -> ('a -> 'b Deferred.t) Core_kernel.Staged.tval cycle_start : unit -> Core_kernel.Time.tcycle_start ()returns the result ofTime.now ()called at the beginning of cycle.
val cycle_start_ns : unit -> Async_kernel__.Import.Time_ns.tval cycle_times : unit -> Core_kernel.Time.Span.t Async_kernel__.Async_stream.tcycle_times ()returns a stream that is extended with an element at the start of each Async cycle, with the amount of time that the previous cycle took, as determined by calls toTime.nowat the beginning and end of the cycle.
val cycle_times_ns : unit -> Async_kernel__.Import.Time_ns.Span.t Async_kernel__.Async_stream.tval long_cycles : at_least:Async_kernel__.Import.Time_ns.Span.t -> Async_kernel__.Import.Time_ns.Span.t Async_kernel__.Async_stream.tlong_cycles ~at_leastreturns a stream of cycles whose duration is at leastat_least.long_cyclesis more efficient thancycle_timesbecause it only allocates a stream entry when there is a long cycle, rather than on every cycle.
val cycle_count : unit -> intcycle_count ()returns the total number of Async cycles that have happened.
val total_cycle_time : unit -> Async_kernel__.Import.Time_ns.Span.ttotal_cycle_time ()returns the total (wall) time spent executing jobs in Async cycles.
val event_precision : unit -> Core_kernel.Time.Span.tThe
alarm_precisionof the timing-wheel used to implement Async'sClock.
val event_precision_ns : unit -> Async_kernel__.Import.Time_ns.Span.tval force_current_cycle_to_end : unit -> unitforce_current_cycle_to_end ()causes no more normal priority jobs to run in the current cycle, and for the end-of-cycle jobs (i.e., writes) to run, and then for the cycle to end.
val set_max_num_jobs_per_priority_per_cycle : int -> unitset_max_num_jobs_per_priority_per_cycle intsets the maximum number of jobs that will be done at each priority within each Async cycle. The default is500.max_num_jobs_per_priority_per_cycleretrieves the current value.
val max_num_jobs_per_priority_per_cycle : unit -> intval set_record_backtraces : bool -> unitset_record_backtraces do_recordsets whether Async should keep in the execution context the history of stack backtraces (obtained viaBacktrace.get) that led to the current job. If an Async job has an unhandled exception, this backtrace history will be recorded in the exception. In particular the history will appear in an unhandled exception that reaches the main monitor. This can have a substantial performance impact, both in running time and space usage.
val yield : unit -> unit Deferred.tyield ()returns a deferred that becomes determined after the current cycle completes. This can be useful to improve fairness byyielding within a computation to give other jobs a chance to run.
val yield_until_no_jobs_remain : unit -> unit Deferred.tyield_until_no_jobs_remain ()returns a deferred that becomes determined the next time Async's job queue is empty. This is useful in tests when one needs to wait for the completion of all the jobs based on what's in the queue, when those jobs might create other jobs -- without depending on I/O or the passage of wall-clock time.
val yield_every : n:int -> (unit -> unit Deferred.t) Core_kernel.Staged.tyield_every ~nreturns a function that will act asyieldeveryncalls and asreturn ()the rest of the time. This is useful for improving fairness in circumstances where you don't have good control of the batch size, but can insert a deferred into every iteration.yield_everyraises ifn <= 0.
val num_jobs_run : unit -> intnum_jobs_run ()returns the number of jobs that have been run since starting. The returned value includes the currently running job.
val num_pending_jobs : unit -> intnum_pending_jobsreturns the number of jobs that are queued to run by the scheduler.
module Expert : sig ... endmodule Private = Async_kernel__.Scheduler