Class: Epi::Job

Inherits:
Object
  • Object
show all
Extended by:
Forwardable
Defined in:
lib/epi/job.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(job_description, state) ⇒ Job

Returns a new instance of Job.



16
17
18
19
20
21
22
 
# File 'lib/epi/job.rb', line 16

def initialize(job_description, state)
  @job_description = job_description
  @triggers = job_description.triggers.map { |t| Trigger.make self, *t }
  @expected_count = state['expected_count'] || job_description.initial_processes
  @pids = state['pids']
  @dying_pids = state['dying_pids']
end

Instance Attribute Details

#expected_count ⇒ Object

Returns the value of attribute expected_count.



8
9
10
 
# File 'lib/epi/job.rb', line 8

def expected_count
  @expected_count
end

#job_description ⇒ Object (readonly)

Returns the value of attribute job_description.



7
8
9
 
# File 'lib/epi/job.rb', line 7

def job_description
  @job_description
end

Instance Method Details

#dying_count ⇒ Object 



124
125
126
 
# File 'lib/epi/job.rb', line 124

def dying_count
  dying_pids.count
end

#dying_pids ⇒ Hash

Get a hash of PIDs, with internal process IDs as keys and PIDs as values, for process that are dying

Examples:

{'1a2v3c4d' => 4820}

Returns:

  • (Hash) 


44
45
46
 
# File 'lib/epi/job.rb', line 44

def dying_pids
  @dying_pids ||= {}
end

#logger ⇒ Object 



12
13
14
 
# File 'lib/epi/job.rb', line 12

def logger
  Epi.logger
end

#pid_key(proc_id) ⇒ String|NilClass

Get the data key for the PID file of the given process ID or PID

Parameters:

  • proc_id (String|Fixnum) —

    Example: '1a2b3c4d' or 1234

Returns:

  • (String|NilClass) —

    Example: pids/job_id/1ab3c4d.pid, or nil if not found



51
52
53
54
 
# File 'lib/epi/job.rb', line 51

def pid_key(proc_id)
  proc_id = pids.key(proc_id) if Fixnum === proc_id
  proc_id && job_description.pid_key(proc_id)
end

#pids ⇒ Hash

Get a hash of PIDs, with internal process IDs as keys and PIDs as values

Examples:

{'1a2v3c4d' => 4820}

Returns:

  • (Hash) 


36
37
38
 
# File 'lib/epi/job.rb', line 36

def pids
  @pids ||= {}
end

#replace(pid, &callback) ⇒ Object

Replace a running process with a new one

Parameters:

  • pid (Fixnum) —

    PID of the process to replace



130
131
132
133
134
135
 
# File 'lib/epi/job.rb', line 130

def replace(pid, &callback)
  stop_one pid do
    start_one while running_count < expected_count
    callback.call if callback
  end
end

#restart! ⇒ Object 



103
104
105
106
107
108
109
110
111
112
 
# File 'lib/epi/job.rb', line 103

def restart!
  count = expected_count
  if count > 0
    self.expected_count = 0
    sync!
    self.expected_count = count
    sync!
  end
  self
end

#run_triggers! ⇒ Object 



80
81
82
 
# File 'lib/epi/job.rb', line 80

def run_triggers!
  @triggers.each &:try
end

#running_count ⇒ Object 



120
121
122
 
# File 'lib/epi/job.rb', line 120

def running_count
  pids.count
end

#running_processes ⇒ Object 



114
115
116
117
118
 
# File 'lib/epi/job.rb', line 114

def running_processes
  pids.map do |proc_id, pid|
    [proc_id, ProcessStatus[pid] || RunningProcess.new(pid)]
  end.to_h
end

#shutdown!(&callback) ⇒ Object 



84
85
86
87
88
89
90
91
92
93
94
95
96
 
# File 'lib/epi/job.rb', line 84

def shutdown!(&callback)
  count = running_count
  if count > 0
    count.times do
      stop_one do
        count -= 1
        callback.call if callback && count == 0
      end
    end
  else
    callback.call if callback
  end
end

#state ⇒ Object

noinspection RubyStringKeysInHashInspection



25
26
27
28
29
30
31
 
# File 'lib/epi/job.rb', line 25

def state
  {
      'expected_count' => expected_count,
      'pids' => pids,
      'dying_pids' => dying_pids
  }
end

#sync! ⇒ Object

Stops processes that shouldn't run, starts process that should run, and fires event handlers



58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
 
# File 'lib/epi/job.rb', line 58

def sync!

  # Remove non-running PIDs from the list
  pids.reject { |_, pid| ProcessStatus.pids.include? pid }.each do |proc_id, pid|
    logger.debug "Lost process #{pid}"
    pids.delete proc_id
  end

  # Remove non-running PIDs from the dying list. This is just in case
  # the daemon crashed before it was able to clean up a dying worker
  # (i.e. it sent a TERM but didn't get around to sending a KILL)
  dying_pids.select! { |_, pid| ProcessStatus.pids.include? pid }

  # TODO: clean up processes that never died how they should have

  # Run new processes
  start_one while running_count < expected_count

  # Kill old processes
  stop_one while running_count > expected_count
end

#terminate! ⇒ Object 



98
99
100
101
 
# File 'lib/epi/job.rb', line 98

def terminate!
  self.expected_count = 0
  sync!
end