<commit_before>(ns onyx.scheduling.percentage-job-scheduler
  (:require [onyx.scheduling.common-job-scheduler :as cjs]
            [onyx.scheduling.common-task-scheduler :as cts]
            [onyx.log.commands.common :as common]))

(defn sort-jobs-by-pct [replica]
    (let [indexed
          (map-indexed
           (fn [k j]
             {:position k :job j :pct (get-in replica [:percentages j])})
           (reverse (:jobs replica)))]
      (reverse (sort-by (juxt :pct :position) indexed))))

(defn min-allocations [jobs n-peers]
  (mapv
   (fn [job]
     (let [n (int (Math/floor (* (* 0.01 (:pct job)) n-peers)))]
       (assoc job :capacity n)))
   jobs))

(defn maximum-jobs-to-use [jobs]
  (reduce
   (fn [all {:keys [pct] :as job}]
     (let [sum (apply + (map :pct all))]
       (if (<= (+ sum pct) 100)
         (conj all job)
         (reduced all))))
   []
   jobs))

(defmethod cjs/job-offer-n-peers :onyx.job-scheduler/percentage
  [replica]
  (let [n-peers (count (:peers replica))
        sorted-jobs (sort-jobs-by-pct replica)
        jobs-to-use (maximum-jobs-to-use sorted-jobs)
        init-allocations (min-allocations jobs-to-use n-peers)]
    (into {} (map (fn [j] {(:job j) (:capacity j)}) init-allocations))))

(defn job-peer-count [replica job]
  (apply + (map count (vals (get-in replica [:allocations job])))))

(defmethod cjs/sort-job-priority :onyx.job-scheduler/percentage
  [replica jobs]
  (sort-by (juxt #(.indexOf ^clojure.lang.PersistentVector (vec (:jobs replica)) %)
                 (partial job-peer-count replica))<commit_msg>Rename function to better describe purpose<commit_after>(ns onyx.scheduling.percentage-job-scheduler
  (:require [onyx.scheduling.common-job-scheduler :as cjs]
            [onyx.scheduling.common-task-scheduler :as cts]
            [onyx.log.commands.common :as common]))

(defn sort-jobs-by-pct [replica]
    (let [indexed
          (map-indexed
           (fn [k j]
             {:position k :job j :pct (get-in replica [:percentages j])})
           (reverse (:jobs replica)))]
      (reverse (sort-by (juxt :pct :position) indexed))))

(defn min-allocations [jobs n-peers]
  (mapv
   (fn [job]
     (let [n (int (Math/floor (* (* 0.01 (:pct job)) n-peers)))]
       (assoc job :capacity n)))
   jobs))

(defn drop-jobs-overflow [jobs]
  (reduce
   (fn [all {:keys [pct] :as job}]
     (let [sum (apply + (map :pct all))]
       (if (<= (+ sum pct) 100)
         (conj all job)
         (reduced all))))
   []
   jobs))

(defmethod cjs/job-offer-n-peers :onyx.job-scheduler/percentage
  [replica]
  (let [n-peers (count (:peers replica))
        sorted-jobs (sort-jobs-by-pct replica)
        jobs-to-use (drop-jobs-overflow sorted-jobs)
        init-allocations (min-allocations jobs-to-use n-peers)]
    (into {} (map (fn [j] {(:job j) (:capacity j)}) init-allocations))))

(defn job-peer-count [replica job]
  (apply + (map count (vals (get-in replica [:allocations job])))))

(defmethod cjs/sort-job-priority :onyx.job-scheduler/percentage
  [replica jobs]
  (sort-by (juxt #(.indexOf ^clojure.lang.PersistentVector (vec (:jobs replica)) %)
                 (partial job-peer-count replica))
{'fraction_non_alphanumeric': 0.16724518775639002, 'fraction_numerical': 0.003786683496371095, 'mean_word_length': 2.7648456057007125, 'pattern_counts': {'":': 0, '<': 5, '<?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': '668766', 'n_tokens_mistral': 1212, 'n_tokens_neox': 1142, 'n_words': 284}