<commit_before>module Pwrake

  class TaskQueue

    def initialize(*args)
      @finished = false
      @halt = false
      @mutex = Mutex.new
      @cv = ConditionVariable.new
      @th_end = []
      @enable_steal = true
      @q = []
      @reservation = {}
      @reserved_q = {}
    end

    attr_reader :mutex
    attr_accessor :enable_steal

    def reserve(item)
      @reservation[item] = Thread.current
    end

    def halt
      @mutex.synchronize do
        @halt = true
      end
    end

    def resume
      @mutex.synchronize do
        @halt = false
        @cv.broadcast
      end
    end

    def enq_synchronize
      if @halt
        ret = yield
      else
        @mutex.synchronize do
          ret = yield
          enq_finish
        end
        @cv.broadcast
      end
      @reserved_q.keys.each do |th|
        Log.debug "--- run #{th}";
        th.run
      end
      ret
    end

    def enq_finish
    end


    def enq(item,hint=nil)
      if th = @reservation[item]
        @reserved_q[th] = item
      else
        enq_impl(item,hint)
      end
    end


    def enq_bak(item,hint=nil)
      # Log.debug "--- #{self.class}#enq #{item.inspect}"
      th = nil
      if @halt
        if th = @reservation[item]
          @reserved_q[th] = item
        else
          enq_impl(item,hint)
        end
      else
        @mutex.synchronize do
          if th = @reservation[item]
            @reserved_q[th] = item
          else
            enq_impl(item,hint)
            @cv.signal
          end
        end
      end
      @reserved_q.keys.each{|th|
        Log.debug "--- run #{th}";
        th.run
      }
    end

    def enq_impl(item,hint)
      @q.push(item)          # FIFO Queue
    end


    def deq(hint=nil)
      # Log.debug "--- #{self.class}#deq #{self.inspect}"
      n = 0
      loop do
        @mutex.synchronize do
          if @th_end.first == Thread.current
            @th_end.shift
            return false

          elsif @halt
            Log.debug "--- halt in #{self.class}#deq @q=#{@q.inspect}"
            @cv.wait(@mutex)
            n = 0

          elsif item = @reserved_q.delete(Thread.current)
            Log.debug "--- deq from reserved_q=#{item.inspect}"
            return item

          elsif empty? # no item in queue
            Log.debug "--- empty=true in #{self.class}#deq @finished=#{@finished.inspect}"
            if @finished
              @cv.signal
              return false
            end
            Log.debug "--- waiting in #{self.class}#deq @finished=#{@finished.inspect}"
            @cv.wait(@mutex)
            n = 0

          else
            if t = deq_impl(hint,n)
              Log.debug "--- #{self.class}#deq #{t.inspect}"
              return t
            end
            n += 1
          end
        end
      end
    end

    def deq_impl(hint,n)
      @q.shift               # FIFO Queue
    end

    def clear
      @q.clear
      @reserved_q.clear
    end

    def empty?
      @q.empty? && @reserved_q.empty?
    end

    def finish
      Log.debug "--- #{self.class}#finish"
      @finished = true
      @cv.signal
    end

    def stop
      @mutex.synchronize do
        clear
        finish
      end
    end
<commit_msg>modify Log<commit_after>module Pwrake

  class TaskQueue

    def initialize(*args)
      @finished = false
      @halt = false
      @mutex = Mutex.new
      @cv = ConditionVariable.new
      @th_end = []
      @enable_steal = true
      @q = []
      @reservation = {}
      @reserved_q = {}
    end

    attr_reader :mutex
    attr_accessor :enable_steal

    def reserve(item)
      @reservation[item] = Thread.current
    end

    def halt
      @mutex.synchronize do
        @halt = true
      end
    end

    def resume
      @mutex.synchronize do
        @halt = false
        @cv.broadcast
      end
    end

    def enq_synchronize
      if @halt
        ret = yield
      else
        @mutex.synchronize do
          ret = yield
          enq_finish
        end
        @cv.broadcast
      end
      @reserved_q.keys.each do |th|
        Log.debug "--- run #{th}";
        th.run
      end
      ret
    end

    def enq_finish
    end


    def enq(item,hint=nil)
      if th = @reservation[item]
        @reserved_q[th] = item
      else
        enq_impl(item,hint)
      end
    end


    def enq_bak(item,hint=nil)
      # Log.debug "--- #{self.class}#enq #{item.inspect}"
      th = nil
      if @halt
        if th = @reservation[item]
          @reserved_q[th] = item
        else
          enq_impl(item,hint)
        end
      else
        @mutex.synchronize do
          if th = @reservation[item]
            @reserved_q[th] = item
          else
            enq_impl(item,hint)
            @cv.signal
          end
        end
      end
      @reserved_q.keys.each{|th|
        Log.debug "--- run #{th}";
        th.run
      }
    end

    def enq_impl(item,hint)
      @q.push(item)          # FIFO Queue
    end


    def deq(hint=nil)
      # Log.debug "--- #{self.class}#deq #{self.inspect}"
      n = 0
      loop do
        @mutex.synchronize do
          if @th_end.first == Thread.current
            @th_end.shift
            return false

          elsif @halt
            Log.debug "--- halt in TQ#deq @q=#{@q.inspect}"
            @cv.wait(@mutex)
            n = 0

          elsif item = @reserved_q.delete(Thread.current)
            Log.debug "--- deq from reserved_q=#{item.inspect}"
            return item

          elsif empty? # no item in queue
            #Log.debug "--- empty=true in #{self.class}#deq @finished=#{@finished.inspect}"
            if @finished
              @cv.signal
              return false
            end
            #Log.debug "--- waiting in #{self.class}#deq @finished=#{@finished.inspect}"
            @cv.wait(@mutex)
            n = 0

          else
            if t = deq_impl(hint,n)
              Log.debug "--- TQ#deq #{t.inspect}"
              return t
            end
            n += 1
          end
        end
      end
    end

    def deq_impl(hint,n)
      @q.shift               # FIFO Queue
    end

    def clear
      @q.clear
      @reserved_q.clear
    end

    def empty?
      @q.empty? && @reserved_q.empty?
    end

    def finish
      Log.debug "--- TQ#finish"
      @finished = true
      @cv.signal
    end

    def stop
      @mutex.synchronize do
        clear
        finish
      end
    end

{'fraction_non_alphanumeric': 0.10434782608695652, 'fraction_numerical': 0.0012422360248447205, 'mean_word_length': 1.2913553895410885, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '1325649', 'n_tokens_mistral': 2118, 'n_tokens_neox': 2037, 'n_words': 477}