1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
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
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
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
|
#!/usr/bin/env ruby
banner = ['WWWWatch', 'version 1.0.0', '2003-2006', 'PRAGMA ADE/POD']
$: << File.expand_path(File.dirname($0)) ; $: << File.join($:.last,'lib') ; $:.uniq!
require 'base/switch'
require 'base/logger'
require 'www/common'
require 'monitor'
require 'fileutils'
require 'ftools'
require 'tempfile'
require 'timeout'
require 'thread'
class Watch < Monitor
include Common
@@session_prefix = ''
@@check_factor = 4
@@process_timeout = 1*60*60
@@fast_wait_loop = false
@@session_line = /^\s*(?![\#\%])(.*?)\s*\=\s*(.*?)\s*$/o
@@session_begin = 'begin exa session'
@@session_end = 'end exa session'
attr_accessor :root_path, :work_path, :create, :cache_path, :delay, :max_threads, :max_age, :verbose
def initialize(logger) # we need to register all @vars here becase of the monitor
@threads = Hash.new
@files = Array.new
@stats = Hash.new
@skips = Hash.new
@root_path = ''
@work_path = Dir.tmpdir
@cache_path = @work_path
@last_action = Time.now
@delay = 1
@max_threads = 5
@max_age = @@process_timeout
@logger = logger
@verbose = false
@create = false
@onlyonerun = false
# [:INT, :TERM, :EXIT].each do |signal|
# trap(signal) do
# kill
# exit # rescue false
# end
# end
# at_exit do
# kill
# end
end
def trace
if @verbose && @logger then
@logger.report("exception: #{$!})")
$@.each do |t|
@logger.report(">> #{t}")
end
end
end
def report(str)
@logger.report(str) if @logger
end
def setup
@threads = Hash.new
@files = Array.new
@stats = Hash.new
@skips = Hash.new
@root_path = File.expand_path(File.join(File.dirname(Dir.pwd),'.')) if @root_path.empty?
@work_path = File.expand_path(File.join(@root_path,'work','watch')) if @work_path.empty?
# @cache_path = File.expand_path(File.join(@root_path,'work','cache')) if @cache_path.empty?
@cache_path = File.expand_path(File.join(File.dirname(@work_path),'cache')) if @cache_path.empty?
if @create then
begin File.makedirs(@work_path) ; rescue ; end
begin File.makedirs(@cache_path) ; rescue ; end
end
unless File.writable?(@work_path) then
@work_path = File.expand_path(File.join(Dir.tmpdir,'work','watch'))
if @create then
begin File.makedirs(@work_path) ; rescue ; end
end
end
unless File.writable?(@cache_path) then
@cache_path = File.expand_path(File.join(Dir.tmpdir,'work','cache'))
if @create then
begin File.makedirs(@cache_path) ; rescue ; end
end
end
unless File.writable?(@work_path) then
puts "no valid work path: #{@work_path}"
exit! rescue false # no checking, no at_exit done
end
unless File.writable?(@cache_path) then
puts "no valid cache path: #{@cache_path}" ; # no reason to exit
end
@last_action = Time.now
report("watching path #{@work_path}") if @verbose
end
def lock(lck)
begin
report("watchdog: locking #{lck}") if @verbose
File.open(lck,'w') do |f|
f << Time.now
end
rescue
trace
end
end
def unlock(lck)
begin
report("watchdog: unlocking #{lck}") if @verbose
File.delete(lck)
rescue
trace
end
end
def kill
@threads.each do |t|
t.kill rescue false
end
end
def restart
@files = Array.new
@skips = Hash.new
@stats = Hash.new
kill # threads
end
def collect
begin
@files = Array.new
Dir.glob("#{@work_path}/#{@@session_prefix}*.ses").each do |sessionfile|
sessionfile = File.expand_path(sessionfile)
begin
if @threads.key?(sessionfile) then
# leave alone
elsif (Time.now - File.mtime(sessionfile)) > @max_age.to_i then
# delete
FileUtils::rm_r(sessionfile) rescue false
FileUtils::rm_r(sessionfile.sub(/ses$/,'dir')) rescue false
FileUtils::rm_r(sessionfile.sub(/ses$/,'lck')) rescue false
begin
FileUtils::rm_r(File.join(@cache_path, File.basename(sessionfile.sub(/ses$/,'dir'))))
rescue
report("watchdog: problems in cache cleanup #{$!}") # if @verbose
end
@stats.delete(sessionfile) rescue false
@skips.delete(sessionfile) rescue false
report("watchdog: removing session #{sessionfile}") if @verbose
elsif ! @skips.key?(sessionfile) then
@files << sessionfile
report("watchdog: checking session #{sessionfile}") if @verbose
end
rescue
# maybe purged in the meantime
end
end
rescue
if File.directory?(@work_path) then
@files = Array.new
else
# maybe dir is deleted (manual cleanup)
restart
end
end
begin
Dir.glob("#{@cache_path}/*.dir").each do |dirname|
begin
if (Time.now - File.mtime(dirname)) > @max_age.to_i then
begin
FileUtils::rm_r(dirname)
rescue
report("watchdog: problems in cache cleanup #{$!}") # if @verbose
end
end
rescue
# maybe purged in the meantime
end
end
rescue
end
end
def purge
begin
Dir.glob("#{@work_path}/#{@@session_prefix}*").each do |sessionfile|
sessionfile = File.expand_path(sessionfile)
begin
if (Time.now - File.mtime(sessionfile)) > @max_age.to_i then
begin
if FileTest.directory?(sessionfile) then
FileUtils::rm_r(sessionfile)
else
File.delete(sessionfile)
end
rescue
end
begin
@stats.delete(sessionfile)
@skips.delete(sessionfile)
rescue
end
report("watchdog: purging session #{sessionfile}") if @verbose
end
rescue
# maybe purged in the meantime
end
end
rescue
end
end
def loaded_session_data(filename)
begin
if data = IO.readlines(filename) then
return data if (data.first =~ /^[\#\%]\s*#{@@session_begin}/o) && (data.last =~ /^[\#\%]\s*#{@@session_end}/o)
end
rescue
trace
end
return nil
end
def load(sessionfile)
# we assume that we get an exception when the file is locked
begin
if data = loaded_session_data(sessionfile) then
report("watchdog: loading session #{sessionfile}") if @verbose
vars = Hash.new
data.each do |line|
begin
if line.chomp =~ /^(.*?)\s*\=\s*(.*?)\s*$/o then
key, value = $1, $2
vars[key] = value
end
rescue
end
end
return vars
else
return nil
end
rescue
trace
return nil
end
end
def save(sessionfile, vars)
begin
report("watchdog: saving session #{sessionfile}") if @verbose
if @stats.key?(sessionfile) then
@stats[sessionfile] = File.mtime(sessionfile)
elsif @stats[sessionfile] == File.mtime(sessionfile) then
else
# construct data first
str = "\# #{@@session_begin}\n"
for k,v in vars do
str << "#{k}=#{v}\n"
end
str << "\# #{@@session_end}\n"
# save as fast as possible
File.open(sessionfile,'w') do |f|
f.puts(str)
end
end
rescue
report("watchdog: unable to save session #{sessionfile}") if @verbose
trace
return false
else
return true
end
end
def launch
begin
@files.each do |sessionfile|
if @threads.length < @max_threads then
begin
if ! @skips.key?(sessionfile) && (vars = load(sessionfile)) then
if (id = vars['id']) && vars['status'] then
if vars['status'] == 'running: background' then
@last_action = Time.now
@threads[sessionfile] = Thread.new(vars, sessionfile) do |vars, sessionfile|
begin
report("watchdog: starting thread #{sessionfile}") if @verbose
dir = File.expand_path(sessionfile.sub(/ses$/,'dir'))
lck = File.expand_path(sessionfile.sub(/ses$/,'lck'))
start_of_run = Time.now
start_of_job = start_of_run.dup
max_time = @max_age
begin
start_of_job = vars['starttime'].to_i || start_of_run
start_of_job = start_of_run if start_of_job == 0
rescue
start_of_job = Time.now
end
begin
max_runtime = vars['maxtime'].to_i || @max_age
max_runtime = @max_age if max_runtime == 0
max_runtime = max_runtime - (Time.now.to_i - start_of_job.to_i)
rescue
max_runtime = @max_age
end
lock(lck)
if max_runtime > 0 then
command = vars['command'] || ''
if ! command.empty? then
vars['status'] = 'running: busy'
vars['timeout'] = max_runtime.to_s
save(sessionfile,vars)
timeout(max_runtime) do
begin
command = command_string(dir,command,'process.log')
report("watchdog: #{command}") if @verbose
system(command)
rescue TimeoutError
vars['status'] = 'running: timeout'
rescue
trace
vars['status'] = 'running: aborted'
else
vars['status'] = 'running: finished'
vars['runtime'] = sprintf("%.02f",(Time.now - start_of_run))
vars['endtime'] = Time.now.to_i.to_s
end
end
else
vars['status'] = 'running: aborted' # no command
end
else
vars['status'] = 'running: aborted' # not enough time
end
save(sessionfile,vars)
unlock(lck)
report("watchdog: ending thread #{sessionfile}") if @verbose
@threads.delete(sessionfile)
rescue
trace
end
end
else
report("watchdog: skipping - id (#{vars['id']}) / status (#{vars['status']})") if @verbose
end
if @onlyonerun then
@skips[sessionfile] = true
else
@skips.delete(sessionfile)
end
else
# not yet ok
end
else
# maybe a lock
end
rescue
trace
end
else
break
end
end
rescue
trace
end
end
def wait
begin
# report(Time.now.to_s) if @verbose
loop do
@threads.delete_if do |k,v|
begin
v == nil || v.stop?
rescue
true
else
false
end
end
if @threads.length == @max_threads then
if @delay > @max_threads then
sleep(@delay)
else
sleep(@max_threads)
end
break if @@fast_wait_loop
else
sleep(@delay)
break
end
end
rescue
trace
end
end
def check
begin
time = Time.now
if (time - @last_action) > @@check_factor*@max_age then
report("watchdog: cleanup") if @verbose
@stats = Hash.new
@last_action = time
kill
end
rescue
trace
end
end
def cycle
loop do
begin
collect
launch
wait
check
rescue
trace
report("watchdog: some problem, restarting loop")
end
end
end
end
class Commands
include CommandBase
def watch
if watch = setup then
watch.cycle
else
report("provide valid work path")
end
end
def main
watch
end
private
def setup
if watch = Watch.new(logger) then
watch.root_path = @commandline.option('root')
watch.work_path = @commandline.option('work')
watch.cache_path = @commandline.option('cache')
watch.create = @commandline.option('create')
watch.verbose = @commandline.option('verbose')
begin
watch.max_threads = @commandline.option('threads').to_i
rescue
watch.max_threads = 5
end
watch.setup
end
return watch
end
end
logger = Logger.new(banner.shift)
commandline = CommandLine.new
commandline.registervalue('root', '')
commandline.registervalue('work', '')
commandline.registervalue('cache', '')
commandline.registervalue('threads', '5')
commandline.registerflag('create')
commandline.registeraction('watch', '[--work=path] [--root=path] [--create]')
commandline.registerflag('verbose')
commandline.registeraction('help')
commandline.registeraction('version')
commandline.expand
Commands.new(commandline,logger,banner).send(commandline.action || 'main')
|