class Croupier::TaskManagerType

Overview

TaskManager is a singleton that keeps track of all tasks

Defined in:

croupier.cr

Instance Method Summary

Instance Method Detail

def _dependencies(outputs : Array(String)) #

Helper function for dependencies Uses memoization to avoid exponential blowup when many tasks share dependencies


[View source]
def _run_tasks(task_names, run_all : Bool = false, dry_run : Bool = false, keep_going : Bool = false, early_cutoff : Bool = true) #

Internal helper to run tasks serially


[View source]
def _run_tasks_parallel(task_names : Array(String) = [] of String, run_all : Bool = false, dry_run : Bool = false, keep_going : Bool = false, early_cutoff : Bool = true) #

Internal helper to run tasks concurrently.

Whenever a task is ready, launch it in a separate fiber. On Crystal >= 1.18 the default execution context is resized to the worker count, so ready tasks run with real multi-core parallelism; on older Crystal this degrades to cooperative concurrency.

Worker fibers only execute tasks and report each outcome over the results channel; this coordinating fiber owns all shared bookkeeping (finished / failed / error collections, stale transitions, early-cutoff notifications), so none of it needs a lock. Receiving batch.size results is the wave barrier. ameba:disable Metrics/CyclomaticComplexity


[View source]
def add_mutex(name : String) #

[View source]
def all_inputs #

[View source]
def auto_mode=(auto_mode : Bool) #

If true, it's running in auto mode


[View source]
def auto_mode? : Bool #

If true, it's running in auto mode


[View source]
def auto_run(targets : Array(String) = [] of String) #

[View source]
def auto_stop #

[View source]
def before_run_hook : Proc(Set(String), Nil) #

If set, it's called in auto mode after changes are detected but before tasks run Receives the list of changed files as an argument


[View source]
def before_run_hook=(before_run_hook : Proc(Set(String), Nil)) #

If set, it's called in auto mode after changes are detected but before tasks run Receives the list of changed files as an argument


[View source]
def check_dependencies #

Check if all inputs are correct: They should all be either task outputs or existing files


[View source]
def cleanup #

Remove all tasks and everything else (good for tests)


[View source]
def dependencies(outputs : Array(String)) #

Get a task list of what tasks need to be done to produce outputs The list is sorted so it can be executed in order


[View source]
def dependencies(output : String) #

Get a task list of what tasks need to be done to produce output The list is sorted so it can be executed in order Overloaded to accept a single string for convenience


[View source]
def depends_on(input : String) #

[View source]
def depends_on(inputs : Array(String)) #

[View source]
def early_cutoff=(early_cutoff : Bool) #

If true, enable early cutoff optimization (skip tasks when upstream outputs unchanged)


[View source]
def early_cutoff? : Bool #

If true, enable early cutoff optimization (skip tasks when upstream outputs unchanged)


[View source]
def fast_dirs=(fast_dirs : Bool) #

If true, directories depend on a list of files, not its contents


[View source]
def fast_dirs? : Bool #

If true, directories depend on a list of files, not its contents


[View source]
def fast_mode=(fast_mode : Bool) #

If true, only compare file dates


[View source]
def fast_mode? : Bool #

If true, only compare file dates


[View source]
def get(key) #

[View source]
def inputs(targets : Array(String)) #

The set of all inputs for the given tasks


[View source]
def invalidate_graph_cache #

Invalidate the cached task graph


[View source]
def invalidate_graph_cache_no_store #

Invalidate the graph cache without setting k/v (for internal use from Task.initialize)


[View source]
def last_run : Hash(String, String) #

SHA1 of files from last run


[View source]
def last_run=(last_run : Hash(String, String)) #

SHA1 of files from last run


[View source]
def lock_mutex(name : String) #

[View source]
def mark_stale_inputs #

Read state of last run, then scan inputs and compare


[View source]
def modified : Set(String) #

Registry of modified files, which will make tasks stale


[View source]
def modified=(modified : Set(String)) #

Registry of modified files, which will make tasks stale


[View source]
def modified?(key : String) : Bool #

Whether key (a file or kv:// key) was modified since the last run.


[View source]
def mutexes #

A hash of mutexes required by tasks


[View source]
def mutexes=(mutexes) #

A hash of mutexes required by tasks


[View source]
def next_run : Hash(String, String) #

SHA1 of input files as of ending this run


[View source]
def next_run=(next_run : Hash(String, String)) #

SHA1 of input files as of ending this run


[View source]
def previous_output_hash(output : String) : String | Nil #

The hash recorded for output by the last completed run, if any.


[View source]
def progress_callback : Proc(String, Nil) #

If set, it's called after every task finishes


[View source]
def progress_callback=(progress_callback : Proc(String, Nil)) #

If set, it's called after every task finishes


[View source]
def propagate_staleness #

Propagate staleness through the task graph in a single forward pass. This replaces the expensive recursive staleness checking with an O(V+E) algorithm that's critical for tasks with many inputs.


[View source]
def record_output_hash(output : String, new_hash : String) : Nil #

Record the hash of a task output for the next run's state file. Thread-safe for parallel task workers.


[View source]
def register_subtask(master_id : String, subtask : Task) #

Register a subtask with the task manager


[View source]
def remove_subtasks(master_id : String) #

Remove all subtasks belonging to a master task


[View source]
def run_tasks(run_all : Bool = false, dry_run : Bool = false, parallel : Bool = false, keep_going : Bool = false, early_cutoff : Bool | Nil = nil) #

Run all stale tasks in dependency order

If run_all is true, run non-stale tasks too If dry_run is true, only log what would be done, but don't do it If parallel is true, run tasks in parallel If keep_going is true, keep going even if a task fails If early_cutoff is true, skip tasks when upstream outputs are unchanged


[View source]
def run_tasks(targets : Array(String), run_all : Bool = false, dry_run : Bool = false, parallel : Bool = false, keep_going : Bool = false, early_cutoff : Bool | Nil = nil) #

Run the tasks needed to create or update the requested targets

If run_all is true, run non-stale tasks too If dry_run is true, only log what would be done, but don't do it If parallel is true, run tasks in parallel If keep_going is true, keep going even if a task fails If early_cutoff is true, skip tasks when upstream outputs are unchanged


[View source]
def save_run #

We ran all tasks, store the current state


[View source]
def scan_inputs #

Scan all inputs and return a hash with their sha1.

Plain files and the contents of directory inputs are hashed in parallel (a pool of worker fibers bounded by CPU count), since both the disk read and the hashing are independent per file.


[View source]
def set(key, value) #

[View source]
def sorted_task_graph #

[View source]
def state_file : String #

Path to the state file that stores hashes between runs


[View source]
def state_file=(state_file : String) #

Path to the state file that stores hashes between runs


[View source]
def tasks : Hash(String, Croupier::Task) #

Registry of all tasks


[View source]
def tasks=(tasks : Hash(String, Croupier::Task)) #

Registry of all tasks


[View source]
def this_run : Hash(String, String) #

SHA1 of files as of starting this run


[View source]
def this_run=(this_run : Hash(String, String)) #

SHA1 of files as of starting this run


[View source]
def unlock_mutex(name : String) #

[View source]
def use_persistent_store(path : String) #

Use a persistent k/v store in this path instead of the default memory store


[View source]
def watch(targets : Array(String) = [] of String) #

[View source]