RubyGems - perfectqueue - Versions diffs - 0.7.0 → 0.7.1 - Mend

perfectqueue 0.7.0 → 0.7.1

This diff represents the content of publicly available package versions that have been released to one of the supported registries. The information contained in this diff is provided for informational purposes only and reflects changes between package versions as they appear in their respective public registries.

Files changed (7) hide show

data/ChangeLog +7 -0
data/README.rdoc +23 -23
data/lib/perfectqueue/backend/simpledb.rb +10 -10
data/lib/perfectqueue/command/perfectqueue.rb +44 -30
data/lib/perfectqueue/version.rb +1 -1
data/lib/perfectqueue/worker.rb +31 -16
metadata +3 -3

data/ChangeLog CHANGED Viewed

@@ -1,4 +1,11 @@
+== 2011-08-31 version 0.7.1
+* Fixed kill behavior
+* Supported to read multiple configuration files
+* Optimized --list subcommand
 == 2011-08-31 version 0.7.0
 * Backend uses logical delete instead of deleting tasks when task is finished

data/README.rdoc CHANGED Viewed

@@ -37,14 +37,14 @@ PerfectQueue uses following database schema:
   $ perfectqueue \
         --database mysql://user:password@localhost/mydb \
         --table perfectqueue \
-        --push unique-key-id='{"any":"data"}'
+        --push unique-key-id '{"any":"data"}'
   # SimpleDB
   $ perfectqueue \
         --simpledb your-simpledb-domain-name \
         -k AWS_KEY_ID \
         -s AWS_SECRET_KEY \
-        --push unique-key-id='{"any":"data"}'
+        --push unique-key-id '{"any":"data"}'
 *Using* *PerfectQueue* *library:*
@@ -118,33 +118,33 @@ Use _perfectqueue_ command to execute a command.
   Usage: perfectqueue [options] [-- <ARGV-for-exec-or-run>]
-          --push ID=DATA               Push a task to the queue
+          --push <ID> <DATA>           Push a task to the queue
           --list                       Show queued tasks
-          --cancel ID                  Cancel a queued task
-          --configure PATH.yaml        Write configuration file
+          --cancel <ID>                Cancel a queued task
+          --configure <PATH.yaml>      Write configuration file
-          --exec COMMAND               Execute command
-          --run SCRIPT.rb              Run method named 'run' defined in the script
+          --exec <COMMAND>             Execute command
+          --run <SCRIPT.rb>            Run method named 'run' defined in the script
-      -f, --file PATH.yaml             Read configuration file
+      -f, --file <PATH.yaml>           Read configuration file
       -C, --run-class                  Class name for --run (default: ::Run)
-      -t, --timeout SEC                Time for another worker to take over a task when this worker goes down (default: 600)
-      -b, --heartbeat-interval SEC     Threshold time to extend the timeout (heartbeat interval) (default: timeout * 3/4)
-      -x, --kill-timeout SEC           Threshold time to kill a task process (default: timeout * 10)
-      -X, --kill-interval SEC          Threshold time to retry killing a task process (default: 60)
-      -i, --poll-interval SEC          Polling interval (default: 1)
-      -r, --retry-wait SEC             Time to retry a task when it is failed (default: same as timeout)
-      -e, --expire SEC                 Threshold time to expire a task (default: 345600 (4days))
+      -t, --timeout <SEC>              Time for another worker to take over a task when this worker goes down (default: 600)
+      -b, --heartbeat-interval <SEC>   Threshold time to extend the timeout (heartbeat interval) (default: timeout * 3/4)
+      -x, --kill-timeout <SEC>         Threshold time to kill a task process (default: timeout * 10)
+      -X, --kill-interval <SEC>        Threshold time to retry killing a task process (default: 60)
+      -i, --poll-interval <SEC>        Polling interval (default: 1)
+      -r, --retry-wait <SEC>           Time to retry a task when it is failed (default: same as timeout)
+      -e, --expire <SEC>               Threshold time to expire a task (default: 345601 (4days))
-          --database URI               Use RDBMS for the backend database (e.g.: mysql://user:password@localhost/mydb)
-          --table NAME                 backend: name of the table (default: perfectqueue)
-          --simpledb DOMAIN            Use Amazon SimpleDB for the backend database (e.g.: --simpledb mydomain -k KEY_ID -s SEC_KEY)
-      -k, --key-id ID                  AWS Access Key ID
-      -s, --secret-key KEY             AWS Secret Access Key
+          --database <URI>             Use RDBMS for the backend database (e.g.: mysql://user:password@localhost/mydb)
+          --table <NAME>               backend: name of the table (default: perfectqueue)
+          --simpledb <DOMAIN>          Use Amazon SimpleDB for the backend database (e.g.: --simpledb mydomain -k KEY_ID -s SEC_KEY)
+      -k, --key-id <ID>                AWS Access Key ID
+      -s, --secret-key <KEY>           AWS Secret Access Key
-      -w, --worker NUM                 Number of worker threads (default: 1)
-      -d, --daemon PIDFILE             Daemonize (default: foreground)
-      -o, --log PATH                   log file path
+      -w, --worker <NUM>               Number of worker threads (default: 1)
+      -d, --daemon <PIDFILE>           Daemonize (default: foreground)
+      -o, --log <PATH>                 log file path
       -v, --verbose                    verbose mode

data/lib/perfectqueue/backend/simpledb.rb CHANGED Viewed

@@ -27,16 +27,16 @@ class SimpleDBBackend < Backend
   end
   def list(&block)
-    @domain.items.each {|item|
-      id = item.name
-      attrs = item.data.attributes
-      salt = attrs['created_at'].first
-      if salt && !salt.empty?
-        created_at = int_decode(salt)
-        data = attrs['data'].first
-        timeout = int_decode(attrs['timeout'].first)
-        yield id, created_at, data, timeout
-      end
+    @domain.items.select('timeout', 'data', 'created_at',
+                        :where => "created_at != '' AND timeout > '#{int_encode(0)}'",
+                        :order => [:timeout, :asc],
+                        :consistent_read => @consistent_read) {|itemdata|
+      id = itemdata.name
+      attrs = itemdata.attributes
+      created_at = int_decode(attrs['created_at'].first)
+      data = attrs['data'].first
+      timeout = int_decode(attrs['timeout'].first)
+      yield id, created_at, data, timeout
     }
   end

data/lib/perfectqueue/command/perfectqueue.rb CHANGED Viewed

@@ -31,108 +31,108 @@ op.on('-v', '--verbose', "verbose mode", TrueClass) {|b|
 }
 op.separator("")
-op.on('--push ID=DATA', 'Push a task to the queue') {|s|
+op.on('--push <ID> <DATA>', 'Push a task to the queue') {|s|
   type = :push
-  id, data = s.split('=',2)
+  id = s
 }
 op.on('--list', 'Show queued tasks', TrueClass) {|b|
   type = :list
 }
-op.on('--cancel ID', 'Cancel a queued task') {|s|
+op.on('--cancel <ID>', 'Cancel a queued task') {|s|
   type = :cancel
   id = s
 }
-op.on('--configure PATH.yaml', 'Write configuration file') {|s|
+op.on('--configure <PATH.yaml>', 'Write configuration file') {|s|
   type = :conf
   confout = s
 }
 op.separator("")
-op.on('--exec COMMAND', 'Execute command') {|s|
+op.on('--exec <COMMAND>', 'Execute command') {|s|
   type = :exec
   conf[:exec] = s
 }
-op.on('--run SCRIPT.rb', 'Run method named \'run\' defined in the script') {|s|
+op.on('--run <SCRIPT.rb>', 'Run method named \'run\' defined in the script') {|s|
   type = :run
   conf[:run] = s
 }
 op.separator("")
-op.on('-f', '--file PATH.yaml', 'Read configuration file') {|s|
-  conf[:file] = s
+op.on('-f', '--file <PATH.yaml>', 'Read configuration file') {|s|
+  (conf[:files] ||= []) << s
 }
 op.on('-C', '--run-class', 'Class name for --run (default: ::Run)') {|s|
   conf[:run_class] = s
 }
-op.on('-t', '--timeout SEC', 'Time for another worker to take over a task when this worker goes down (default: 600)', Integer) {|i|
+op.on('-t', '--timeout <SEC>', 'Time for another worker to take over a task when this worker goes down (default: 600)', Integer) {|i|
   conf[:timeout] = i
 }
-op.on('-b', '--heartbeat-interval SEC', 'Threshold time to extend the timeout (heartbeat interval) (default: timeout * 3/4)', Integer) {|i|
+op.on('-b', '--heartbeat-interval <SEC>', 'Threshold time to extend the timeout (heartbeat interval) (default: timeout * 3/4)', Integer) {|i|
   conf[:heartbeat_interval] = i
 }
-op.on('-x', '--kill-timeout SEC', 'Threshold time to kill a task process (default: timeout * 10)', Integer) {|i|
+op.on('-x', '--kill-timeout <SEC>', 'Threshold time to kill a task process (default: timeout * 10)', Integer) {|i|
   conf[:kill_timeout] = i
 }
-op.on('-X', '--kill-interval SEC', 'Threshold time to retry killing a task process (default: 60)', Integer) {|i|
+op.on('-X', '--kill-interval <SEC>', 'Threshold time to retry killing a task process (default: 60)', Integer) {|i|
   conf[:kill_interval] = i
 }
-op.on('-i', '--poll-interval SEC', 'Polling interval (default: 1)', Integer) {|i|
+op.on('-i', '--poll-interval <SEC>', 'Polling interval (default: 1)', Integer) {|i|
   conf[:poll_interval] = i
 }
-op.on('-r', '--retry-wait SEC', 'Time to retry a task when it is failed (default: same as timeout)', Integer) {|i|
+op.on('-r', '--retry-wait <SEC>', 'Time to retry a task when it is failed (default: same as timeout)', Integer) {|i|
   conf[:retry_wait] = i
 }
-op.on('-e', '--expire SEC', 'Threshold time to expire a task (default: 345600 (4days))', Integer) {|i|
+op.on('-e', '--expire <SEC>', 'Threshold time to expire a task (default: 345601 (4days))', Integer) {|i|
   conf[:expire] = i
 }
 op.separator("")
-op.on('--database URI', 'Use RDBMS for the backend database (e.g.: mysql://user:password@localhost/mydb)') {|s|
+op.on('--database <URI>', 'Use RDBMS for the backend database (e.g.: mysql://user:password@localhost/mydb)') {|s|
   conf[:backend_database] = s
 }
-op.on('--table NAME', 'backend: name of the table (default: perfectqueue)') {|s|
+op.on('--table <NAME>', 'backend: name of the table (default: perfectqueue)') {|s|
   conf[:backend_table] = s
 }
-op.on('--simpledb DOMAIN', 'Use Amazon SimpleDB for the backend database (e.g.: --simpledb mydomain -k KEY_ID -s SEC_KEY)') {|s|
+op.on('--simpledb <DOMAIN>', 'Use Amazon SimpleDB for the backend database (e.g.: --simpledb mydomain -k KEY_ID -s SEC_KEY)') {|s|
   conf[:backend_simpledb] = s
 }
-op.on('-k', '--key-id ID', 'AWS Access Key ID') {|s|
+op.on('-k', '--key-id <ID>', 'AWS Access Key ID') {|s|
   conf[:backend_key_id] = s
 }
-op.on('-s', '--secret-key KEY', 'AWS Secret Access Key') {|s|
+op.on('-s', '--secret-key <KEY>', 'AWS Secret Access Key') {|s|
   conf[:backend_secret_key] = s
 }
 op.separator("")
-op.on('-w', '--worker NUM', 'Number of worker threads (default: 1)', Integer) {|i|
+op.on('-w', '--worker <NUM>', 'Number of worker threads (default: 1)', Integer) {|i|
   conf[:workers] = i
 }
-op.on('-d', '--daemon PIDFILE', 'Daemonize (default: foreground)') {|s|
+op.on('-d', '--daemon <PIDFILE>', 'Daemonize (default: foreground)') {|s|
   conf[:daemon] = s
 }
-op.on('-o', '--log PATH', "log file path") {|s|
+op.on('-o', '--log <PATH>', "log file path") {|s|
   conf[:log] = s
 }
@@ -159,15 +159,29 @@ begin
   end
   op.parse!(argv)
-  if argv.length != 0
-    usage nil
+  case type
+  when :push
+    if argv.length != 1
+      usage nil
+    end
+    data = argv[0]
+  else
+    if argv.length != 0
+      usage nil
+    end
   end
-  if conf[:file]
+  if conf[:files]
     require 'yaml'
-    yaml = YAML.load File.read(conf[:file])
+    docs = ''
+    conf[:files].each {|file|
+      docs << File.read(file)
+    }
     y = {}
-    yaml.each_pair {|k,v| y[k.to_sym] = v }
+    YAML.load_documents(docs) {|yaml|
+      yaml.each_pair {|k,v| y[k.to_sym] = v }
+    }
     conf = defaults.merge(y).merge(conf)
@@ -265,7 +279,7 @@ when :cancel
   if canceled
     puts "Task id=#{id} is canceled."
   else
-    puts "Task id=#{id} does not exist."
+    puts "Task id=#{id} does not exist. abort"
   end
 when :push
@@ -273,7 +287,7 @@ when :push
   if submitted
     puts "Task id=#{id} is submitted."
   else
-    puts "Task id=#{id} already exists."
+    puts "Task id=#{id} is duplicated. abort."
   end
 when :exec, :run

data/lib/perfectqueue/version.rb CHANGED Viewed

@@ -1,5 +1,5 @@
 module PerfectQueue
-VERSION = '0.7.0'
+VERSION = '0.7.1'
 end

data/lib/perfectqueue/worker.rb CHANGED Viewed

@@ -37,28 +37,32 @@ class MonitorThread
           @cond.wait(@mutex)
         end
       }
-      while true
-        sleep 1
-        @mutex.synchronize {
-          return if @engine.finished?
-          break unless @token
-          now = Time.now.to_i
-          try_extend(now)
-          try_kill(now)
-        }
-      end
+      process
     end
   rescue
     @engine.stop($!)
   end
+  def process
+    while true
+      sleep 1
+      @mutex.synchronize {
+        return if @engine.finished?
+        return unless @token
+        now = Time.now.to_i
+        try_extend(now)
+        try_kill(now)
+      }
+    end
+  end
   def try_extend(now)
     if now >= @heartbeat_time && !@canceled
-      @log.debug "extending timeout=#{now+@timeout} id=#{@task.id}"
+      @log.debug "extending timeout=#{now+@timeout} id=#{@task_id}"
       begin
         @backend.update(@token, now+@timeout)
       rescue CanceledError
-        @log.info "task id=#{@task.id} is canceled."
+        @log.info "task id=#{@task_id} is canceled."
         @canceled = true
         @kill_time = now
       end
@@ -75,8 +79,15 @@ class MonitorThread
   def kill!
     if @kill_proc
-      @log.info "killing #{@task.id}..."
-      @kill_proc.call rescue nil
+      @log.info "killing id=#{@task_id}..."
+      begin
+        @kill_proc.call
+      rescue
+        @log.info "kill failed id=#{@task_id}: #{$!}"
+        $!.backtrace.each {|bt|
+          $log.debug "  #{bt}"
+        }
+      end
     end
   end
@@ -90,10 +101,11 @@ class MonitorThread
     @thread.join
   end
-  def set(token)
+  def set(token, task_id)
     @mutex.synchronize {
       now = Time.now.to_i
       @token = token
+      @task_id = task_id
       @heartbeat_time = now + @heartbeat_interval
       @kill_time = now + @kill_timeout
       @kill_proc = nil
@@ -165,7 +177,7 @@ class Worker
   def process(token, task)
     @log.info "processing task id=#{task.id}"
-    @monitor.set(token)
+    @monitor.set(token, task.id)
     success = false
     begin
       run = @run_class.new(task)
@@ -181,6 +193,9 @@ class Worker
     rescue
       @log.info "failed id=#{task.id}: #{$!}"
+      $!.backtrace.each {|bt|
+        $log.debug "  #{bt}"
+      }
     ensure
       @monitor.reset(success)

metadata CHANGED Viewed

@@ -1,13 +1,13 @@
 --- !ruby/object:Gem::Specification
 name: perfectqueue
 version: !ruby/object:Gem::Version
-  hash: 3
+  hash: 1
   prerelease: false
   segments:
   - 0
   - 7
-  - 0
-  version: 0.7.0
+  - 1
+  version: 0.7.1
 platform: ruby
 authors:
 - Sadayuki Furuhashi