<commit_before>module Pwrake

  class TaskQueue

    def initialize(core_map, group_map=nil)
      @q = []
      @empty = []
      @finished = false

      @enable_steal = true
      @q_no_action = Array.new

      @idle_cores = core_map.dup

      pri = Rake.application.pwrake_options['QUEUE_PRIORITY'] || "FIFO"#"RANK"
      case pri
      when /prio/i
        @array_class = PriorityQueueArray
      when /fifo/i
        @array_class = FifoQueueArray # Array # Fifo
      when /lifo/i
        @array_class = LifoQueueArray
      when /lihr/i
        @array_class = LifoHrfQueueArray
      when /prhr/i
        @array_class = PriorityHrfQueueArray
      when /rank/i
        @array_class = RankQueueArray
      else
        raise RuntimeError,"unknown option for QUEUE_PRIORITY: "+pri
      end
      #Log.debug "@array_class=#{@array_class.inspect}"
      init_queue(core_map, group_map)
    end

    def init_queue(core_map, group_map=nil)
      #@q_input = @array_class.new(core_map.size)
      @q_input = Array.new
      @q_no_input = Array.new
    end

    #attr_reader :mutex
    attr_accessor :enable_steal

    # enq
    def enq(tw)
      if tw.nil? || tw.actions.empty?
        @q_no_action.push(tw)
      else
        enq_body(tw)
      end
    end

    def enq_body(tw)
      enq_impl(tw)
    end

    def enq_impl(tw)
      if tw.task.kind_of?(Rake::FileTask) and !tw.prerequisites.empty?
        @q_input.push(tw)
      else
        @q_no_input.push(tw)
      end
    end


    def deq_task(&block) # simple version
      queued = deq_loop(&block)
      if queued>0
        Log.debug "queued:#{queued} @idle_cores:#{@idle_cores.inspect}"
      end
    end

    def deq_loop(steal,&block)
      queued = 0
      while true
        count = 0
        @idle_cores.keys.each do |hid|
          #if t = deq(@workers[hid].host)
          if tw = deq_impl(hid,steal)
            Log.debug "deq: #{tw.name}"
            if @idle_cores[hid] < tw.n_used_cores
              enq(tw) # check me
            else
              @idle_cores.decrease(hid, tw.n_used_cores)
              yield(tw,hid)
              count += 1
              queued += 1
            end
          end
        end
        break if count == 0
      end
      queued
    end

    def deq_impl(hint=nil, steal=nil)
      @q_no_action.shift ||
        @q_input.shift ||
        @q_no_input.shift
    end

    def clear
      @q_no_action.clear
      @q_input.clear
      @q_no_input.clear
    end

    def empty?
      @q_no_action.empty? &&<commit_msg>return if empty? in TaskQueue#deq_loop<commit_after>module Pwrake

  class TaskQueue

    def initialize(core_map, group_map=nil)
      @q = []
      @empty = []
      @finished = false

      @enable_steal = true
      @q_no_action = Array.new

      @idle_cores = core_map.dup

      pri = Rake.application.pwrake_options['QUEUE_PRIORITY'] || "FIFO"#"RANK"
      case pri
      when /prio/i
        @array_class = PriorityQueueArray
      when /fifo/i
        @array_class = FifoQueueArray # Array # Fifo
      when /lifo/i
        @array_class = LifoQueueArray
      when /lihr/i
        @array_class = LifoHrfQueueArray
      when /prhr/i
        @array_class = PriorityHrfQueueArray
      when /rank/i
        @array_class = RankQueueArray
      else
        raise RuntimeError,"unknown option for QUEUE_PRIORITY: "+pri
      end
      #Log.debug "@array_class=#{@array_class.inspect}"
      init_queue(core_map, group_map)
    end

    def init_queue(core_map, group_map=nil)
      #@q_input = @array_class.new(core_map.size)
      @q_input = Array.new
      @q_no_input = Array.new
    end

    #attr_reader :mutex
    attr_accessor :enable_steal

    # enq
    def enq(tw)
      if tw.nil? || tw.actions.empty?
        @q_no_action.push(tw)
      else
        enq_body(tw)
      end
    end

    def enq_body(tw)
      enq_impl(tw)
    end

    def enq_impl(tw)
      if tw.task.kind_of?(Rake::FileTask) and !tw.prerequisites.empty?
        @q_input.push(tw)
      else
        @q_no_input.push(tw)
      end
    end


    def deq_task(&block) # simple version
      queued = deq_loop(&block)
      if queued>0
        Log.debug "queued:#{queued} @idle_cores:#{@idle_cores.inspect}"
      end
    end

    def deq_loop(steal,&block)
      queued = 0
      while true
        count = 0
        @idle_cores.keys.each do |hid|
          if empty?
            return queued
          #if t = deq(@workers[hid].host)
          elsif tw = deq_impl(hid,steal)
            Log.debug "deq: #{tw.name}"
            if @idle_cores[hid] < tw.n_used_cores
              enq(tw) # check me
            else
              @idle_cores.decrease(hid, tw.n_used_cores)
              yield(tw,hid)
              count += 1
              queued += 1
            end
          end
        end
        break if count == 0
      end
      queued
    end

    def deq_impl(hint=nil, steal=nil)
      @q_no_action.shift ||
        @q_input.shift ||
        @q_no_input.shift
    end

    def clear
      @q_no_action.clear
      @q_input.clear
      @q_no_input.clear
    end

    def empty?
      @q_no_action.empty? &&
{'fraction_non_alphanumeric': 0.09805068226120858, 'fraction_numerical': 0.0023391812865497076, 'mean_word_length': 1.8115068493150686, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, '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': '1053549', 'n_tokens_mistral': 1853, 'n_tokens_neox': 1804, 'n_words': 356}