Class: QC::Worker
- Inherits:
-
Object
- Object
- QC::Worker
- Defined in:
- lib/queue_classic/worker.rb
Overview
A Worker object can process jobs from one or many queues.
Instance Attribute Summary collapse
-
#queues ⇒ Object
Returns the value of attribute queues.
-
#running ⇒ Object
Returns the value of attribute running.
Instance Method Summary collapse
-
#call(job) ⇒ Object
Each job includes a method column.
-
#fork_and_work ⇒ Object
Calls Worker#work but after the current process is forked.
-
#handle_failure(job, e) ⇒ Object
This method will be called when an exception is raised during the execution of the job.
-
#initialize(args = {}) ⇒ Worker
constructor
Creates a new worker but does not start the worker.
-
#lock_job ⇒ Object
Attempt to lock a job in the queue’s table.
- #log(data) ⇒ Object
-
#process(queue, job) ⇒ Object
A job is processed by evaluating the target code.
-
#setup_child ⇒ Object
This method should be overriden if your worker is forking and you need to re-establish database connections.
-
#start ⇒ Object
Commences the working of jobs.
-
#stop ⇒ Object
Signals the worker to stop taking new work.
-
#work ⇒ Object
Blocks on locking a job, and once a job is locked, it will process the job.
Constructor Details
#initialize(args = {}) ⇒ Worker
Creates a new worker but does not start the worker. See Worker#start. This method takes a single hash argument. The following keys are read:
- fork_worker
-
Worker forks each job execution.
- wait_interval
-
Time to wait between failed lock attempts
- connection
-
PGConn object.
- q_name
-
Name of a single queue to process.
- q_names
-
Names of queues to process. Will process left to right.
- top_bound
-
Offset to the head of the queue. 1 == strict FIFO.
18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 |
# File 'lib/queue_classic/worker.rb', line 18 def initialize(args={}) @fork_worker = args[:fork_worker] || QC::FORK_WORKER @wait_interval = args[:wait_interval] || QC::WAIT_TIME if args[:connection] @conn_adapter = ConnAdapter.new(args[:connection]) else @conn_adapter = QC.default_conn_adapter end @queues = setup_queues(@conn_adapter, (args[:q_name] || QC::QUEUE), (args[:q_names] || QC::QUEUES), (args[:top_bound] || QC::TOP_BOUND)) log(args.merge(:at => "worker_initialized")) @running = true end |
Instance Attribute Details
#queues ⇒ Object
Returns the value of attribute queues.
8 9 10 |
# File 'lib/queue_classic/worker.rb', line 8 def queues @queues end |
#running ⇒ Object
Returns the value of attribute running.
8 9 10 |
# File 'lib/queue_classic/worker.rb', line 8 def running @running end |
Instance Method Details
#call(job) ⇒ Object
Each job includes a method column. We will use ruby’s eval to grab the ruby object from memory. We send the method to the object and pass the args.
129 130 131 132 133 134 |
# File 'lib/queue_classic/worker.rb', line 129 def call(job) args = job[:args] receiver_str, _, = job[:method].rpartition('.') receiver = eval(receiver_str) receiver.send(, *args) end |
#fork_and_work ⇒ Object
Calls Worker#work but after the current process is forked. The parent process will wait on the child process to exit.
62 63 64 65 66 |
# File 'lib/queue_classic/worker.rb', line 62 def fork_and_work cpid = fork {setup_child; work} log(:at => :fork, :pid => cpid) Process.wait(cpid) end |
#handle_failure(job, e) ⇒ Object
This method will be called when an exception is raised during the execution of the job.
138 139 140 |
# File 'lib/queue_classic/worker.rb', line 138 def handle_failure(job,e) $stderr.puts("count#qc.job-error=1 job=#{job} error=#{e.inspect}") end |
#lock_job ⇒ Object
Attempt to lock a job in the queue’s table. If a job can be locked, this method returns an array with 2 elements. The first element is the queue from which the job was locked and the second is a hash representation of the job. If a job is returned, its locked_at column has been set in the job’s row. It is the caller’s responsibility to delete the job row from the table when the job is complete.
86 87 88 89 90 91 92 93 94 95 96 97 |
# File 'lib/queue_classic/worker.rb', line 86 def lock_job log(:at => "lock_job") job = nil while @running @queues.each do |queue| if job = queue.lock return [queue, job] end end @conn_adapter.wait(@wait_interval, *@queues.map {|q| q.name}) end end |
#log(data) ⇒ Object
149 150 151 |
# File 'lib/queue_classic/worker.rb', line 149 def log(data) QC.log(data) end |
#process(queue, job) ⇒ Object
A job is processed by evaluating the target code. if the job is evaluated with no exceptions then it is deleted from the queue. If the job has raised an exception the responsibility of what to do with the job is delegated to Worker#handle_failure. If the job is not finished and an INT signal is traped, this method will unlock the job in the queue.
106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 |
# File 'lib/queue_classic/worker.rb', line 106 def process(queue, job) start = Time.now finished = false begin call(job).tap do queue.delete(job[:id]) finished = true end rescue => e handle_failure(job, e) finished = true ensure if !finished queue.unlock(job[:id]) end ttp = Integer((Time.now - start) * 1000) QC.measure("time-to-process=#{ttp} source=#{queue.name}") end end |
#setup_child ⇒ Object
This method should be overriden if your worker is forking and you need to re-establish database connections
145 146 147 |
# File 'lib/queue_classic/worker.rb', line 145 def setup_child log(:at => "setup_child") end |
#start ⇒ Object
Commences the working of jobs. start() spins on @running –which is initialized as true. This method is the primary entry point to starting the worker. The canonical example of starting a worker is as follows: QC::Worker.new.start
41 42 43 44 45 46 47 |
# File 'lib/queue_classic/worker.rb', line 41 def start QC.unlock_jobs_of_dead_workers while @running @fork_worker ? fork_and_work : work end end |
#stop ⇒ Object
Signals the worker to stop taking new work. This method has no immediate effect. However, there are two loops in the worker (one in #start and another in #lock_job) which check the @running variable to determine if further progress is desirable. In the case that @running is false, the aforementioned methods will short circuit and cause the blocking call to #start to unblock.
56 57 58 |
# File 'lib/queue_classic/worker.rb', line 56 def stop @running = false end |