pool:: ActionPool for processing actions
 Creates a new timer
 Argument hash: {:pool, :logger, :auto_start}
 Forcibly wakes the timer early
540595
def wakeup
      raise NotRunning.new unless running?
      if(@sleeper.waiters > 0)
        @sleeper.signal
      else
        @timer_thread.wakeup if @timer_thread.alive? && @timer_thread.stop?
      end
    end
1107309
def start
      Filum.logger.info "Starting Workerpool"
      Larva::WorkerPool.start(@processors)
      Filum.logger.info "Workerpool Finished"
    end
391761
def _run(self):
        
        # dispatch auth data on first scheduling after start
        state_change = ActionUpdateTransportAuthData(
            f'{self._user_id}/{self._client.api.token}',
        )
        self.greenlet.name = f'MatrixTransport._run node:{pex(self._raiden_service.address)}'
        self._raiden_service.handle_and_track_state_change(state_change)
        try:
            # waits on _stop_event.ready()
            self._global_send_worker()
            # children crashes should throw an exception here
        except gevent.GreenletExit:  # killed without exception
            self._stop_event.set()
            gevent.killall(self.greenlets)  # kill children
            raise  # re-raise to keep killed status
        except Exception:
            self.stop()  # ensure cleanup and wait on subtasks
            raise
88372
def start
      @stop_mutex.synchronize do
        fail 'already stopped' if @stopped
      end
      until @workers.size == @size.to_i
        new_worker_queue = Queue.new
        @ready_workers << new_worker_queue
        next_thread = Thread.new(new_worker_queue) do |jobs|
          catch(:exit) do  # allows { throw :exit } to kill a thread
            loop_execute_jobs(jobs)
          end
          remove_current_thread
        end
        @workers << next_thread
      end
    end
1006113
def pop
      alarm = (Time.now + @wait_timeout)
      clnt = nil
      dirty = false
      while !@shutdown
        raise_alarm if (Time.now > alarm)
        @mutex.synchronize do
          begin
            if clnt = @_pool.pop
              dirty = true
            else
              clnt = _fetch_new(&@client_block)
            end
          ensure
            if clnt
              _checkout(clnt)
              @cond.signal
            else
              @reaper.wakeup if @reaper && _dead_clients?
              @cond.wait(@mutex,@wait_timeout)
            end
          end
        end
        break if clnt
      end
      clean_client(clnt) if dirty && clnt
      clnt
    end
1239806
def start debug=false,method=:every_minute,*args
      @stopped = false
      @suspended = false
      @dead = Queue.new
      @thread = CronR::Utils.send(method,debug,*args) {
        time = self.time
        @mutex.synchronize {
          if @stopped then
            # It's important we put something on this queue ONLY AFTER
            # we've acquired the mutex...
            @dead.enq(true)
            true
          elsif @suspended then
          else
            self.run(time)
          end
        }
      }
    end
1306232
def _enable_thread_pool(func):
    

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        self = args[0]
        if self.enable_thread_pool and hasattr(self, 'thread_pool'):
            future = self.thread_pool.submit(func, *args, **kwargs)
            is_async = kwargs.get('is_async')
            if is_async is None or not is_async:
                timeout = kwargs.get('timeout')
                if timeout is None:
                    timeout = 2
                try:
                    result = future.result(timeout=timeout)
                except TimeoutError as e:
                    self.logger.exception(e)
                    result = None
                return result
            return future
        else:
            return func(*args, **kwargs)

    return wrapper
1352809
def _start_managers(self):
        
        self._task_queue = _Weave(self._tasks, self.stride)
        # here we determine the size of the maximum memory consumption
        self._semaphore_value = (self.buffer or (len(self._tasks) * self.stride))
        self._pool_semaphore = Semaphore(self._semaphore_value)


        # start the pool getter thread
        self._pool_getter = Thread(target=self._pool_get, args=(self._getout, \
                self._task_results, self._next_available, \
                self._task_next_lock, self._next_skipped, len(self._tasks), \
                len(self.pool), id(self)))
        self._pool_getter.deamon = True
        self._pool_getter.start()

        # start the pool putter thread
        self._pool_putter = Thread(target=self._pool_put, args=\
                (self._pool_semaphore, self._task_queue, self._putin, \
                len(self.pool), id(self), self._stopping.isSet))
        self._pool_putter.deamon = True
        self._pool_putter.start()
1127191
def timer=(t)
      raise ArgumentError.new('Expecting an ActionTimer::Timer') unless t.is_a?(ActionTimer::Timer)
      @timer = t
    end
209246
def run(self):
    
    while True:
      if self.pool.name:
        self.idle = True

      try:
        # Wait 60 seconds for a message, otherwise exit. This ensures that the
        # threadpool will be trimmed down when load is light.
        task = self._queue.get(timeout=60)

        if self.pool.name:
          self.idle = False

        try:
          # The pool told us to quit, likely because it is stopping.
          if task == STOP_MESSAGE:
            return

          self.ProcessTask(*task)
        finally:
          self._queue.task_done()

      except queue.Empty:
        if self._RemoveFromPool():
          return

      # Try to trim old threads down when they get too old. This helps the
      # thread pool size to shrink, even when it is not idle (e.g. if it is CPU
      # bound) since threads are forced to exit, but new threads will not be
      # created if the utilization is too high - resulting in a reduction of
      # threadpool size under CPU load.
      if time.time() - self.started > 600 and self._RemoveFromPool():
        return
37137
def launch_worker
    @worker_threads << Thread.new {
      while work = @work_queue.pop
        begin
          #RJR::Logger.debug "launch thread pool job #{work}"
          @running_queue << work
          work.exec(@thread_lock)
          # TODO cleaner / more immediate way to pop item off running_queue
          #RJR::Logger.debug "finished thread pool job #{work}"
        rescue Exception => e
          # FIXME also send to rjr logger at a critical level
          puts "Thread raised Fatal Exception #{e}"
          puts "\n#{e.backtrace.join("\n")}"
        end
      end
    }
  end
862563
function (action, payload, args, immediate) {
        var instruction = this.pool.create(action, payload, args);

        // based on whether the immediate flag is set, add to the top or bottom of the instruction queue.
        (immediate ? this.pool.unshift : this.pool.push)(instruction);

        return instruction;
    }
728441
def add(hash, &func)
      raise ArgumentError.new('Expecting hash of arguments') unless hash.is_a?(Hash)
      raise ArgumentError.new('A period must be provided for timed action') unless hash[:period]
      raise ArgumentError.new('Block must be provided') unless block_given?
      raise ArgumentError.new('Block must accept data value') if hash[:data] && func.arity == 0
      args = {:once => false, :data => nil, :owner => nil}.merge(hash)
      action = Action.new(args.merge(:timer => self), &func)
      @add_lock.synchronize{ @new_actions << action }
      wakeup if running?
      action
    end
477316
def on(method, *args, &block)

      if [:initialize, :before, :setup, :after].include? method
        raise "#{method} should not be called directly"
      end

      buffer  = [] # Container for the response buffer  
      threads = [] # Container for the execution threads

  		@registry.each do |s|
        if s.respond_to?(method.to_sym, false)
          threads << Thread.new do 
            responder = DispatchResponder.new(s)
          
            responder.trap(@timeout) do 
              if s.method(method.to_sym).arity == 0
                s.send(method, &block)
              else
                s.send(method, *args, &block)
              end
            end

            responder.finalize

            buffer << responder.to_h
          end
  			end
  		end

      threads.map(&:join)

      buffer
  	end
1100133
def run(action, args)
      args = args.respond_to?(:fixed_flatten) ? args.fixed_flatten(1) : args.flatten(1)
      begin
        unless(@action_timeout.zero?)
          Timeout::timeout(@action_timeout) do
            action.call(*args)
          end
        else
          action.call(*args)
        end
      rescue Timeout::Error => boom
        @logger.warn("Pool thread reached max execution time for action: #{boom}")
        raise boom
      end
    end
809282
def <<(action)
      case action
        when Proc
          queue(action)
        when Array
          raise ArgumentError.new('Actions to be processed by the pool must be a proc/lambda or [proc/lambda, [*args]]') unless action.size == 2 and action[0].is_a?(Proc) and action[1].is_a?(Array)
          queue(action[0], action[1])
        else
          raise ArgumentError.new('Actions to be processed by the pool must be a proc/lambda or [proc/lambda, [*args]]')
      end
      nil
    end
1306942
def start_time
      @start_time ||= @events.select { |event|
        event.is_a?(RequestWillBeSent)
      }.select { |event|
        event.request['url'] == @url
      }.map { |event|
        event.timestamp
      }.first
    end