class Croupier::TaskManagerType

• Croupier::TaskManagerType
• Reference
• Object

Overview

TaskManager is a singleton that keeps track of all tasks. Its methods live in focused files under src/croupier/: kv_store.cr, hash_state.cr, graph.cr, runner.cr and watcher.cr.

Defined in:

croupier.cr
croupier/graph.cr
croupier/hash_state.cr
croupier/kv_store.cr
croupier/runner.cr
croupier/watcher.cr

Constant Summary

AUTORUN_RETRY_MAX_DELAY = 1.0

AUTORUN_RETRY_MIN_DELAY = 0.01

Autorun retry backoff bounds, in seconds: consecutive failures slow the loop from the change-poll minimum to at most one attempt per second; any success resets to the minimum.

FAST_MODE_GRACE = 1.0

Seconds subtracted from the fast-mode baseline, absorbing timestamp granularity and clock skew. The cost is sometimes re-detecting an input modified just before the previous scan.

NO_CONSUMERS = [] of Task

Shared empty for consumers.fetch misses

SCAN_CHUNK_SIZE = 64

How many files each scan worker hashes at a time: small enough that one huge file doesn't hold up much other work, big enough that channel overhead stays negligible

STATE_VERSION = "1"

Version of the state-file schema, stored as __version. A mismatch (or a file with no version) discards all recorded hashes: one full rebuild instead of comparing hashes computed by a different scheme.

Instance Method Summary

• #add_input(task_key : String, input : String) : Bool

  Add input to the inputs of the task registered as task_key.

• #add_mutex(name : String)

  Register the mutex name, keeping the existing lock if there is one: replacing it could swap out a lock a running task holds, and tasks sharing the name would stop excluding each other.

• #all_inputs : Set(String)
• #auto_mode=(auto_mode : Bool)

  If true, it's running in auto mode

• #auto_mode? : Bool

  If true, it's running in auto mode

• #auto_run(targets : Array(String) = [] of String)
• #auto_stop
• #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 set of modified paths (files and kv:// keys)

• #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 set of modified paths (files and kv:// keys)

• #check_dependencies(targets : Array(String) | Nil = nil)

  Raise UnknownInputsError unless every input is a kv:// key, a task, or an existing file.

• #cleanup

  Remove all tasks and everything else (good for tests)

• #dependencies(outputs : Array(String))

  The tasks needed to produce outputs (including themselves), in execution order.

• #dependencies(output : String)

  Single-output convenience overload.

• #depends_on(input : String)

  Outputs of every task that depends, directly or not, on input.

• #depends_on(inputs : Array(String))
• #early_cutoff=(early_cutoff : Bool)

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

• #early_cutoff? : Bool

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

• #fast_dirs=(fast_dirs : Bool)

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

• #fast_dirs? : Bool

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

• #fast_mode=(fast_mode : Bool)

  If true, only compare file dates

• #fast_mode? : Bool

  If true, only compare file dates

• #get(key)
• #hash_directory(path : String) : String

  Hash a single directory input.

• #inputs(targets : Array(String))

  Every input of the given targets and their dependencies.

• #invalidate_graph_cache

  Mark the cached graph and input set stale.

• #last_run : Hash(String, String)

  Hashes recorded by the previous run (from the state file, or folded in by the previous auto mode cycle).

• #last_run=(last_run : Hash(String, String))

  Hashes recorded by the previous run (from the state file, or folded in by the previous auto mode cycle).

• #lock_mutex(name : String)
• #mark_stale_inputs(run_all : Bool = false, targets : Array(String) | Nil = nil)

  Compare inputs against the last run and leave the changed ones in @modified for propagate_staleness.

• #modified : Set(String)

  Inputs (files and kv:// keys) modified since the last run; they make the tasks that consume them stale.

• #modified=(modified : Set(String))

  Inputs (files and kv:// keys) modified since the last run; they make the tasks that consume them stale.

• #modified?(key : String) : Bool

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

• #mutexes : Hash(String, Sync::Mutex)

  A hash of mutexes required by tasks

• #mutexes=(mutexes : Hash(String, Sync::Mutex))

  A hash of mutexes required by tasks

• #next_run : Hash(String, String)

  Output hashes recorded by tasks during this run.

• #next_run=(next_run : Hash(String, String))

  Output hashes recorded by tasks during this run.

• #progress_callback : Proc(String, Nil)

  If set, it's called after every task finishes

• #progress_callback=(progress_callback : Proc(String, Nil))

  If set, it's called after every task finishes

• #propagate_staleness(run_all : Bool = false)

  Mark every task stale or fresh in one O(V+E) pass: find the tasks stale on their own, then everything downstream of them.

• #register_task(task : Task, explicit_id : String | Nil) : Nil

  Register a newly constructed task (called by Task.new).

• #remove_task(task_key : String) : Nil

  Remove the task registered as task_key, under every key it is registered under (one per output) and its id.

• #run_tasks(targets : Array(String) | Nil = nil, run_all : Bool = false, dry_run : Bool = false, parallel : Bool = false, keep_going : Bool = false, early_cutoff : Bool | Nil = nil)

  Run the stale tasks needed to create or update targets (every task when nil), in dependency order

• #save_run

  Save the current state.

• #scan_inputs(scope : Set(String) | Nil = nil)

  Scan the given inputs (all of them by default) and return a hash with their sha1.

• #set(key, value) : Bool

  Store a value and return whether it changed.

• #state_file : String

  Path to the state file that stores hashes between runs

• #state_file=(state_file : String)

  Path to the state file that stores hashes between runs

• #swap_output_hash(output : String, new_hash : String) : String | Nil

  Record the hash of a task output for the next run's state file, and return the hash the last run recorded for it, in one locked step.

• #tasks : Croupier::TaskRegistry

  Registry of all tasks, keyed by each output (or by id for tasks without outputs).

• #this_run : Hash(String, String)

  Input hashes scanned at the start of this run.

• #this_run=(this_run : Hash(String, String))

  Input hashes scanned at the start of this run.

• #unlock_mutex(name : String)
• #use_persistent_store(path : String)

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

• #watch(targets : Array(String) = [] of String) : Nil

  Watch the inputs of targets (all tasks by default) and queue changed paths in @queued_changes.

• #with_state_lock(dry_run : Bool, &)

  Serialize runs on the same state file across processes.

Instance Method Detail