Class: Patriot::JobStore::InMemoryStore

Inherits:
Base
  • Object
show all
Includes:
Util::Logger, Util::Retry
Defined in:
lib/patriot/job_store/in_memory_store.rb

Overview

a JobStore implementation on memory

Constant Summary

Constants included from Util::Config

Util::Config::ADMIN_USER_KEY, Util::Config::DEFAULT_CONFIG, Util::Config::DEFAULT_PLUGIN_DIR, Util::Config::INFO_SERVER_PORT_KEY, Util::Config::PASSWORD_KEY, Util::Config::PLUGIN_DIR_KEY, Util::Config::PLUGIN_INIT_SCRIPT, Util::Config::PLUGIN_KEY, Util::Config::PLUGIN_LIB_DIR, Util::Config::USERNAME_KEY, Util::Config::WORKER_HOST_KEY, Util::Config::WORKER_USER_KEY

Instance Method Summary collapse

Methods included from Util::Retry

execute_with_retry

Methods included from Util::Logger

#create_logger

Methods included from Util::Config

#load_config, #load_plugins

Methods inherited from Base

#get, #process_subsequent

Constructor Details

#initialize(store_id, config) ⇒ InMemoryStore

Returns a new instance of InMemoryStore.

See Also:



11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
 
# File 'lib/patriot/job_store/in_memory_store.rb', line 11

def initialize(store_id, config)
  @config = config
  @logger = create_logger(config)
  @mutex  = Mutex.new
  @jobs       = {} # hash from job_id to job content in hash
  # hash from state to list of job_id
  @job_states = {Patriot::JobStore::JobState::INIT      => [],
                 Patriot::JobStore::JobState::SUCCEEDED => [Patriot::JobStore::INITIATOR_JOB_ID],
                 Patriot::JobStore::JobState::WAIT      => [],
                 Patriot::JobStore::JobState::RUNNING   => [],
                 Patriot::JobStore::JobState::SUSPEND   => [],
                 Patriot::JobStore::JobState::FAILED    => [],
                 Patriot::JobStore::JobState::DISCARDED => []}
  @producers   = {} # hash from job_id to produces products
  @consumers   = {} # hash from job_id to referece products
  @job_history = {} # hash from job_id to a array of its execution hisotry
end

Instance Method Details

#acceptable?(job) ⇒ Boolean

Returns:

  • (Boolean)

See Also:



60
61
62
63
 
# File 'lib/patriot/job_store/in_memory_store.rb', line 60

def acceptable?(job)
  raise "invalid class #{job.class}" unless job.is_a?(Patriot::JobStore::Job)
  return true
end

#delete_job(job_id) ⇒ Object

See Also:



302
303
304
305
306
307
308
309
310
 
# File 'lib/patriot/job_store/in_memory_store.rb', line 302

def delete_job(job_id)
  job_id = job_id.to_sym
  @mutex.synchronize do
    @job_states.each{|s,js| js.delete(job_id)}
    @jobs.delete(job_id)
    @producers.delete(job_id)
    @consumers.delete(job_id)
  end
end

#find_jobs_by_state(state, opts = {}) ⇒ Object

See Also:



190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
 
# File 'lib/patriot/job_store/in_memory_store.rb', line 190

def find_jobs_by_state(state, opts = {})
  all_records = @job_states[state] - [Patriot::JobStore::INITIATOR_JOB_ID]
  size        = all_records.size
  opts        = {:limit => size, :offset => 0}.merge(opts)
  filter      = opts.has_key?(:filter_exp) ? Regexp.new(opts[:filter_exp].gsub(/(?<!\\)%/,'.*').gsub(/(?<!\\)_/,'.')) : nil
  result      = []
  opts[:offset].upto(size).each do |i|
    break if i >= size
    break if result.size >= opts[:limit]
    job_id = all_records[size - 1 - i].to_s
    next  if !filter.nil? && !filter.match(job_id)
    result << job_id
  end
  return result
end

#get_consumers(products, opts = {:include_attrs => [Patriot::Command::STATE_ATTR]}) ⇒ Object

See Also:



173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
 
# File 'lib/patriot/job_store/in_memory_store.rb', line 173

def get_consumers(products, opts = {:include_attrs => [Patriot::Command::STATE_ATTR]})
  opts = {:include_attrs => []}.merge(opts)
  products = [products] unless products.is_a?(Array)
  consumers = []
  products.each{|product|
    @consumers.map{|pid, prods|
      if prods.include?(product)
        job = @jobs[pid].filter_attributes(opts[:include_attrs])
        job[:job_id] = pid.to_s
        consumers.push(job)
      end
    }
  }
  return consumers.uniq
end

#get_execution_history(job_id, opts = {}) ⇒ Object

See Also:



207
208
209
210
 
# File 'lib/patriot/job_store/in_memory_store.rb', line 207

def get_execution_history(job_id, opts = {})
  opts = {:limit => 1, :order => :DESC}
  return @job_history[job_id.to_sym] || []
end

#get_graph(job_id, opts = {}) ⇒ Array

get nodes and edges information to render graph

Parameters:

  • job_id (String) —

    JOB ID

  • opts (Hash) (defaults to: {}) —

    options

Returns:

  • (Array) —
    nodes, edges 


216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
 
# File 'lib/patriot/job_store/in_memory_store.rb', line 216

def get_graph(job_id, opts = {})
  job = get(job_id)
  history = get_execution_history(job_id, {})[0]

  hashed_job = {
    :job_id => job.job_id,
    :history => history,
    :depth => 0
  }.merge(job.attributes)

  # set self node
  nodes = {job_id => hashed_job}
  edges = []

  _set_dependency(
    :producers,
    opts[:producer_depth],
    nodes,
    edges,
    hashed_job
  )

  _set_dependency(
    :consumers,
    opts[:consumer_depth],
    nodes,
    edges,
    hashed_job
  )

  return {:nodes => nodes, :edges => edges}
end

#get_job(job_id) ⇒ Object

See Also:



149
150
151
152
153
 
# File 'lib/patriot/job_store/in_memory_store.rb', line 149

def get_job(job_id)
  return nil if job_id.nil?
  raise "string is expected but job_id is a #{job_id.class}" unless job_id.is_a?(String)
  return @jobs[job_id.to_sym]
end

#get_job_size(opts = {}) ⇒ Object

See Also:



290
291
292
293
294
295
296
297
298
299
 
# File 'lib/patriot/job_store/in_memory_store.rb', line 290

def get_job_size(opts = {})
  opts  = {:ignore_states => []}.merge(opts)
  sizes = {}
  @job_states.each do |s,js|
    next if opts[:ignore_states].include?(s)
    sizes[s] = js.size
    sizes[s] = sizes[s] -1 if s == Patriot::JobStore::JobState::SUCCEEDED
  end
  return sizes
end

#get_job_tickets(host, nodes, options = {}) ⇒ Object

See Also:



66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
 
# File 'lib/patriot/job_store/in_memory_store.rb', line 66

def get_job_tickets(host, nodes, options = {})
  nodes = [nodes] unless nodes.is_a?(Array)
  @mutex.synchronize do
    return @job_states[Patriot::JobStore::JobState::WAIT].map do |wid|
      job = @jobs[wid]
      # check host and node
      next unless job[Patriot::Command::EXEC_NODE_ATTR].nil?      || nodes.include?(job[Patriot::Command::EXEC_NODE_ATTR])
      next unless job[Patriot::Command::EXEC_HOST_ATTR].nil?      || host == job[Patriot::Command::EXEC_HOST_ATTR]
      next unless job[Patriot::Command::START_DATETIME_ATTR].nil? || Time.now > job[Patriot::Command::START_DATETIME_ATTR]
      # check dependency
      reference = @consumers[wid] || []
      producers = @producers.map{|pid, prods| pid unless (prods & reference).empty?}.compact
      next if !reference.empty? && producers.empty? # no producer exists
      next if producers.any?{|pjid| !@job_states[Patriot::JobStore::JobState::SUCCEEDED].include?(pjid)}
      JobTicket.new(wid.to_s, job.update_id, job[Patriot::Command::EXEC_NODE_ATTR])
    end.compact
  end
end

#get_producers(products, opts = {:include_attrs => [Patriot::Command::STATE_ATTR]}) ⇒ Object

See Also:



156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
 
# File 'lib/patriot/job_store/in_memory_store.rb', line 156

def get_producers(products, opts = {:include_attrs => [Patriot::Command::STATE_ATTR]})
  opts = {:include_attrs => []}.merge(opts)
  products = [products] unless products.is_a?(Array)
  producers = []
  products.each{|product|
    @producers.map{|pid, prods|
      if prods.include?(product)
        job = @jobs[pid].filter_attributes(opts[:include_attrs])
        job[:job_id] = pid.to_s
        producers.push(job)
      end
    }
  }
  return producers.uniq
end

#offer_to_execute(job_ticket) ⇒ Object

See Also:



86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
 
# File 'lib/patriot/job_store/in_memory_store.rb', line 86

def offer_to_execute(job_ticket)
  job_id    = job_ticket.job_id.to_sym
  update_id = job_ticket.update_id
  @mutex.synchronize do
    unless _check_and_set_state(job_id, update_id, Patriot::JobStore::JobState::WAIT, Patriot::JobStore::JobState::RUNNING)
      @logger.debug("execution of job: #{job_id} is skipped")
      return
    end
    job = @jobs[job_id]
    raise "no entry found for #{job_ticket}" if job.nil?
    begin
      # TODO make the max number of histories configurable and keep multiple histories
      execution_id         = Time.now.to_i
      @job_history[job_id] = [{:id       => execution_id,
                               :job_id   => job_id.to_s,
                               :host     => job_ticket.exec_host,
                               :node     => job_ticket.exec_node,
                               :thread   => job_ticket.exec_thread,
                               :begin_at => Time.now
                             }]
      return {:execution_id => execution_id, :command => job.to_command(@config)}
    rescue Exception => e
      _check_and_set_state(job_id, update_id, Patriot::JobStore::JobState::RUNNING, Patriot::JobStore::JobState::FAILED)
      raise e
    end
  end
end

#register(update_id, jobs) ⇒ Object

See Also:



30
31
32
33
34
35
 
# File 'lib/patriot/job_store/in_memory_store.rb', line 30

def register(update_id, jobs)
  jobs.each{|job| raise "#{job.job_id} is not acceptable" unless acceptable?(job) }
  @mutex.synchronize do
    jobs.each {|job| _upsert(update_id, job) }
  end
end

#report_completion_status(job_ticket) ⇒ Object

See Also:



115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
 
# File 'lib/patriot/job_store/in_memory_store.rb', line 115

def report_completion_status(job_ticket)
  job_id    = job_ticket.job_id.to_sym
  update_id = job_ticket.update_id
  exit_code = job_ticket.exit_code
  raise "exit code is not set " if exit_code.nil?
  state     = Patriot::JobStore::EXIT_CODE_TO_STATE[exit_code]
  raise "invalid exit code #{exit_code} " if state.nil?
  @mutex.synchronize do
    # TODO save finish_time to history server
    last_history = @job_history[job_id]
    raise "illegal state job_history is not set for #{job_id}" if last_history.nil? || last_history.empty?
    last_history = last_history[0]
    # TODO make the max number of histories configurable and keep multiple histories
    @job_history[job_id] = [last_history.merge({:exit_code => exit_code, :end_at => Time.now, :description => job_ticket.description})]
    return _check_and_set_state(job_id, update_id, Patriot::JobStore::JobState::RUNNING, state)
  end
end

#set_state(update_id, job_ids, new_state) ⇒ Object

See Also:



134
135
136
137
138
139
140
141
142
143
144
145
146
 
# File 'lib/patriot/job_store/in_memory_store.rb', line 134

def set_state(update_id, job_ids, new_state)
  @mutex.synchronize do
    job_ids = job_ids.map do |jid|
      @jobs[jid.to_sym][Patriot::Command::STATE_ATTR] = new_state
      jid.to_sym
    end
    @job_states.each do |s,jobs|
      next if s == new_state
      @job_states[s] -= job_ids
    end
    @job_states[new_state] += job_ids
  end
end