worker.rb in narou-2.7.0

- old
+ new

@@ -15,10 +15,12 @@
     @queue = Queue.new
     @size = 0
     @mutex = Mutex.new
     @worker_thread = nil
     @push_server = Narou::PushServer.instance
+    @cancel_signal = false
+    @thread_of_block_executing = nil
   end
 
   def running?
     !@worker_thread.!
   end
@@ -27,11 +29,23 @@
     return if running?
     @worker_thread = Thread.new do
       loop do
         begin
           q = @queue.pop
-          q[:block].call
+          if canceled?
+            @queue.clear
+            @cancel_signal = false
+          else
+            #q[:block].call
+            @thread_of_block_executing = Thread.new do
+              q[:block].call
+            end
+            @thread_of_block_executing.tap do |th|
+              th.join
+              th = nil
+            end
+          end
         rescue SystemExit
         rescue Exception => e
           # Workerスレッド内での例外は表示するだけしてスレッドは生かしたままにする
           STDOUT.puts $@.shift + ": #{e.message} (#{e.class})"
           $@.each do |b|
@@ -45,10 +59,33 @@
         end
       end
     end
   end
 
+  def self.cancel
+    instance.cancel
+  end
+
+  def cancel
+    @mutex.synchronize do
+      if @size > 0
+        @cancel_signal = true
+        @size = 0
+        @thread_of_block_executing.raise(Interrupt) if @thread_of_block_executing
+        Thread.pass
+      end
+    end
+  end
+
+  def self.canceled?
+    instance.canceled?
+  end
+
+  def canceled?
+    @cancel_signal
+  end
+
   def stop
     @worker_thread.kill if @worker_thread
     @worker_thread = nil
   end
 
@@ -80,9 +117,10 @@
   end
 
   def countdown
     @mutex.synchronize do
       @size -= 1
+      @size = 0 if @size < 0
     end
   end
 end
 