Class: Omnizip::Parallel::JobScheduler

Inherits:
Object
  • Object
show all
Defined in:
lib/omnizip/parallel/job_scheduler.rb

Overview

Job scheduler for load balancing and work distribution

Manages job assignment to workers using different strategies:

  • Dynamic: Workers pull jobs as they become available (default)
  • Static: Pre-assign equal chunks to each worker

Examples:

Create scheduler with dynamic strategy

scheduler = Omnizip::Parallel::JobScheduler.new(strategy: :dynamic)
scheduler.schedule_jobs(jobs, worker_count: 4)

Create scheduler with static strategy

scheduler = Omnizip::Parallel::JobScheduler.new(strategy: :static)
assignments = scheduler.schedule_jobs(jobs, worker_count: 4)

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(strategy: :dynamic) ⇒ JobScheduler

Initialize job scheduler

Parameters:

  • strategy (Symbol) (defaults to: :dynamic) —

    :dynamic or :static



25
26
27
28
 
# File 'lib/omnizip/parallel/job_scheduler.rb', line 25

def initialize(strategy: :dynamic)
  @strategy = strategy
  validate_strategy!
end

Instance Attribute Details

#strategy ⇒ Symbol (readonly)

Returns scheduling strategy.

Returns:

  • (Symbol) —

    scheduling strategy



20
21
22
 
# File 'lib/omnizip/parallel/job_scheduler.rb', line 20

def strategy
  @strategy
end

Instance Method Details

#calculate_load_balance(assignments) ⇒ Float

Calculate load balance quality metric

Parameters:

  • assignments (Hash) —

    worker_id => [jobs] mapping

Returns:

  • (Float) —

    balance score (0.0 = perfect, 1.0 = worst)



63
64
65
66
67
68
69
70
71
72
73
74
75
76
 
# File 'lib/omnizip/parallel/job_scheduler.rb', line 63

def calculate_load_balance(assignments)
  return 0.0 if assignments.empty?

  # Calculate total size per worker
  worker_sizes = assignments.transform_values do |jobs|
    jobs.sum { |job| job_size(job) }
  end

  sizes = worker_sizes.values
  return 0.0 if sizes.empty? || sizes.max.zero?

  # Balance = (max - min) / max
  (sizes.max - sizes.min).to_f / sizes.max
end

#estimate_completion_time(jobs, worker_count:, bytes_per_second: 10_000_000) ⇒ Float

Estimate completion time based on job sizes and worker count

Parameters:

  • jobs (Array) —

    array of jobs with :size attribute

  • worker_count (Integer) —

    number of workers

  • bytes_per_second (Float) (defaults to: 10_000_000) —

    processing rate

Returns:

  • (Float) —

    estimated seconds to completion



50
51
52
53
54
55
56
57
 
# File 'lib/omnizip/parallel/job_scheduler.rb', line 50

def estimate_completion_time(jobs, worker_count:,
bytes_per_second: 10_000_000)
  total_bytes = jobs.sum { |job| job_size(job) }
  return 0.0 if total_bytes.zero? || worker_count.zero?

  # Simple estimate: total bytes / (workers * rate)
  total_bytes.to_f / (worker_count * bytes_per_second)
end

#schedule_jobs(jobs, worker_count:) ⇒ Hash, Array

Schedule jobs for workers

Parameters:

  • jobs (Array) —

    array of jobs to schedule

  • worker_count (Integer) —

    number of workers

Returns:

  • (Hash, Array) —

    assignments (strategy-dependent)



35
36
37
38
39
40
41
42
 
# File 'lib/omnizip/parallel/job_scheduler.rb', line 35

def schedule_jobs(jobs, worker_count:)
  case @strategy
  when :dynamic
    schedule_dynamic(jobs, worker_count)
  when :static
    schedule_static(jobs, worker_count)
  end
end