Sha256: 80cf842d06c5a2f6c05f22db4e56534eb332724a656514da8b2f2c9b46b6472a
Contents?: true
Size: 1.53 KB
Versions: 30
Compression:
Stored size: 1.53 KB
Contents
require_relative 'executor'
module Concurrent
# @!macro single_thread_executor
class RubySingleThreadExecutor
include RubyExecutor
include SerialExecutor
# Create a new thread pool.
#
# @see http://docs.oracle.com/javase/tutorial/essential/concurrency/pools.html
# @see http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Executors.html
# @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html
def initialize(opts = {})
@queue = Queue.new
@thread = nil
init_executor
end
protected
# @!visibility private
def execute(*args, &task)
supervise
@queue << [args, task]
end
# @!visibility private
def shutdown_execution
@queue << :stop
stopped_event.set unless alive?
end
# @!visibility private
def kill_execution
@queue.clear
@thread.kill if alive?
end
# @!visibility private
def alive?
@thread && @thread.alive?
end
# @!visibility private
def supervise
@thread = new_worker_thread unless alive?
end
# @!visibility private
def new_worker_thread
Thread.new do
Thread.current.abort_on_exception = false
work
end
end
# @!visibility private
def work
loop do
task = @queue.pop
break if task == :stop
begin
task.last.call(*task.first)
rescue => ex
# let it fail
log DEBUG, ex
end
end
stopped_event.set
end
end
end
Version data entries
30 entries across 30 versions & 1 rubygems