Class: AlWorker::Program

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

Overview

プログラム実行

Constant Summary collapse

@@programs =

Returns 実行中リスト.

Returns:

  • (Hash) —

    実行中リスト

{}
@@mutex_programs =

Returns 実行中リストmutex.

Returns:

  • (Mutex) —

    実行中リストmutex

Mutex.new

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(program = nil, *args) ⇒ Program

constructor

Parameters:

  • program (String) (defaults to: nil) —

    実行するプログラム名

  • args (Array) —

    引数



113
114
115
116
117
118
119
120
121
122
123
124
 
# File 'lib/al_worker_program.rb', line 113

def initialize( program = nil, *args )
  @state = :none
  @program = program.dup
  @args = args
  @env = {}
  @options = {}
  @mode_sync = :sync
  @mode_single = :single
  @at_end = nil
  @at_system_exit = :kill
  @mutex_wait = Mutex.new
end

Instance Attribute Details

#args ⇒ Array

Returns 引数.

Returns:

  • (Array) —

    引数



77
78
79
 
# File 'lib/al_worker_program.rb', line 77

def args
  @args
end

#at_end ⇒ Proc

Returns プログラム実行終了時の動作.

Returns:

  • (Proc) —

    プログラム実行終了時の動作



92
93
94
 
# File 'lib/al_worker_program.rb', line 92

def at_end
  @at_end
end

#at_system_exit ⇒ Symbol

Returns プログラムの終了時　終了させられる(:kill)／留まる(:stay).

Returns:

  • (Symbol) —

    プログラムの終了時　終了させられる(:kill)／留まる(:stay)



101
102
103
 
# File 'lib/al_worker_program.rb', line 101

def at_system_exit
  @at_system_exit
end

#env ⇒ Hash

Returns 環境変数.

Returns:

  • (Hash) —

    環境変数



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

def env
  @env
end

#mode_single ⇒ Symbol

Returns 単独実行　単独(:single)／複数（:plural).

Returns:

  • (Symbol) —

    単独実行　単独(:single)／複数（:plural)



89
90
91
 
# File 'lib/al_worker_program.rb', line 89

def mode_single
  @mode_single
end

#mode_sync ⇒ Symbol

Returns 実行終了イベント動作　同期(:sync)／非同期(:async).

Returns:

  • (Symbol) —

    実行終了イベント動作　同期(:sync)／非同期(:async)



86
87
88
 
# File 'lib/al_worker_program.rb', line 86

def mode_sync
  @mode_sync
end

#mutex_wait ⇒ Mutex

Returns waitメソッド同期用.

Returns:

  • (Mutex) —

    waitメソッド同期用



104
105
106
 
# File 'lib/al_worker_program.rb', line 104

def mutex_wait
  @mutex_wait
end

#name ⇒ String

Returns 内部識別名.

Returns:

  • (String) —

    内部識別名



74
75
76
 
# File 'lib/al_worker_program.rb', line 74

def name
  @name
end

#options ⇒ Hash

Returns 実行オプション.

Returns:

  • (Hash) —

    実行オプション



83
84
85
 
# File 'lib/al_worker_program.rb', line 83

def options
  @options
end

#pid ⇒ Integer (readonly)

Returns プロセスID.

Returns:

  • (Integer) —

    プロセスID



95
96
97
 
# File 'lib/al_worker_program.rb', line 95

def pid
  @pid
end

#process_status ⇒ Process::Status (readonly)

Returns プロセス終了ステータス.

Returns:

  • (Process::Status) —

    プロセス終了ステータス



98
99
100
 
# File 'lib/al_worker_program.rb', line 98

def process_status
  @process_status
end

#program ⇒ String

Returns プログラム名.

Returns:

  • (String) —

    プログラム名



71
72
73
 
# File 'lib/al_worker_program.rb', line 71

def program
  @program
end

#state ⇒ Symbol (readonly)

Returns 状態 ( :none :run :done :error ).

Returns:

  • (Symbol) —

    状態 ( :none :run :done :error )



68
69
70
 
# File 'lib/al_worker_program.rb', line 68

def state
  @state
end

Class Method Details

.kill(program, signal = :TERM) ⇒ Object

プログラム名を指定して実行中断

Parameters:

  • program (String) —

    中断するプログラム名

  • signal (Integer, String, Symbol) (defaults to: :TERM) —

    シグナル



56
57
58
59
60
61
62
63
64
 
# File 'lib/al_worker_program.rb', line 56

def self.kill( program, signal = :TERM )
  @@mutex_programs.synchronize {
    @@programs.each_value do |pgm|
      if pgm.program == program
        Process.kill( signal, pgm.pid ) rescue 0
      end
    end
  }
end

.programs ⇒ Object

プログラム実行中リストのアクセッサ



26
27
28
 
# File 'lib/al_worker_program.rb', line 26

def self.programs
  return @@programs
end

.run(program, *args) { ... } ⇒ AlWorker::Program, NilClass

プログラム実行

Parameters:

  • program (String) —

    実行するプログラム名

  • args (Array) —

    引数

Yields:

  • プログラム終了時の動作

Returns:



40
41
42
43
44
45
46
47
 
# File 'lib/al_worker_program.rb', line 40

def self.run( program, *args, &block )
  pgm = self.new( program, *args )
  if pgm.run( &block )
    return pgm
  else
    return nil
  end
end

Instance Method Details

#alive? ⇒ Boolean

実行中か？

Returns:

  • (Boolean) —

    実行中か？



226
227
228
 
# File 'lib/al_worker_program.rb', line 226

def alive?()
  return @state == :run
end

#kill(signal = :TERM) ⇒ Object

実行中断

Parameters:

  • signal (Integer, String, Symbol) (defaults to: :TERM) —

    シグナル



236
237
238
239
 
# File 'lib/al_worker_program.rb', line 236

def kill( signal = :TERM )
  return nil  if ! @pid
  Process.kill( signal, @pid ) rescue nil
end

#run(*arg) { ... } ⇒ Boolean

実行する

Parameters:

  • arg (Array) —

    ブロック内に渡す引数

Yields:

  • タイムアップ時の動作

Returns:

  • (Boolean) —

    開始できたか？



134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
 
# File 'lib/al_worker_program.rb', line 134

def run( *arg )
  # ２重実行防止
  return false  if @state == :run
  if @mode_single == :single
    @name ||= @program
    @@mutex_programs.synchronize {
      return false  if @@programs[@name]
      @@programs[@name] = self
    }
  end

  # プログラム実行
  @state = :run
  begin
    if ! @env.empty? || ! @options.empty?
      @pid = spawn( @env, @program, *@args, @options )
    else
      @pid = spawn( @program, *@args )
    end

  rescue =>ex
    @state = :error
    @@mutex_programs.synchronize {
      @@programs.delete( @name )
    }
    raise ex
  end

  if @mode_single != :single
    @name = "#{program}##{pid}"
    @@mutex_programs.synchronize {
      @@programs[@name] = self
    }
  end

  @thread = Thread.start( arg ) { |arg|
    # プロセス終了待ち
    @mutex_wait.synchronize {
      (pid,@process_status) = Process.waitpid2( @pid )
      @state = :done
    }
    @@mutex_programs.synchronize {
      @@programs.delete( @name )
    }

    # 終了時、ブロックが与えられて入れば実行
    if block_given?
      begin
        if mode_sync == :sync
          AlWorker.mutex_sync.synchronize { yield( *arg ) }
        else
          yield( *arg )
        end

      rescue Exception => ex
        raise ex  if ex.class == SystemExit
        AlWorker.log( ex )
      end
    end

    # 終了時、Procオブジェクトが与えられていれば実行
    if @at_end
      begin
        if mode_sync == :sync
          AlWorker.mutex_sync.synchronize { @at_end.call( self, *arg ) }
        else
          @at_end.call( self, *arg )
        end

      rescue Exception => ex
        raise ex  if ex.class == SystemExit
        AlWorker.log( ex )
      end
    end
  }

  # waitpid2が実行されたことを確実にする。
  while @mutex_wait.try_lock
    @mutex_wait.unlock()
    break if @state != :run
    Thread.pass()
  end

  return true
end

#wait(timeout = nil) ⇒ NilClass

実行終了を待つ

Parameters:

  • timeout (Numeric) (defaults to: nil) —

    待ち時間タイムアウト

Returns:

  • (NilClass) —

    nilならタイムアウト



248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
 
# File 'lib/al_worker_program.rb', line 248

def wait( timeout = nil )
  return true  if ! @thread

  if timeout
    begin
      Timeout::timeout( timeout ) {
        @mutex_wait.synchronize {
          # waiting waitpid2 in run().
        }
      }
    rescue Timeout::Error
      return nil
    end
  else
    @mutex_wait.synchronize {
      # waiting waitpid2 in run().
    }
  end
  return true
end