<commit_before># -*- coding: utf-8 -*-
require 'thread'

module Isono
  class ThreadPool
    include Logger

    class WorkerTerminateError < StandardError; end
    class TimeoutError < StandardError; end
    
    def initialize(worker_num=1, name=nil, opts={})
      set_instance_logger(name)
      @queue = ::Queue.new
      @name = name
      @opts = {:stucked_queue_num=>20}.merge(opts)
      @last_stuck_warn_at = Time.now
      
      @worker_threads = {}
      worker_num.times {
        t = Thread.new {
          loop {
            begin
              while op = @queue.pop
                if @queue.size > @opts[:stucked_queue_num] && Time.now - @last_stuck_warn_at > 5.0
                  logger.warn("too many stacked workers: #{@queue.size}")
                  @last_stuck_warn_at = Time.now
                end
                op.call
              end
            rescue WorkerTerminateError
              # someone indicated to terminate this thread
              # exit from the current loop
              break
            rescue Exception => e
              self.logger.error(e)
            end
          }
          @worker_threads.delete(Thread.current.__id__)
          logger.info("#{Thread.current} is being terminated")
        }
        @worker_threads[t.__id__] = t
      }
    end

    # Pass a block to a worker thread. The job is queued until the
    # worker thread found.
    # @param [Bool] immediage
    # @param [Proc] blk A block to be proccessed on a worker thread.
    def pass(immediate=true, &blk)
      if immediate && member_thread?
        return blk.call
      end
      
      @queue << blk
    end

    # Send a block to a worker thread similar with pass(). but this
    # get the caller thread waited until the block proceeded in a
    # worker thread.
    # @param [Bool] immediage
    # @param [Float] time_out 
    # @param [Proc] blk
    def barrier(immediate=true, time_out=nil, &blk)
      if immediate && member_thread?
        return blk.call
      end
      
      q = ::Queue.new
      time_start = ::Time.now
      
      self.pass {
        begin
          q << blk.call
        rescue Exception => e
          q << e
        end
      }

      em_sig = nil
      if time_out
        em_sig = EventMachine.add_timer(time_out) {
          q << TimeoutError.new
        }
      end
      
      res = q.shift
      EventMachine.cancel_timer(em_sig)
      time_elapsed = ::Time.now - time_start
      logger.debug("Elapsed time for #{blk}: #{time_elapsed} secs") if time_elapsed > 0.05
      if res.is_a?(Exception)
        raise res
      end
      res
      
    end

    def clear
      @queue.clear
    end

    # Immediatly shutdown all the worker threads
    def shutdown()
      @worker_threads.each {|t|
        t.raise WorkerTerminateError
      }<commit_msg>fix shutdown() in ThreadPool that called key part of hash.<commit_after># -*- coding: utf-8 -*-
require 'thread'

module Isono
  class ThreadPool
    include Logger

    class WorkerTerminateError < StandardError; end
    class TimeoutError < StandardError; end
    
    def initialize(worker_num=1, name=nil, opts={})
      set_instance_logger(name)
      @queue = ::Queue.new
      @name = name
      @opts = {:stucked_queue_num=>20}.merge(opts)
      @last_stuck_warn_at = Time.now
      
      @worker_threads = {}
      worker_num.times {
        t = Thread.new {
          loop {
            begin
              while op = @queue.pop
                if @queue.size > @opts[:stucked_queue_num] && Time.now - @last_stuck_warn_at > 5.0
                  logger.warn("too many stacked workers: #{@queue.size}")
                  @last_stuck_warn_at = Time.now
                end
                op.call
              end
            rescue WorkerTerminateError
              # someone indicated to terminate this thread
              # exit from the current loop
              break
            rescue Exception => e
              self.logger.error(e)
            end
          }
          @worker_threads.delete(Thread.current.__id__)
          logger.info("#{Thread.current} is being terminated")
        }
        @worker_threads[t.__id__] = t
      }
    end

    # Pass a block to a worker thread. The job is queued until the
    # worker thread found.
    # @param [Bool] immediage
    # @param [Proc] blk A block to be proccessed on a worker thread.
    def pass(immediate=true, &blk)
      if immediate && member_thread?
        return blk.call
      end
      
      @queue << blk
    end

    # Send a block to a worker thread similar with pass(). but this
    # get the caller thread waited until the block proceeded in a
    # worker thread.
    # @param [Bool] immediage
    # @param [Float] time_out 
    # @param [Proc] blk
    def barrier(immediate=true, time_out=nil, &blk)
      if immediate && member_thread?
        return blk.call
      end
      
      q = ::Queue.new
      time_start = ::Time.now
      
      self.pass {
        begin
          q << blk.call
        rescue Exception => e
          q << e
        end
      }

      em_sig = nil
      if time_out
        em_sig = EventMachine.add_timer(time_out) {
          q << TimeoutError.new
        }
      end
      
      res = q.shift
      EventMachine.cancel_timer(em_sig)
      time_elapsed = ::Time.now - time_start
      logger.debug("Elapsed time for #{blk}: #{time_elapsed} secs") if time_elapsed > 0.05
      if res.is_a?(Exception)
        raise res
      end
      res
      
    end

    def clear
      @queue.clear
    end

    # Immediatly shutdown all the worker threads
    def shutdown()
      @worker_threads.each {|id, t|
        t.__send__(:raise, WorkerTerminateError)
        Thread.pass
      }
{'fraction_non_alphanumeric': 0.08710071565718275, 'fraction_numerical': 0.0031419095828242277, 'mean_word_length': 1.700282752120641, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5490739', 'n_tokens_mistral': 1711, 'n_tokens_neox': 1672, 'n_words': 475}