diff --git a/pbatch/README b/pbatch/README index e9f1a72..6af5be9 100644 --- a/pbatch/README +++ b/pbatch/README @@ -1,13 +1,31 @@ -- config file gebruiken -- meerdere parallele batches -- er mogen er X tegelijk draaien +====== += TODO +====== + - prioriteiten? mogelijk in de toekomst... mbv jobs naamgeving? prefix H voor high, N voor normal en L voor low - noodstop +processors pollen zelf de jobs dir voor nieuwe taken? +master zorgt voor het draaien van juiste aantal processors +master kan processors een signaal geven dat ze geen nieuwe mogen opstarten + +threading? + +jobs die input uit een file willen? + +====== += Done +====== + +- iets inbouwen dat jobnummering fijn blijft ook al komen er digits bij: + 7 8 9 10 + ineens wordt 10 voor 7 gescheduled... +- config file gebruiken +- meerdere parallele batches +- er mogen er X tegelijk draaien - scheduling door jobs uit directory naar processing dir over te hevelen - indien failure naar faildir of nogmaals (X keer?) -- wanneer klaar incl resultaten naar done dir - +- dirs bijv. zo: jobs processors/job#/job @@ -20,15 +38,4 @@ failure/job#/rundir failure/job#/output monitor/processor# symlink naar output current job -processors pollen zelf de jobs dir voor nieuwe taken? -master zorgt voor het draaien van juiste aantal processors -master kan processors een signaal geven dat ze geen nieuwe mogen opstarten - -threading? - -jobs die input uit een file willen? - - -iets inbouwen dat jobnummering fijn blijft ook al komen er digits bij: -7 8 9 10 -ineens wordt 10 voor 7 gescheduled... + wanneer klaar incl resultaten naar done dir diff --git a/pbatch/pb.rb b/pbatch/pb.rb index e337e3e..8b0e602 100755 --- a/pbatch/pb.rb +++ b/pbatch/pb.rb @@ -95,14 +95,17 @@ def increasejobscounter file.seek(0) file.puts jobscounter } - File.new(COUNTERFILE, "r+").flock(File::LOCK_UN) + File.new(COUNTERFILE).flock(File::LOCK_UN) else + # XXX wellicht eerst een touch ofzo doen en dan locken voor de schrijf loop + # geopend wordt + FileUtils.touch(COUNTERFILE) + File.new(COUNTERFILE).flock(File::LOCK_EX) File.open(COUNTERFILE, "w"){|file| - File.new(COUNTERFILE).flock(File::LOCK_EX) jobscounter = 1 file.puts jobscounter } - File.new(COUNTERFILE, "r+").flock(File::LOCK_UN) + File.new(COUNTERFILE).flock(File::LOCK_UN) end return jobscounter end @@ -150,26 +153,6 @@ EOT end end -def master - # als ik threading ga gebruiken is deze dus niet nodig - File.open(RUNNINGFILE, "w").flock(File::LOCK_EX){|file| - file.puts "0" - } - -# echo 0 >"${basedir}/$runningfile" -# -# while sleep $waittime; do -# if [ $(cat "${basedir}/${runningfile}") -lt $maxproc ]; then -# # via ls omdat anders de globbing klote is -# jl=$(ls ${basedir}/jobs/) -# if [ -n "$jl" ]; then -# ${pbcommand} process & -# fi -# readconfig -# fi -# done -end - def deljob jobnumber = ARGV[0] if jobnumber.nil? @@ -226,24 +209,76 @@ def changeprio puts "unimplemented" end +def incrementrunning + # als ik threading ga gebruiken is dit niet meer nodig (want geen runningfile nodig) + File.new(RUNNINGFILE).flock(File::LOCK_EX) + File.open(RUNNINGFILE, "r+"){|file| + running = file.read + running = running.to_i + 1 + file.seek(0) + file.trunc + file.puts running + } + File.new(RUNNINGFILE).flock(File::LOCK_UN) +end + +def decrementrunning + # als ik threading ga gebruiken is dit niet meer nodig (want geen runningfile nodig) + File.new(RUNNINGFILE).flock(File::LOCK_EX) + File.open(RUNNINGFILE, "r+"){|file| + running = file.read + running = running.to_i - 1 + if running < 0 + running = 0 + end + file.seek(0) + file.trunc + file.puts running + } + File.new(RUNNINGFILE).flock(File::LOCK_UN) +end + +def getjob + Dir.entries("#{@basedir}/#{JOBSDIR}").each{|file| + if file == '.' or file == '..' + next + else + if FileTest.file?("#{@basedir}/#{JOBSDIR}/#{file}") + Dir.mkdir("#{@basedir}/#{PROCDIR}/#{file}") + FileUtils.mv("#{@basedir}/#{JOBSDIR}/#{file}", "#{@basedir}/#{PROCDIR}/#{file}/job") + return file + end + end + } +end + +def failjob(jobnumber) + FileUtils.mv("#{@basedir}/#{PROCDIR}/#{jobnumber}", "#{@basedir}/#{FAILDIR}") +end + +def finishjob(jobnumber) + FileUtils.mv("#{@basedir}/#{PROCDIR}/#{jobnumber}", "#{@basedir}/#{RESULTDIR}") +end + def runprocessor -# incrementrunning -# sleep 1 -# runjob=$(getjob) -# if [ -n "${runjob}" ]; then -# echo "Starting job: ${runjob}" -# owd=$(pwd) -# cd "${basedir}/processors/${runjob}" + incrementrunning + sleep 1 # XXX niet nodig als locking echt werkt (of bij threading) + runjob = getjob + if ! runjob.nil? + puts "Starting job: #{runjob}" + owd = Dir.pwd + Dir.cd("#{@basedir}/#{PROCDIR}/#{runjob}") +# XXX IPC nog even uitzoeken... popen3 ? # if "${basedir}/processors/${runjob}/job" 2>&1 > "${basedir}/processors/${runjob}/output" ;then # finishjob ${runjob} -# echo "Finished job: ${runjob}" + puts "Finished job: #{runjob}" # else # failjob ${runjob} -# echo "Failed job: ${runjob}" + puts "Failed job: #{runjob}" # fi -# cd $owd -# fi -# decrementrunning + Dir.cd(owd) + end + decrementrunning end def stop @@ -254,7 +289,36 @@ def stopall puts "unimplemented" end -pbcommand = $0 +def master + if Process.euid != 0 + puts "If your jobs need root priviledges (XXX spelling) you want to run this as root" + end + + # als ik threading ga gebruiken is dit niet meer nodig (want geen runningfile nodig) + if ! FileTest.file?(RUNNINGFILE) + FileUtils.touch(RUNNINGFILE) + end + File.new(RUNNINGFILE).flock(File::LOCK_EX) + File.open(RUNNINGFILE, "w"){|file| + file.puts "0" + } + File.new(RUNNINGFILE).flock(File::LOCK_UN) + +# while sleep $waittime; do +# if [ $(cat "${basedir}/${runningfile}") -lt $maxproc ]; then +# # via ls omdat anders de globbing klote is +# jl=$(ls ${basedir}/jobs/) +# if [ -n "$jl" ]; then +# ${pbcommand} process & +# fi +# readconfig +# fi +# done +end + +# XXX SIGHUP handler die config opnieuw inleest + +@pbcommand = $0 command = ARGV.shift @basedir = Dir.pwd