<commit_before>require "forwardable"
require "pwrake/task/task_rank"

module Pwrake

  class QueueArray < Array

    def initialize(nproc)
      @nproc = nproc
      super()
    end

    def shift(host_info, min_core)
      return super() unless host_info
      i_tried = []
      count = 0
      size.times do |i|
        tw = q_at(i)
        if tw.tried_host?(host_info)
          i_tried << i
        elsif tw.acceptable_for(host_info)
          use_core = tw.use_cores(host_info)
          if use_core > min_core
            Log.debug "qa1: task=#{tw.name} use_core=#{use_core} i=#{i}/#{size} min_core=#{min_core}"
            return q_delete_at(i)
          end
          if min_core > 0
            count += use_core
            if count > @nproc
              break
            end
          end
        end
      end
      i_tried.each do |i|
        tw = q_at(i)
        if tw.acceptable_for(host_info)
          use_core = tw.use_cores(host_info)
          Log.debug "qa4: task=#{tw.name} use_core=#{use_core} i=#{i}/#{size}"
          return q_delete_at(i)
        end
      end
      nil
    end
  end

  class LifoQueueArray < QueueArray
    def q_at(i)
      at(size-1-i)
    end
    def q_delete_at(i)
      delete_at(size-1-i)
    end
  end

  class FifoQueueArray < QueueArray
    def q_at(i)
      at(i)
    end
    def q_delete_at(i)
      delete_at(i)
    end
  end


  # HRF mixin module
  module HrfQueue

    def hrf_init(nproc)
      @nproc = nproc
      @count = []
    end

    def hrf_push(t)
      r = t.rank
      n = t.use_cores(@nproc)
      @count[r] = (@count[r] || 0) + n
    end

    def hrf_get(host_info, min_core)
      (@count.size-1).downto(0) do |r|
        c = @count[r]
        if c && c>0
          t = (c <= @nproc) ?
            pop_last_rank(r, host_info, min_core) :
            pop_super(host_info, min_core)
          hrf_delete(t) if t
          return t
        end
      end
      raise "no element"
      nil
    end

    def pop_last_rank(r, host_info, min_core)
      i_tried = []
      count = 0
      size.times do |i|
        tw = q_at(i)
        if tw.rank == r
          if tw.tried_host?(host_info)
            i_tried << i
          elsif tw.acceptable_for(host_info)
            use_core = tw.use_cores(@nproc)
            if use_core > min_core
              Log.debug "qa5: task=#{tw.name} use_core=#{use_core} @count[rank=#{r}]=#{@count[r]} i=#{i}/#{size} min_core=#{min_core}"
              return q_delete_at(i)
            end
            if min_core > 0
              count += use_core
              if count > @nproc
                break
              end
            end
          end
        end
      end
      i_tried.each do |i|
        tw = q_at(i)
        if tw.acceptable_for(host_info)
          use_core = tw.use_cores(host_info)
          Log.debug "qa6: task=#{tw.name} use_core=#{use_core}"
          return q_delete_at(i)
        end
      end
      nil
    end

    def hrf_delete(t)
      @count[t.rank] -= t.use_cores(@nproc)
    end
  end


  # LIFO + HRF
  class LifoHrfQueueArray
    include HrfQueue
    extend Forwardable<commit_msg>change condition to break multi-core task search in QueueArray from count > @nproc to count >= @nproc<commit_after>require "forwardable"
require "pwrake/task/task_rank"

module Pwrake

  class QueueArray < Array

    def initialize(nproc)
      @nproc = nproc
      super()
    end

    def shift(host_info, min_core)
      return super() unless host_info
      i_tried = []
      count = 0
      size.times do |i|
        tw = q_at(i)
        if tw.tried_host?(host_info)
          i_tried << i
        elsif tw.acceptable_for(host_info)
          use_core = tw.use_cores(host_info)
          if use_core > min_core
            Log.debug "qa1: task=#{tw.name} use_core=#{use_core} i=#{i}/#{size} min_core=#{min_core}"
            return q_delete_at(i)
          end
          if min_core > 0
            count += use_core
            if count >= @nproc
              break
            end
          end
        end
      end
      i_tried.each do |i|
        tw = q_at(i)
        if tw.acceptable_for(host_info)
          use_core = tw.use_cores(host_info)
          Log.debug "qa4: task=#{tw.name} use_core=#{use_core} i=#{i}/#{size}"
          return q_delete_at(i)
        end
      end
      nil
    end
  end

  class LifoQueueArray < QueueArray
    def q_at(i)
      at(size-1-i)
    end
    def q_delete_at(i)
      delete_at(size-1-i)
    end
  end

  class FifoQueueArray < QueueArray
    def q_at(i)
      at(i)
    end
    def q_delete_at(i)
      delete_at(i)
    end
  end


  # HRF mixin module
  module HrfQueue

    def hrf_init(nproc)
      @nproc = nproc
      @count = []
    end

    def hrf_push(t)
      r = t.rank
      n = t.use_cores(@nproc)
      @count[r] = (@count[r] || 0) + n
    end

    def hrf_get(host_info, min_core)
      (@count.size-1).downto(0) do |r|
        c = @count[r]
        if c && c>0
          t = (c <= @nproc) ?
            pop_last_rank(r, host_info, min_core) :
            pop_super(host_info, min_core)
          hrf_delete(t) if t
          return t
        end
      end
      raise "no element"
      nil
    end

    def pop_last_rank(r, host_info, min_core)
      i_tried = []
      count = 0
      size.times do |i|
        tw = q_at(i)
        if tw.rank == r
          if tw.tried_host?(host_info)
            i_tried << i
          elsif tw.acceptable_for(host_info)
            use_core = tw.use_cores(@nproc)
            if use_core > min_core
              Log.debug "qa5: task=#{tw.name} use_core=#{use_core} @count[rank=#{r}]=#{@count[r]} i=#{i}/#{size} min_core=#{min_core}"
              return q_delete_at(i)
            end
            if min_core > 0
              count += use_core
              if count >= @nproc
                break
              end
            end
          end
        end
      end
      i_tried.each do |i|
        tw = q_at(i)
        if tw.acceptable_for(host_info)
          use_core = tw.use_cores(host_info)
          Log.debug "qa6: task=#{tw.name} use_core=#{use_core}"
          return q_delete_at(i)
        end
      end
      nil
    end

    def hrf_delete(t)
      @count[t.rank] -= t.use_cores(@nproc)
    end
  end


  # LIFO + HRF
  class LifoHrfQueueArray
    include HrfQueue
    extend Forwardable
{'fraction_non_alphanumeric': 0.10337220296249607, 'fraction_numerical': 0.00441222817522849, 'mean_word_length': 1.5969721767594107, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 19, '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': '3494273', 'n_tokens_mistral': 2327, 'n_tokens_neox': 2266, 'n_words': 501}