Merge pull request #42 from jdantonio/global-thread-pool

Global thread pool is now a thread pool executor.

Author: jdantonio

demos/global_thread_pool-demo.rb

@@ -0,0 +1,45 @@
+require 'concurrent'
+require 'open-uri'
+
+def get_year_end_closing(symbol, year)
+  uri = "http://ichart.finance.yahoo.com/table.csv?s=#{symbol}&a=11&b=01&c=#{year}&d=11&e=31&f=#{year}&g=m"
+  data = open(uri) {|f| f.collect{|line| line.strip } }
+  price = data[1].split(',')[4]
+  price.to_f
+  [symbol, price.to_f]
+end
+
+def get_top_stock(symbols, year, timeout = 5)
+  stock_prices = symbols.collect{|symbol| Concurrent::dataflow{ get_year_end_closing(symbol, year) }}
+  Concurrent::dataflow(*stock_prices) { |*prices|
+    prices.reduce(['', 0.0]){|highest, price| price.last > highest.last ? price : highest}
+  }.value(timeout)
+end
+
+def timer(*args)
+  return 0,nil unless block_given?
+  t1 = Time.now
+  result = yield(*args)
+  t2 = Time.now
+  return (t2 - t1), result
+end
+
+def strftimer(seconds) # :nodoc:
+  Time.at(seconds).gmtime.strftime('%R:%S.%L')
+end
+
+s_and_p_500 = ['MMM','ABT','ABBV','ACE','ACN','ACT','ADBE','ADT','AES','AET','AFL','A','GAS','APD','ARG','AKAM','AA','ALXN','ATI','ALLE','AGN','ADS','ALL','ALTR','MO','AMZN','AEE','AEP','AXP','AIG','AMT','AMP','ABC','AME','AMGN','APH','APC','ADI','AON','APA','AIV','AAPL','AMAT','ADM','AIZ','T','ADSK','ADP','AN','AZO','AVB','AVY','AVP','BHI','BLL','BAC','BK','BCR','BAX','BBT','BEAM','BDX','BBBY','BMS','BRK.B','BBY','BIIB','BLK','HRB','BA','BWA','BXP','BSX','BMY','BRCM','BF.B','CHRW','CA','CVC','COG','CAM','CPB','COF','CAH','CFN','KMX','CCL','CAT','CBG','CBS','CELG','CNP','CTL','CERN','CF','SCHW','CHK','CVX','CMG','CB','CI','CINF','CTAS','CSCO','C','CTXS','CLF','CLX','CME','CMS','COH','KO','CCE','CTSH','CL','CMCSA','CMA','CSC','CAG','COP','CNX','ED','STZ','GLW','COST','COV','CCI','CSX','CMI','CVS','DHI','DHR','DRI','DVA','DE','DLPH','DAL','DNR','XRAY','DVN','DO','DTV','DFS','DISCA','DG','DLTR','D','DOV','DOW','DPS','DTE','DD','DUK','DNB','ETFC','EMN','ETN','EBAY','ECL','EIX','EW','EA','EMC','EMR','ESV','ETR','EOG','EQT','EFX','EQR','EL','EXC','EXPE','EXPD','ESRX','XOM','FFIV','FB','FDO','FAST','FDX','FIS','FITB','FSLR','FE','FISV','FLIR','FLS','FLR','FMC','FTI','F','FRX','FOSL','BEN','FCX','FTR','GME','GCI','GPS','GRMN','GD','GE','GGP','GIS','GM','GPC','GNW','GILD','GS','GT','GOOG','GWW','HAL','HOG','HAR','HRS','HIG','HAS','HCP','HCN','HP','HES','HPQ','HD','HON','HRL','HSP','HST','HCBK','HUM','HBAN','ITW','IR','TEG','INTC','ICE','IBM','IGT','IP','IPG','IFF','INTU','ISRG','IVZ','IRM','JBL','JEC','JNJ','JCI','JOY','JPM','JNPR','KSU','K','KEY','GMCR','KMB','KIM','KMI','KLAC','KSS','KRFT','KR','LB','LLL','LH','LRCX','LM','LEG','LEN','LUK','LLY','LNC','LLTC','LMT','L','LO','LOW','LSI','LYB','MTB','MAC','M','MRO','MPC','MAR','MMC','MAS','MA','MAT','MKC','MCD','MHFI','MCK','MJN','MWV','MDT','MRK','MET','MCHP','MU','MSFT','MHK','TAP','MDLZ','MON','MNST','MCO','MS','MOS','MSI','MUR','MYL','NBR','NDAQ','NOV','NTAP','NFLX','NWL','NFX','NEM','NWSA','NEE','NLSN','NKE','NI','NE','NBL','JWN','NSC','NTRS','NOC','NU','NRG','NUE','NVDA','KORS','ORLY','OXY','OMC','OKE','ORCL','OI','PCG','PCAR','PLL','PH','PDCO','PAYX','BTU','PNR','PBCT','POM','PEP','PKI','PRGO','PETM','PFE','PM','PSX','PNW','PXD','PBI','PCL','PNC','RL','PPG','PPL','PX','PCP','PCLN','PFG','PG','PGR','PLD','PRU','PEG','PSA','PHM','PVH','QEP','PWR','QCOM','DGX','RRC','RTN','RHT','REGN','RF','RSG','RAI','RHI','ROK','COL','ROP','ROST','RDC','R','SWY','CRM','SNDK','SCG','SLB','SNI','STX','SEE','SRE','SHW','SIAL','SPG','SLM','SJM','SNA','SO','LUV','SWN','SE','STJ','SWK','SPLS','SBUX','HOT','STT','SRCL','SYK','STI','SYMC','SYY','TROW','TGT','TEL','TE','THC','TDC','TSO','TXN','TXT','HSY','TRV','TMO','TIF','TWX','TWC','TJX','TMK','TSS','TSCO','RIG','TRIP','FOXA','TSN','TYC','USB','UNP','UNH','UPS','X','UTX','UNM','URBN','VFC','VLO','VAR','VTR','VRSN','VZ','VRTX','VIAB','V','VNO','VMC','WMT','WAG','DIS','GHC','WM','WAT','WLP','WFC','WDC','WU','WY','WHR','WFM','WMB','WIN','WEC','WYN','WYNN','XEL','XRX','XLNX','XL','XYL','YHOO','YUM','ZMH','ZION','ZTS']
+year = 2013
+
+#puts "Starting the sequential calculation..."
+#time, prices = timer do
+  #s_and_p_500.inject({}){|memo, symbol| memo[symbol] = get_year_end_closing(symbol, year).last; memo }
+#end
+#puts "Sequential time: #{strftimer(time)}"
+
+puts "Starting the concurrent calculation..."
+time, prices = timer do
+  futures = s_and_p_500.collect{|symbol| Concurrent::Future.execute{ get_year_end_closing(symbol, year) } }
+  futures.inject({}){|memo, future| memo[future.value.first] = future.value.last; memo }
+end
+puts "Concurrent time: #{strftimer(time)}"

lib/concurrent.rb

@@ -37,7 +37,6 @@
 require 'concurrent/immediate_executor'
 require 'concurrent/per_thread_executor'
 require 'concurrent/thread_pool_executor'
-require 'concurrent/uses_global_thread_pool'
 
 # Modern concurrency tools for Ruby. Inspired by Erlang, Clojure, Scala, Haskell,
 # F#, C#, Java, and classic concurrency patterns.

lib/concurrent/agent.rb

@@ -1,8 +1,8 @@
 require 'thread'
 require 'observer'
 
+require 'concurrent/configuration'
 require 'concurrent/dereferenceable'
-require 'concurrent/uses_global_thread_pool'
 require 'concurrent/utilities'
 
 module Concurrent
@@ -34,7 +34,7 @@ module Concurrent
   #   @return [Fixnum] the maximum number of seconds before an update is cancelled
   class Agent
     include Dereferenceable
-    include UsesGlobalThreadPool
+    include OptionsParser
 
     # The default timeout value (in seconds); used when no timeout option
     # is given at initialization
@@ -46,7 +46,15 @@ class Agent
     #
     # @param [Object] initial the initial value
     # @param [Hash] opts the options used to define the behavior at update and deref
+    #
     # @option opts [Fixnum] :timeout (TIMEOUT) maximum number of seconds before an update is cancelled
+    #
+    # @option opts [Boolean] :operation (false) when +true+ will execute the future on the global
+    #   operation pool (for long-running operations), when +false+ will execute the future on the
+    #   global task pool (for short-running tasks)
+    # @option opts [object] :executor when provided will run all operations on
+    #   this executor rather than the global thread pool (overrides :operation)
+    #
     # @option opts [String] :dup_on_deref (false) call +#dup+ before returning the data
     # @option opts [String] :freeze_on_deref (false) call +#freeze+ before returning the data
     # @option opts [String] :copy_on_deref (nil) call the given +Proc+ passing the internal value and
@@ -57,6 +65,7 @@ def initialize(initial, opts = {})
       @validator = Proc.new { |result| true }
       @timeout = opts.fetch(:timeout, TIMEOUT).freeze
       @observers = CopyOnWriteObserverSet.new
+      @executor = get_executor_from(opts)
       init_mutex
       set_deref_options(opts)
     end
@@ -116,7 +125,7 @@ def validate(&block)
     # @yieldparam [Object] value the current value
     # @yieldreturn [Object] the new value
     def post(&block)
-      Agent.thread_pool.post{ work(&block) } unless block.nil?
+      @executor.post{ work(&block) } unless block.nil?
     end
 
     # Update the current value with the result of the given block operation

lib/concurrent/async.rb

@@ -1,6 +1,8 @@
 require 'thread'
+require 'concurrent/configuration'
 require 'concurrent/ivar'
 require 'concurrent/future'
+require 'concurrent/thread_pool_executor'
 
 module Concurrent
 
@@ -169,8 +171,9 @@ class AsyncDelegator # :nodoc:
       #
       # @param [Object] delegate the object to wrap and delegate method calls to
       # @param [Mutex] mutex the mutex lock to use when delegating method calls
-      def initialize(delegate, mutex)
+      def initialize(delegate, executor, mutex)
         @delegate = delegate
+        @executor = executor
         @mutex = mutex
       end
 
@@ -191,7 +194,7 @@ def method_missing(method, *args, &block)
 
         self.define_singleton_method(method) do |*args|
           Async::validate_argc(@delegate, method, *args)
-          Concurrent::Future.execute do
+          Concurrent::Future.execute(executor: @executor) do
             mutex.synchronize do
               @delegate.send(method, *args, &block)
             end
@@ -237,7 +240,7 @@ def method_missing(method, *args, &block)
     #
     # @see Concurrent::Future
     def async
-      @__async_delegator__ ||= AsyncDelegator.new(self, await.mutex)
+      @__async_delegator__ ||= AsyncDelegator.new(self, executor, await.mutex)
     end
     alias_method :future, :async
 
@@ -272,5 +275,16 @@ def await
       @__await_delegator__ ||= AwaitDelegator.new(self, Mutex.new)
     end
     alias_method :delay, :await
+
+    def executor=(executor)
+      raise ArgumentError.new('executor has already been set') unless @__async__executor__.nil?
+      @__async__executor__ = executor
+    end
+
+    private
+
+    def executor
+      @__async__executor__ ||= Concurrent.configuration.global_task_pool
+    end
   end
 end

lib/concurrent/configuration.rb

@@ -1,4 +1,5 @@
-require 'concurrent/per_thread_executor'
+require 'concurrent/thread_pool_executor'
+require 'concurrent/processor_count'
 
 module Concurrent
   class << self
@@ -10,13 +11,89 @@ def self.configure
   end
 
   class Configuration
-    attr_accessor :global_thread_pool
+    attr_accessor :global_task_pool
+    attr_accessor :global_operation_pool
 
     def initialize
-      @global_thread_pool = Concurrent::PerThreadExecutor.new
+    end
+
+    def cores
+      @cores ||= Concurrent::processor_count
+    end
+
+    def global_task_pool
+      @global_task_pool ||= Concurrent::ThreadPoolExecutor.new(
+        min_threads: [2, cores].max,
+        max_threads: [20, cores * 15].max,
+        idletime: 2 * 60,                  # 2 minutes
+        max_queue: 0,                      # unlimited
+        overflow_policy: :abort            # raise an exception
+      )
+    end
+
+    def global_operation_pool
+      @global_operation_pool = Concurrent::ThreadPoolExecutor.new(
+        min_threads: [2, cores].max,
+        max_threads: [2, cores].max,
+        idletime: 10 * 60,                 # 10 minutes
+        max_queue: [20, cores * 15].max,
+        overflow_policy: :abort            # raise an exception
+      )
+    end
+
+    def global_task_pool=(executor)
+      finalize_executor(@global_task_pool)
+      @global_task_pool = executor
+    end
+
+    def global_operation_pool=(executor)
+      finalize_executor(@global_operation_pool)
+      @global_operation_pool = executor
+    end
+
+    private
+
+    def finalize_executor(executor)
+      return if executor.nil?
+      if executor.respond_to?(:shutdown)
+        executor.shutdown
+      elsif executor.respond_to?(:kill)
+        executor.kill
+      end
+    rescue
+      # suppress
+    end
+  end
+
+  module OptionsParser
+
+    def get_executor_from(opts = {})
+      if opts.has_key?(:executor)
+        opts[:executor]
+      elsif opts[:operation] == true || opts[:task] == false
+        Concurrent.configuration.global_operation_pool
+      else
+        Concurrent.configuration.global_task_pool
+      end
     end
   end
 
+  def task(*args, &block)
+    Concurrent.configuration.global_task_pool.post(*args, &block)
+  end
+  module_function :task
+
+  def operation(*args, &block)
+    Concurrent.configuration.global_operation_pool.post(*args, &block)
+  end
+  module_function :operation
+
   # create the default configuration on load
   self.configuration = Configuration.new
+
+  # set exit hook to shutdown global thread pools
+  at_exit do
+    self.configuration.global_task_pool = nil
+    self.configuration.global_operation_pool = nil
+  end
 end

lib/concurrent/dataflow.rb

@@ -1,5 +1,6 @@
 require 'concurrent/atomic'
 require 'concurrent/future'
+require 'concurrent/per_thread_executor'
 
 module Concurrent
 

lib/concurrent/future.rb

@@ -1,7 +1,7 @@
 require 'thread'
 
+require 'concurrent/configuration'
 require 'concurrent/obligation'
-require 'concurrent/uses_global_thread_pool'
 require 'concurrent/safe_task_executor'
 
 module Concurrent
@@ -41,13 +41,18 @@ module Concurrent
   # @see http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Future.html java.util.concurrent.Future
   class Future < IVar
     include Obligation
-    include UsesGlobalThreadPool
+    include OptionsParser
 
     # Create a new +Future+ in the +:unscheduled+ state.
     #
     # @yield the asynchronous operation to perform
     #
     # @param [Hash] opts the options to create a message with
+    # @option opts [Boolean] :operation (false) when +true+ will execute the future on the global
+    #   operation pool (for long-running operations), when +false+ will execute the future on the
+    #   global task pool (for short-running tasks)
+    # @option opts [object] :executor when provided will run all operations on
+    #   this executor rather than the global thread pool (overrides :operation)
     # @option opts [String] :dup_on_deref (false) call +#dup+ before returning the data
     # @option opts [String] :freeze_on_deref (false) call +#freeze+ before returning the data
     # @option opts [String] :copy_on_deref (nil) call the given +Proc+ passing the internal value and
@@ -59,6 +64,7 @@ def initialize(opts = {}, &block)
       super(IVar::NO_VALUE, opts)
       @state = :unscheduled
       @task = block
+      @executor = get_executor_from(opts)
     end
 
     # Execute an +:unscheduled+ +Future+. Immediately sets the state to +:pending+ and
@@ -80,7 +86,7 @@ def initialize(opts = {}, &block)
     # @since 0.5.0
     def execute
       if compare_and_set_state(:pending, :unscheduled)
-        Future.thread_pool.post { work }
+        @executor.post{ work }
         self
       end
     end
@@ -105,7 +111,7 @@ def execute
     #
     # @since 0.5.0
     def self.execute(opts = {}, &block)
-      return Future.new(opts, &block).execute
+      Future.new(opts, &block).execute
     end
 
     protected :set, :fail, :complete
@@ -117,6 +123,5 @@ def work # :nodoc:
       success, val, reason = SafeTaskExecutor.new(@task).execute
       complete(success, val, reason)
     end
-    
   end
 end

lib/concurrent/java_cached_thread_pool.rb

@@ -36,6 +36,8 @@ def initialize(opts = {})
           idletime, java.util.concurrent.TimeUnit::SECONDS,
           java.util.concurrent.SynchronousQueue.new,
           OVERFLOW_POLICIES[@overflow_policy].new)
+
+        at_exit { self.kill }
       end
     end
   end

lib/concurrent/java_fixed_thread_pool.rb

@@ -28,6 +28,8 @@ def initialize(num_threads, opts = {})
           @max_queue, java.util.concurrent.TimeUnit::SECONDS,
           java.util.concurrent.LinkedBlockingQueue.new,
           OVERFLOW_POLICIES[@overflow_policy].new)
+
+        at_exit { self.kill }
       end
     end
   end

lib/concurrent/java_thread_pool_executor.rb

@@ -60,6 +60,8 @@ def initialize(opts = {})
           min_length, max_length,
           idletime, java.util.concurrent.TimeUnit::SECONDS,
           queue, OVERFLOW_POLICIES[@overflow_policy].new)
+
+        at_exit { self.kill }
       end
 
       def min_length