Wait while the task is alive

To confirm task is alive, wait task_heartbeat in set_task.

Author: nurse

lib/perfectqueue/task_metadata.rb

@@ -57,6 +57,10 @@ def timeout
       end
     end
 
+    def last_heartbeat
+      @attributes[:timeout] || 0
+    end
+
     def finished?
       status == TaskStatus::FINISHED
     end

lib/perfectqueue/task_monitor.rb

@@ -28,7 +28,6 @@ def initialize(config, child_heartbeat=nil, force_stop=nil)
       @child_heartbeat_interval = (@config[:child_heartbeat_interval] || 2).to_i
       @task_heartbeat_interval = (@config[:task_heartbeat_interval] || 2).to_i
       @last_child_heartbeat = Time.now.to_i
-      @last_task_heartbeat = Time.now.to_i
 
       @task = nil
 
@@ -59,9 +58,11 @@ def set_task(task, runner)
       task.runner = runner
       @mutex.synchronize {
         @task = task
-        @last_task_heartbeat = @task.timeout.to_i
-        Thread.pass until @thread.stop? if @thread
       }
+      now = Time.now.to_i
+      while @task && @task.last_heartbeat + @task_heartbeat_interval < now
+        sleep 1
+      end
     end
 
     def stop_task(immediate)
@@ -102,7 +103,6 @@ def external_task_heartbeat(task, &block)
       @mutex.synchronize {
         if task == @task
           ret = block.call if block
-          @last_task_heartbeat = Time.now.to_i
         end
         ret
       }
@@ -116,20 +116,18 @@ def run
           next_child_heartbeat = @last_child_heartbeat + @child_heartbeat_interval
 
           if @task
-            next_task_heartbeat = @last_task_heartbeat + @task_heartbeat_interval
+            next_task_heartbeat = @task.last_heartbeat + @task_heartbeat_interval
             next_time = [next_child_heartbeat, next_task_heartbeat].min
           else
-            next_task_heartbeat = nil
             next_time = next_child_heartbeat
           end
 
           next_wait = next_time - now
           @cond.wait(next_wait) if next_wait > 0
 
           now = Time.now.to_i
-          if @task && next_task_heartbeat && next_task_heartbeat <= now
+          if @task && @task.last_heartbeat + @task_heartbeat_interval <= now
             task_heartbeat
-            @last_task_heartbeat = now
           end
 
           if next_child_heartbeat <= now
@@ -146,12 +144,12 @@ def run
 
     private
     def task_heartbeat
-      v = @task.heartbeat!(last_heartbeat: @last_task_heartbeat)
-      @task.attributes[:timeout] = v
-      v
+      task = @task
+      task.attributes[:timeout] = task.heartbeat!(last_heartbeat: task.last_heartbeat)
     rescue
       # finished, cancel_requested, preempted, etc.
       kill_task($!)
+      @task = nil
     end
   end
 

spec/multiprocess/child_process_spec.rb

@@ -73,7 +73,7 @@
   end
 
   describe '#process' do
-    let (:task){ double('task', key: double, timeout: Time.now.to_i) }
+    let (:task){ double('task', key: double, last_heartbeat: Time.now.to_i) }
     before do
       expect(runner_insntace).to receive(:run)
     end

spec/task_monitor_spec.rb

@@ -23,7 +23,6 @@
       ret = double('ret')
       tm.instance_variable_set(:@task, task)
       expect(tm.external_task_heartbeat(task){ret}).to eq(ret)
-      expect(tm.instance_variable_get(:@last_task_heartbeat)).to eq(epoch)
     end
   end
 
@@ -44,7 +43,8 @@
     let (:tm){ PerfectQueue::TaskMonitor.new(logger: double('logger').as_null_object, task_heartbeat_interval: 1) }
     let (:err){ StandardError.new('heartbeat preempted') }
     let (:now){ Time.now.to_i }
-    let (:task){ double('task', timeout: now) }
+    let (:task){ double('task', attributes: {}, last_heartbeat: now) }
+    let (:runner){ double('runner') }
     before do
       tm.set_task(task, double('runner'))
     end
@@ -66,13 +66,10 @@
       it 'update timeout' do
         tasks = client.acquire(now: now-80)
         task = tasks[0]
-        expect(task.timeout.to_i).to eq(now-80+config[:alive_time])
-        tm.set_task(task, double('runner'))
+        expect(task.last_heartbeat).to eq(now-80+config[:alive_time])
         allow(Time).to receive(:now).and_return(now-50)
-        Timeout.timeout(5) do
-          sleep 0.5 until tm.instance_variable_get(:@last_task_heartbeat) == now-50
-        end
-        expect(task.timeout.to_i).to eq(now-50+config[:alive_time])
+        tm.set_task(task, runner)
+        expect(task.last_heartbeat).to eq(now-50+config[:alive_time])
       end
     end
     context 'stolen' do
@@ -94,20 +91,38 @@
         task2 = tasks[0]
         expect(task2.timeout.to_i).to eq(now-60+config[:alive_time])
 
-        tm.set_task(task1, double('runner'))
         allow(Time).to receive(:now).and_return(now-50)
+        expect(runner).to receive(:kill)
+        tm.set_task(task1, runner)
+      end
+    end
+    context 'timeout but can acquire' do
+      before do
+        client.backend.db.tap{|s| s.tables.each{|t| s.drop_table(t) } }
+        client.init_database
+        client.submit('key', 'test1', {'foo' => 1}, {now: now-90,compression: 'gzip'})
+        tm.start
+      end
+      after do
+        tm.stop
+      end
+      it 'raise error' do
+        tasks = client.acquire(now: now-80)
+        task1 = tasks[0]
+        expect(task1.timeout.to_i).to eq(now-80+config[:alive_time])
+
+        allow(Time).to receive(:now).and_return(now-50)
+        tm.set_task(task1, runner)
 
-        flag = false
-        expect(task1.runner).to receive(:kill){flag = true}
-        Timeout.timeout(5){ sleep 0.5 until flag }
+        expect(task1.runner).to eq(runner)
       end
     end
   end
 end
 
 describe PerfectQueue::TaskMonitorHook do
   let (:task) do
-    obj = AcquiredTask.new(double(:client).as_null_object, 'key', {}, double)
+    obj = AcquiredTask.new(double(:client).as_null_object, 'key', {timeout: Time.now.to_i}, double)
     tm = TaskMonitor.new(logger: double('logger').as_null_object)
     tm.set_task(obj, double('runner'))
     obj