Object
# File shell/process-controller.rb, line 34 def activate(pc) process_controllers_exclusive do @ProcessControllers[pc] ||= 0 @ProcessControllers[pc] += 1 end end
# File shell/process-controller.rb, line 51 def each_active_object process_controllers_exclusive do for ref in @ProcessControllers.keys yield ref end end end
# File shell/process-controller.rb, line 41 def inactivate(pc) process_controllers_exclusive do if @ProcessControllers[pc] if (@ProcessControllers[pc] -= 1) == 0 @ProcessControllers.delete(pc) end end end end
# File shell/process-controller.rb, line 143 def active_job?(job) @jobs_sync.synchronize(:SH) do @active_jobs.include?(job) end end
# File shell/process-controller.rb, line 79 def active_jobs @active_jobs end
# File shell/process-controller.rb, line 93 def active_jobs_exist? @jobs_sync.synchronize(:SH) do @active_jobs.empty? end end
schedule a command
# File shell/process-controller.rb, line 106 def add_schedule(command) @jobs_sync.synchronize(:EX) do ProcessController.activate(self) if @active_jobs.empty? start_job command else @waiting_jobs.push(command) end end end
# File shell/process-controller.rb, line 70 def jobs jobs = [] @jobs_sync.synchronize(:SH) do jobs.concat @waiting_jobs jobs.concat @active_jobs end jobs end
# File shell/process-controller.rb, line 87 def jobs_exist? @jobs_sync.synchronize(:SH) do @active_jobs.empty? or @waiting_jobs.empty? end end
kill a job
# File shell/process-controller.rb, line 161 def kill_job(sig, command) @jobs_sync.synchronize(:SH) do if @waiting_jobs.delete command ProcessController.inactivate(self) return elsif @active_jobs.include?(command) begin r = command.kill(sig) ProcessController.inactivate(self) rescue print "Shell: Warn: $!\n" if @shell.verbose? return nil end @active_jobs.delete command r end end end
simple fork
# File shell/process-controller.rb, line 194 def sfork(command, &block) pipe_me_in, pipe_peer_out = IO.pipe pipe_peer_in, pipe_me_out = IO.pipe Thread.critical = true STDOUT.flush ProcessController.each_active_object do |pc| for jobs in pc.active_jobs jobs.flush end end pid = fork { Thread.critical = true Thread.list.each do |th| th.kill unless [Thread.main, Thread.current].include?(th) end STDIN.reopen(pipe_peer_in) STDOUT.reopen(pipe_peer_out) ObjectSpace.each_object(IO) do |io| if ![STDIN, STDOUT, STDERR].include?(io) io.close unless io.closed? end end yield } pipe_peer_in.close pipe_peer_out.close command.notify "job(%name:##{pid}) start", @shell.debug? Thread.critical = false th = Thread.start { Thread.critical = true begin _pid = nil command.notify("job(%id) start to waiting finish.", @shell.debug?) Thread.critical = false _pid = Process.waitpid(pid, nil) rescue Errno::ECHILD command.notify "warn: job(%id) was done already waitipd." _pid = true ensure # when the process ends, wait until the command termintes if _pid else command.notify("notice: Process finishing...", "wait for Job[%id] to finish.", "You can use Shell#transact or Shell#check_point for more safe execution.") redo end Thread.exclusive do @job_monitor.synchronize do terminate_job(command) @job_condition.signal command.notify "job(%id) finish.", @shell.debug? end end end } return pid, pipe_me_in, pipe_me_out end
start a job
# File shell/process-controller.rb, line 118 def start_job(command = nil) @jobs_sync.synchronize(:EX) do if command return if command.active? @waiting_jobs.delete command else command = @waiting_jobs.shift return unless command end @active_jobs.push command command.start # start all jobs that input from the job for job in @waiting_jobs start_job(job) if job.input == command end end end
terminate a job
# File shell/process-controller.rb, line 150 def terminate_job(command) @jobs_sync.synchronize(:EX) do @active_jobs.delete command ProcessController.inactivate(self) if @active_jobs.empty? start_job end end end
wait for all jobs to terminate
# File shell/process-controller.rb, line 181 def wait_all_jobs_execution @job_monitor.synchronize do begin while !jobs.empty? @job_condition.wait(@job_monitor) end ensure redo unless jobs.empty? end end end
# File shell/process-controller.rb, line 137 def waiting_job?(job) @jobs_sync.synchronize(:SH) do @waiting_jobs.include?(job) end end