[Backgroundrb-devel] How to add some process control...

[David Lemstra]

Hi,
First off, thanks Ezra for showing the way with BackgrounDrb. It's a
real godsend.

I wanted to add a bit of job control to the system so that I could
cancel DRB processes and so on. I thought I'd share how I did it in case
anyone should find it useful or have feedback.

I wanted to be able to put checkpoints into the drb job so that I could
let the job do whatever needs to be done (cleanup etc) and then
terminate it gracefully, so that complicates things a little bit. The
code is pretty simple, but I'm just learning DRB, so it took me a while
to get it right.

In MiddleMan::new_worker(), the job is first created, then started. This
is because the instance functions I added to BackgrounDrb::Rails don't
seem to be available when kicked off from BackgrounDRb::Rails.initialize()

def new_worker(opts={})
  @mutex.synchronize {
    job_key = opts[:job_key] || gen_key
    unless self[job_key]
      self[job_key] =
instantiate_worker(opts[:class]).new(job_key,opts[:args])
      self[job_key].start_process
      return job_key
    else
      raise ::BackgrounDRbDuplicateKeyError
    end
  }
end

In MiddleMan::delete_worker(), I've added steps where, if the job is
still alive, activate the kill signal to the job, and wait for it to
clean itself up and activate the :safe_to_kill signal.
Then kill it.

def delete_worker(key)
  @mutex.synchronize {
    if @jobs[key]
      if (@jobs[key].thread.alive?)
        @jobs[key].thread[:kill] = true;
        @jobs[key].thread[:safe_to_kill].wait(@mutex)
        @jobs[key].thread.kill
      end
      @jobs.delete(key)
    end
    @timestamps.delete(key) if @timestamps.has_key?(key)
  }
end

Inside BackgrounDrb:Rails:
you can use terminate() to establish the checkpoints/killpoints inside
your job code, or check_terminate/terminate? to check, run cleanup code,
and then raise the :safe_to_kill signal.

def initialize(job_key, args={})
  @logger = BACKGROUNDRB_LOGGER
  @thread = nil;
  @job_key = job_key
  @args = args
end

def start_process
  begin
    @thread = Thread.new do
      Thread.current[:safe_to_kill] = ConditionVariable.new
      Thread.current[:kill] = false
      do_work(@args)
    end
  rescue Exception => e
    @logger.error e.inspect
  end
end

def check_terminate
  raise "Somehow this worker doesn't have a registered thread" if
thread.nil?
  return @thread[:kill]
end

def terminate?
  terminate if check_terminate
end

def terminate
  if check_terminate
    Thread.critical = true
    @logger.info "Canceling job #{@job_key}"
    @thread[:safe_to_kill].signal
    @thread.stop
  end
  return
end

Hope someone finds it useful,
David Lemstra