Class: Tengine::Job::Runtime::Signal

Inherits:
Object
  • Object
show all
Defined in:
lib/tengine/job/runtime/signal.rb

Defined Under Namespace

Modules: Transmittable Classes: Error, Reservation

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(event) ⇒ Signal 



23
24
25
26
 
# File 'lib/tengine/job/runtime/signal.rb', line 23

def initialize(event)
  @event = event
  reset
end

Instance Attribute Details

#callback ⇒ Object

start.job.job.tengineイベントによって ジョブは :ready -> :starting -> :running に遷移するが、 一度のroot_jobnet.update_with_lock では :starting が保存されないので、 2回のroot_jobnet.update_with_lock に分けることができるようにするための 処理を記憶しておく属性です



20
21
22
 
# File 'lib/tengine/job/runtime/signal.rb', line 20

def callback
  @callback
end

#callbacks ⇒ Object (readonly)

Returns the value of attribute callbacks.



21
22
23
 
# File 'lib/tengine/job/runtime/signal.rb', line 21

def callbacks
  @callbacks
end

#data ⇒ Object

現時点ではジョブのrunからackを返す際にPIDを保持するために使用します。



13
14
15
 
# File 'lib/tengine/job/runtime/signal.rb', line 13

def data
  @data
end

#event ⇒ Object (readonly)

Returns the value of attribute event.



9
10
11
 
# File 'lib/tengine/job/runtime/signal.rb', line 9

def event
  @event
end

#paths ⇒ Object (readonly)

Returns the value of attribute paths.



9
10
11
 
# File 'lib/tengine/job/runtime/signal.rb', line 9

def paths
  @paths
end

#reservations ⇒ Object (readonly)

Returns the value of attribute reservations.



9
10
11
 
# File 'lib/tengine/job/runtime/signal.rb', line 9

def reservations
  @reservations
end

#受け渡しのためにデータを一時的に保持する属性。 ⇒ Object

現時点ではジョブのrunからackを返す際にPIDを保持するために使用します。



13
 
# File 'lib/tengine/job/runtime/signal.rb', line 13

attr_accessor :data

Instance Method Details

#cache(*args) ⇒ Object 



71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
 
# File 'lib/tengine/job/runtime/signal.rb', line 71

def cache(*args)
  case args.length
  when 1 then
    obj = args.first
    return nil if obj.nil?
    if obj.is_a?(Array)
      obj.map{|o| cache(o)}
    else
      cache(*cache_key(obj)) || remember(obj)
    end
  when 2 then
    @cache[args]
  else
    raise ArgumentError, "#{self.class.name}#cache requires 1 or 2 arguments"
  end
end

#cache_key(obj) ⇒ Object 



93
94
95
 
# File 'lib/tengine/job/runtime/signal.rb', line 93

def cache_key(obj)
  return [obj.class.name, obj.id.to_s]
end

#cache_list ⇒ Object 



97
98
99
100
101
102
103
104
105
 
# File 'lib/tengine/job/runtime/signal.rb', line 97

def cache_list
  Tengine.logger.debug "-" * 100
  Tengine.logger.debug "#{__FILE__}##{__LINE__}"
  Tengine.logger.debug "object_id: #{object_id}"
  @cache.each do |key, obj|
    Tengine.logger.debug "#{obj.object_id} #{key.inspect} #{obj.inspect}" << (obj.changed? ? " CHANGED" : "")
  end
  Tengine.logger.debug "-" * 100
end

#call_later(&block) ⇒ Object 



42
43
44
 
# File 'lib/tengine/job/runtime/signal.rb', line 42

def call_later(&block)
  @callbacks.push(block)
end

#changed_vertecs ⇒ Object 



107
108
109
110
111
 
# File 'lib/tengine/job/runtime/signal.rb', line 107

def changed_vertecs
  @cache.values.select(&:changed?).
    map{|obj| obj.is_a?(Tengine::Job::Runtime::Edge) ? obj.owner : obj}.
    uniq
end

#execution ⇒ Object 



113
114
115
 
# File 'lib/tengine/job/runtime/signal.rb', line 113

def execution
  @execution ||= Tengine::Job::Runtime::Execution.find(event[:execution_id])
end

#fire(source, event_type_name, properties = {}, options = {}) ⇒ Object 



164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
 
# File 'lib/tengine/job/runtime/signal.rb', line 164

def fire(source, event_type_name, properties = {}, options = {})
  case source
  when Tengine::Job::Runtime::Execution then
    properties[:execution_id] ||= source.id.to_s
    properties[:root_jobnet_id] ||= source.root_jobnet.id.to_s
    properties[:root_jobnet_name_path] ||= source.root_jobnet.name_path
    properties[:target_jobnet_id] ||= source.root_jobnet.id.to_s
    properties[:target_jobnet_name_path] ||= source.root_jobnet.name_path
  else
    properties[:execution_id] ||= self.execution.id.to_s
    properties[:root_jobnet_id] ||= source.root.id.to_s
    properties[:root_jobnet_name_path] ||= source.root.name_path
  end
  # デバッグ用
  # properties[:target_jobnet_name] = source.root.vertex(properties[:target_jobnet_id]).name_path
  options ||= {}
  options[:properties] = properties
  properties.each do |key, value|
    if value.is_a?(Moped::BSON::ObjectId)
      properties[key] = value.to_s
    end
  end
  @reservations << Reservation.new(source, event_type_name, options)
end

#leave(obj, action = :transmit) ⇒ Object 



117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
 
# File 'lib/tengine/job/runtime/signal.rb', line 117

def leave(obj, action = :transmit)
  @paths << obj
  begin
    if obj.is_a?(Tengine::Job::Runtime::Edge)
      if obj.destination.is_a?(Tengine::Job::Runtime::NamedVertex)
        self.call_later do
          cache(obj.destination).send(action, self)
        end
      else
        cache(obj.destination).send(action, self)
      end
    elsif obj.is_a?(Tengine::Job::Runtime::Vertex)
      obj.next_edges.each do |edge|
        # cache_list
        with_paths_backup{ cache(edge).send(action, self) }
      end
    else
      raise Tengine::Job::Runtime::Signal::Error, "leaving unsupported object: #{obj.inspect}"
    end
  rescue Tengine::Job::Runtime::Signal::Error => e
    puts "[#{e.class.name}] #{e.message}\nsignal.paths: #{@paths.inspect}"
    raise e
  end
end

#process_callbacks ⇒ Object 



46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/tengine/job/runtime/signal.rb', line 46

def process_callbacks
  until self.callbacks.empty?
Tengine.logger.debug("-" * 20)
    callbacks.shift.call
  end

  while self.callback
    block, @callback = @callback, nil
    block.call
  end
end

#remember(obj) ⇒ Object 



58
59
60
61
62
63
64
65
66
67
68
69
 
# File 'lib/tengine/job/runtime/signal.rb', line 58

def remember(obj)
  if obj.is_a?(Array)
    obj.each{|o| remember(o)}
  else
    return nil if obj.nil?
    key = cache_key(obj)
    cached = cache(*key)
    return cached if cached
    @cache[key] = obj
  end
  obj
end

#remember_all(vertex) ⇒ Object 



88
89
90
91
 
# File 'lib/tengine/job/runtime/signal.rb', line 88

def remember_all(vertex)
  v = Tengine::Job::Structure::Visitor::AllWithEdge.new{|obj| remember(obj) }
  vertex.accept_visitor(v)
end

#reset ⇒ Object 



28
29
30
31
32
33
34
35
 
# File 'lib/tengine/job/runtime/signal.rb', line 28

def reset
  @cache = {}
  @paths = []
  @reservations = []
  @data = nil
  @callback = nil
  @callbacks = []
end

#with_paths_backup ⇒ Object 



142
143
144
145
146
147
148
149
 
# File 'lib/tengine/job/runtime/signal.rb', line 142

def with_paths_backup
  paths_backup = @paths.dup
  begin
    yield if block_given?
  ensure
    @paths = paths_backup
  end
end