@@ -7,23 +7,20 @@ Background tasks made dead simple. Bonus features ============== -There are bonus features available if you enable multithreading in your environment. +There are bonus features available if you call +BackgroundFu.enable_bonus_features in lib/daemons/background.rb. +This is enabled by default. -These are: +These features are: * monitoring progress (perfect for ajax progress bars) * stopping a running worker in a merciful way. - * restarting a failed or finished job. + * restarting a failed or stopped job. Read the code (short and easy) to discover them. - Quick Start =========== -This plugin previously depended on Kyle Maxwell's daemon_generator plugin from -http://svn.kylemaxwell.com/rails_plugins/daemon_generator/trunk but since it is not -available now, I copied all the needed files into BackgroundFu. - ./script/plugin install https://svn.trix.pl/public/background_fu/ ruby ./script/generate background @@ -44,7 +41,7 @@ Examples In lib/workers/example_worker.rb: -# Simple, non-monitored worker, for those who don't believe in multi-threading. +# Simple, non-monitored worker. class ExampleWorker def add(a, b) @@ -55,10 +52,7 @@ end In lib/workers/example_monitored_worker.rb: -# Bonus features available if allow_concurrency is set to true! # Remeber to include BackgroundFu::WorkerMonitoring. -# Every place where record_progress is invoked is a possible stopping place. - class ExampleMonitoredWorker include BackgroundFu::WorkerMonitoring README @@ -2,23 +2,16 @@ require File.dirname(__FILE__) + "/../../config/environment" -$running = true; -Signal.trap("TERM") do - $running = false - - # We are exiting explicitly. - # This is convenient if you set much longer sleeps below. - # If sleep is long, you would have to wait long for next iteration - # of "while" (which would result in daemon death). - exit -end +# Progress monitoring, stopping/restarting jobs. +# Comment this out if you don't need these features. +BackgroundFu.enable_bonus_features -while($running) do +Signal.trap("TERM") { exit } - if job = Job.pending.first +loop do + if job = Job.find(:first, :conditions => {:state => "pending"}, :order => "id asc") job.get_done! else sleep 5 end - end generators/background/templates/background.rb @@ -1,8 +1,6 @@ -# Remember to enable allow_concurrency in your environment -# and include Background::WorkerMonitoring. -# Every place where record_progress is invoked is a possible stopping place. class ExampleMonitoredWorker + # After including woker monitoring you can invoke record_progress() method. include BackgroundFu::WorkerMonitoring def long_and_monitored generators/background/templates/example_monitored_worker.rb @@ -1,4 +1,4 @@ -# Simple, non-monitored worker, for those who don't believe in multi-threading. +# Simple, non-monitored worker. class ExampleWorker def add(a, b) generators/background/templates/example_worker.rb @@ -10,6 +10,7 @@ class CreateJobs < ActiveRecord::Migration t.integer :progress t.string :state + t.timestamp :started_at t.timestamps end end generators/background/templates/migration.rb @@ -1,5 +1 @@ Dependencies.load_paths << "#{RAILS_ROOT}/lib/workers" - -if ActiveRecord::Base.allow_concurrency - Job.send!(:include, Job::BonusFeatures) -end init.rb @@ -1,5 +1,9 @@ module BackgroundFu - # Whitespace good + # Invoked optionally from lib/daemons/background.rb + def self.enable_bonus_features + ActiveRecord::Base.allow_concurrency = true + Job.send!(:include, Job::BonusFeatures) + end end lib/background_fu.rb @@ -1,10 +1,19 @@ -# Include it in your workers to enable progress monitoring, stopping/restarting jobs. -# You also have to allow_concurrency in ActiveRecord to make this work. +# Include it in your workers to enable progress monitoring and stopping jobs. module BackgroundFu::WorkerMonitoring - def record_progress(progress) - @progress = progress.to_i - throw(:stopped) if @should_stop + # In most cases you will have some loop which will execute known (m) times. + # Every time the loop iterates you increment a counter (n). + # The formula to get progress in percents is: 100 * n / m. + # If you invoke this method with second argument, then this is calculated for you. + # You also can omit second argument and progress will be passed directly to db. + def record_progress(progress_or_iteration, iterations_count = nil) + if iterations_count.to_i > 0 + @progress = ((progress_or_iteration.to_f / iterations_count) * 100).to_i + else + @progress = progress_or_iteration.to_i + end + + throw(:stopping, true) if @stopping end end lib/background_fu/worker_monitoring.rb @@ -3,16 +3,17 @@ # job = Job.enqueue!(MyWorker, :my_method, "my_arg_1", "my_arg_2") class Job < ActiveRecord::Base - States = %w(pending running stopping finished failed) + cattr_accessor :states + self.states = %w(pending running finished failed) serialize :args, Array serialize :result - before_validation_on_create :setup_state - - validates_inclusion_of :state, :in => States + before_create :setup_state validates_presence_of :worker_class, :worker_method - + + attr_readonly :worker_class, :worker_method, :args + def self.enqueue!(worker_class, worker_method, *args) create!( :worker_class => worker_class.to_s, @@ -21,43 +22,54 @@ class Job < ActiveRecord::Base ) end - States.each do |state_name| - # Every call to these methods fetches fresh state value from db. - define_method("#{state_name}?") do - reload.state == state_name - end - - # BackgroundJob.running => array of running jobs, etc. - self.class.send!(:define_method, state_name) do - find_all_by_state(state_name, :order => "id desc") - end - end - # Invoked by a background daemon. def get_done! - update_attribute(:state, "running") - @worker = worker_class.constantize.new - before_work - catch(:stopped) do - self.result = @worker.send!(worker_method, *args) - end - self.state = "finished" + initialize_worker + invoke_worker rescue Exception => e - self.result = [e.message, e.backtrace.join("\n")].join("\n") - self.state = "failed" + rescue_worker(e) ensure - save(false) - after_work + ensure_worker + end + + def initialize_worker + update_attributes!(:started_at => Time.now, :state => "running") + @worker = worker_class.constantize.new + end + + def invoke_worker + self.result = @worker.send!(worker_method, *args) + self.state = "finished" + end + + def rescue_worker(exception) + self.result = [exception.message, exception.backtrace.join("\n")].join("\n\n") + self.state = "failed" + end + + def ensure_worker + self.progress = @worker.instance_variable_get("@progress") + save! + end + + def self.generate_state_helpers + states.each do |state_name| + define_method("#{state_name}?") do + state == state_name + end + + # BackgroundJob.running => array of running jobs, etc. + self.class.send!(:define_method, state_name) do + find_all_by_state(state_name, :order => "id desc") + end + end end + generate_state_helpers def setup_state return unless state.blank? self.state = "pending" end - - # callbacks - def before_work; end - def after_work; end end lib/job.rb @@ -1,18 +1,28 @@ module Job::BonusFeatures - # Registering callbacks def self.included(base) - base.send!(:define_method, :before_work) do - monitor_worker - end + base.states += %w(stopping stopped) + base.generate_state_helpers - base.send!(:define_method, :after_work) do - cleanup_after_threads + base.alias_method_chain :invoke_worker, :threads + base.alias_method_chain :ensure_worker, :threads + end + + def invoke_worker_with_threads + monitor_worker + + res = catch(:stopping) do + invoke_worker_without_threads; nil end + + self.state = res ? "stopped" : "finished" + end + + def ensure_worker_with_threads + ensure_worker_without_threads + cleanup_after_threads end - # When multithreading is enabled, you can ask a worker to terminate a job. - # # The record_progress() method becomes available when your worker class includes # Background::WorkerMonitoring. # @@ -21,36 +31,39 @@ module Job::BonusFeatures # How it works: # 1. invoke job.stop! to set a state (stopping) in a db # 2. Monitoring thread picks up the state change from db - # and sets @should_stop to true in the worker. + # and sets @stopping to true in the worker. # 3. The worker invokes a register_progress() somewhere during execution. - # 4. The record_progress() method throws :stopped symbol if @should_stop == true - # 5. The job catches the :stopped symbol and reacts upon it. + # 4. The record_progress() method throws :stopping symbol if @stopping == true + # 5. The job catches the :stopping symbol and reacts upon it. # 6. The job is stopped in a merciful way. No one gets harmed. def stop! - if running? - update_attribute(:state, "stopping") - end + update_attribute(:state, "stopping") if running? end def restart! - if finished? || failed? - update_attributes(:result => nil, :progress => nil, :state => "pending") + if stopped? || failed? + update_attributes!( + :result => nil, + :progress => nil, + :started_at => nil, + :state => "pending" + ) end end - # Don't get scared. ActiveRecord IS thread-safe. - # To enable progress monitoring you have to manually - # set ActiveRecord::Base.allow_concurrency to true - # in lib/daemons/background.rb. # You don't need to enable multi-threading in your Rails app. # This is the only place where multi-threading occurs - # in the plugin. Progress monitoring is optional. + # in the plugin and is completely optional. + # You can disable these features by commenting out + # BackgroundFu.enable_bonus_features in lib/daemons/background.rb. def monitor_worker Thread.new do - while(running?) + # 1. running? - check if not failed or finished. + # 2. !Job.find(id).stopping? - check if someone ordered stopping the job. + while(running? && !Job.find(id).stopping?) current_progress = @worker.instance_variable_get("@progress") - if current_progress == progress + if current_progress == progress sleep 5 else update_attribute(:progress, current_progress) @@ -58,14 +71,9 @@ module Job::BonusFeatures end end - # If stop! was called on running job, worker has @should_stop set to true. - if(stopping?) - @worker.instance_variable_set("@should_stop", true) - end - - # Ensure last available progress is saved - if(finished?) - update_attribute(:progress, @worker.instance_variable_get("@progress")) + # If someone ordered stopping a job we infrom the worker that it should stop. + if(Job.find(id).stopping?) + @worker.instance_variable_set("@stopping", true) end end end @@ -75,4 +83,13 @@ module Job::BonusFeatures ActiveRecord::Base.verify_active_connections! end + def elapsed + (updated_at - started_at).to_i if !pending? + end + + # seconds to go, based on estimated and progress + def estimated + ((elapsed * 100) / progress) - elapsed if running? && (1..99).include?(progress.to_i) + end + end lib/job/bonus_features.rb 8326f32afc0d224a070c21a43806de05dc7edd94 necro necro@7adf99dd-3210-0410-845e-f78690b30d0d http://github.com/ncr/background-fu/commit/0b29bf867b73a3e0fadac3e8f7800401ccaf22b4 0b29bf867b73a3e0fadac3e8f7800401ccaf22b4 2008-02-19T12:19:05-08:00 2008-02-19T12:19:05-08:00 git-svn-id: https://svn.trix.pl/public/background_fu@2634 7adf99dd-3210-0410-845e-f78690b30d0d e2eacc2083a519243073238bf49aea7c8b2ee5a0 necro necro@7adf99dd-3210-0410-845e-f78690b30d0d