Class: ConcurrentWorker::Worker

Inherits:
Object
  • Object
show all
Defined in:
lib/concurrent_worker.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(*args, **options, &work_block) ⇒ Worker

Returns a new instance of Worker.



79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
 
# File 'lib/concurrent_worker.rb', line 79

def initialize(*args, **options, &work_block)
  @args = args
  @options = options
  set_block(:work_block, &work_block) if work_block
  
  @state = :idle
  @result_callbacks = []
  @retired_callbacks = []

  @snd_queue_max = @options[:snd_queue_max] || 2
  @req_counter = RequestCounter.new
  @options[ :result_callback_interrupt ]  ||= :immediate 
  @options[ :retired_callback_interrupt ] ||= :immediate 

  case @options[:type]
  when :process
    class << self
      include ConcurrentProcess
    end
  when :thread
    class << self
      include ConcurrentThread
    end
  else
    class << self
      include ConcurrentThread
    end
  end
end

Instance Attribute Details

#channel ⇒ Object

Returns the value of attribute channel.



56
57
58
 
# File 'lib/concurrent_worker.rb', line 56

def channel
  @channel
end

#req_counter ⇒ Object (readonly)

Worker : worker class

+cncr_block         : concurrent processing block : thread(as ConcurrentThread)/process(as ConcurrentProcess)
  +base_block       : user defined preparation to exec 'work block'
    +loop_block     : loop of receiving request and exec 'work block'
      +work_block   : user requested work

These blocks are executed with ‘instance_exec’ method of worker, so that they can share instance variables:@xxxx.



67
68
69
 
# File 'lib/concurrent_worker.rb', line 67

def req_counter
  @req_counter
end

#snd_queue_max ⇒ Object (readonly)

Worker : worker class

+cncr_block         : concurrent processing block : thread(as ConcurrentThread)/process(as ConcurrentProcess)
  +base_block       : user defined preparation to exec 'work block'
    +loop_block     : loop of receiving request and exec 'work block'
      +work_block   : user requested work

These blocks are executed with ‘instance_exec’ method of worker, so that they can share instance variables:@xxxx.



67
68
69
 
# File 'lib/concurrent_worker.rb', line 67

def snd_queue_max
  @snd_queue_max
end

Instance Method Details

#add_callback(&callback) ⇒ Object 



109
110
111
112
 
# File 'lib/concurrent_worker.rb', line 109

def add_callback(&callback)
  raise "block is nil" unless callback
  @result_callbacks.push(callback)
end

#add_retired_callback(&callback) ⇒ Object 



128
129
130
131
 
# File 'lib/concurrent_worker.rb', line 128

def add_retired_callback(&callback)
  raise "block is nil" unless callback
  @retired_callbacks.push(callback)
end

#call_result_callbacks(args) ⇒ Object 



117
118
119
120
121
122
123
124
125
126
 
# File 'lib/concurrent_worker.rb', line 117

def call_result_callbacks(args)
  Thread.handle_interrupt(Object => :never) do        
    Thread.handle_interrupt(Object => @options[ :result_callback_interrupt ] ) do
      @result_callbacks.each do |callback|
        callback.call(*args)
      end
    end
    @req_counter.pop
  end
end

#call_retired_callbacks ⇒ Object 



136
137
138
139
140
141
142
 
# File 'lib/concurrent_worker.rb', line 136

def call_retired_callbacks
  Thread.handle_interrupt(Object => @options[ :retired_callback_interrupt ] ) do
    @retired_callbacks.each do |callback|
      callback.call
    end
  end
end

#clear_callbacks ⇒ Object 



113
114
115
 
# File 'lib/concurrent_worker.rb', line 113

def clear_callbacks
  @result_callbacks.clear
end

#clear_retired_callbacks ⇒ Object 



132
133
134
 
# File 'lib/concurrent_worker.rb', line 132

def clear_retired_callbacks
  @retired_callbacks.clear
end

#join ⇒ Object 



215
216
217
218
219
 
# File 'lib/concurrent_worker.rb', line 215

def join
  @req_counter.wait_until_less_than(1)
  quit
  wait_cncr_proc
end

#queue_available? ⇒ Boolean

Returns:

  • (Boolean) 


75
76
77
 
# File 'lib/concurrent_worker.rb', line 75

def queue_available?
  !queue_closed? && @req_counter.size < @snd_queue_max
end

#queue_closed? ⇒ Boolean

Returns:

  • (Boolean) 


69
70
71
 
# File 'lib/concurrent_worker.rb', line 69

def queue_closed?
  @req_counter.closed?
end

#queue_empty? ⇒ Boolean

Returns:

  • (Boolean) 


72
73
74
 
# File 'lib/concurrent_worker.rb', line 72

def queue_empty?
  !queue_closed? && @req_counter.size == 0
end

#quit ⇒ Object 



206
207
208
209
210
211
212
213
 
# File 'lib/concurrent_worker.rb', line 206

def quit
  begin 
    send_req([])
    true
  rescue ClosedQueueError, IOError
    false
  end
end

#req(*args, &work_block) ⇒ Object 



192
193
194
195
196
197
198
199
200
201
202
203
204
 
# File 'lib/concurrent_worker.rb', line 192

def req(*args, &work_block)
  unless @state == :run
    run
  end
  @req_counter.wait_until_less_than(@snd_queue_max)
  begin 
    @req_counter.push([args, work_block])
    send_req([args, work_block])
    true
  rescue ClosedQueueError, IOError
    false
  end
end

#run ⇒ Object 



185
186
187
188
189
190
 
# File 'lib/concurrent_worker.rb', line 185

def run
  @state = :run
  set_default_loop_block unless defined?(@loop_block) && @loop_block
  set_default_base_block unless defined?(@base_block) && @base_block
  cncr_block
end

#set_block(symbol, &block) ⇒ Object 



144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
 
# File 'lib/concurrent_worker.rb', line 144

def set_block(symbol, &block)
  raise "block is nil" unless block
  
  unless [:base_block, :loop_block, :work_block].include?(symbol)
    raise symbol.to_s + " is not used as worker block"
  end
  
  worker_block = Proc.new do |*args|
    self.instance_exec(*args, &block)
  end
  instance_variable_set("@" + symbol.to_s, worker_block)
  
  define_singleton_method("yield_" + symbol.to_s) do |*args|
    blk = instance_variable_get("@" + symbol.to_s)
    if blk
      blk.call(*args)
    else
      raise "block " + symbol.to_s + " is not defined"
    end
  end
end

#set_default_base_block ⇒ Object 



179
180
181
182
183
 
# File 'lib/concurrent_worker.rb', line 179

def set_default_base_block
  set_block(:base_block) do
    yield_loop_block
  end
end

#set_default_loop_block ⇒ Object 



166
167
168
169
170
171
172
173
174
175
176
177
 
# File 'lib/concurrent_worker.rb', line 166

def set_default_loop_block
  set_block(:loop_block) do
    loop do
      break if (req = receive_req).empty?
      (args, work_block) = req
      if work_block
        set_block(:work_block, &work_block)
      end
      send_res(yield_work_block(*args))
    end
  end
end