(ns tech.parallel-test
  (:require [clojure.test :refer :all]
            [tech.parallel :as parallel]
            [tech.parallel.for :as parallel-for]
            [clojure.tools.logging :as log]))


(deftest order-indexed-sequence
  (is (= (vec (range 1000))
         (vec
          (parallel/order-indexed-sequence identity
                                           (shuffle (range 1000)))))))


(deftest ordered-queued-sequence
  (let [result (:sequence (parallel/queued-sequence inc [(range 1000)]
                                                    :ordered? true))]
    (is (= (count (distinct result)) 1000))
    (is (= (vec result) (vec (range 1 1001))))))


(deftest queued-sequence-with-exception
  (let [result (:sequence (parallel/queued-sequence (fn [idx]
                                                      (when (= idx 100)
                                                        (throw (Exception. "Failure!!")))
                                                      (inc idx))
                                                    [(range 1000)]))]
    (is (thrown? RuntimeException (vec result)))))


(deftest queue-pmap-no-queue
  (let [result (vec (parallel/queued-pmap 0 inc (range 100)))]
    (is (= result (vec (range 1 101))))))


(deftest ordered-queued-sequence-items-in-flight
  (let [items-in-flight (atom 0)
        queue-depth 10
        input-sequence (range 1000)
        proc-fn (fn [item]
                  (swap! items-in-flight inc)
Add 1 to queue depth because we do n't decrement until dereference
                  (when-not (<= @items-in-flight (+ queue-depth 1))
                    (throw (ex-info (format "Items in flight invariant borked: %s"
                                            @items-in-flight)
                                    {:queue-depth queue-depth
                                     :items-in-flight @items-in-flight
                                     :item item})))
                  (inc item))
        result (map (fn [item]
                      (swap! items-in-flight dec)
                      item)
                    (:sequence (parallel/queued-sequence proc-fn [input-sequence]
                                                         :queue-depth 10
                                                         :ordered? true)))]
    (is (= (vec result) (vec (range 1 1001))))))


(deftest ordered-queue-sequence-items-in-flight-2
  (let [queue-depth 5
Infinite sequence of double arrays but only queue - depth + 1 repeating
        data-arrays (->> (repeatedly (+ queue-depth 1) #(double-array 1))
                         (repeat)
                         (mapcat identity))
        proc-fn (fn [idx ^doubles data-array]
                  (aset data-array 0 (double idx))
                  data-array)
if the invariant is n't met then we will get duplicates where one of the threads
        ;;overwrites a data array it should not.
        result (map (fn [^doubles data-array]
                      (long (aget data-array 0)))
                    (:sequence (parallel/queued-sequence proc-fn [(range 1000) data-arrays]
                                                         :queue-depth queue-depth
                                                         :ordered? true)))]
        (is (= (vec result) (vec (range 1000))))))


(deftest basic-parallel-for
  (let [result-ary (int-array 1000)]
    (parallel/parallel-for idx 1000
                           (aset result-ary idx (inc idx)))
    (is (= (count (distinct (vec result-ary))) 1000))))

(defmacro or=
  "Is a equal to b or c"
  [a b c]
  `(let [a# ~a]
     (or (= a# ~b)
         (= a# ~c))))

(deftest buffered-seq
  (let [realized-count (atom 0)
        test-seq (->> (parallel/queued-pmap
                       1
                       (fn [idx]
                         (swap! realized-count inc)
                         idx)
                       (range 100))
                      (parallel/buffered-seq 5)
                      (map identity))]
If the pmap thread has launched then this number is 1 .
    ;;If it hasn't yet then this number is 0.  We haven't blocked
on it yet so it could be legitimately be either 1 or 0 .
    ;;This is true for any of the realized counts and it is important
    ;;to understand this ambiguity.
    (is (or= @realized-count 0 1))
    (is (= 0 (first test-seq)))
    (is (or= @realized-count 5 6))
    (is (= 1 (second test-seq)))
    (is (or= @realized-count 6 7))))


(def ^:dynamic *increment-count* 0)

(deftest thread-bindings
  (with-bindings {#'*increment-count* 1}
    (let [increment-atom (atom 0)]
      (->> (range 100)
           (parallel/queued-pmap 10 (fn [_]
                                      (swap! increment-atom (fn [val]
                                                              (+ val *increment-count*)))))
           dorun)
      (is (= @increment-atom 100)))))


(deftest precise-queue-behavior
  (let [access-count (atom 0)
        in-flight-max 2
        check-in-flight-max (fn []
                              (when (> @access-count in-flight-max)
                                (throw (ex-info "In flight max constraint violation"
                                                {:max-val @access-count}))))]
    (->> (range 1000)
         (parallel/queued-pmap (- in-flight-max 1)
                               (fn [_]
                                 (swap! access-count inc)
                                 (Thread/sleep 10)
                                 (check-in-flight-max)))
         (map (fn [_]
                (Thread/sleep 10)
                (check-in-flight-max)
                (swap! access-count dec)))
         vec)
    (is (= @access-count 0))))


(deftest short-sequence-clean-shutdown
  ;;A very short sequence can result in an error if there are more threads than potential processors and
the launch sequence does n't finish before one of the threads shuts down the sequence .
  (let [error
        (try
          (dotimes [iter 10]
            (->> (parallel/queued-pmap 100 identity [1 2 3])
                 doall)
            nil)
          (catch Throwable e e))]
    (is (nil? error))))


(deftest short-sequence-error-clean-shutdown
  ;;A very short sequence can result in an error if there are more threads than
potential processors and the launch sequence does n't finish before one of the
  ;;threads causes an exception.

  (dotimes [iter 20]
    (try
      (->> (parallel/queued-pmap 100 (fn [& args]
                                       (throw (ex-info "Sequence failure" {})))
                                 [1 2 3])
          doall)
      (throw (ex-info "should not make it here"))
      (catch Throwable e
        (is (not (instance? java.util.concurrent.RejectedExecutionException e)))
        nil))))


(deftest pskink-test
  (log/info "Logged exception expected!!")
  (let [test-atom (atom 0)]
    (parallel/psink!
     #(if (= 5 %)
        (throw (ex-info "hey" {}))
        (do
          (swap! test-atom inc)
          (Thread/sleep (rand-int 200))))
     (range 100))
    (is (= 99 @test-atom))))


(deftest parallel-memoize
  (let [test-atom (atom 0)]
    (dotimes [iter 100]
      (let [inc-fn (parallel/memoize #(swap! test-atom inc))]
        (->> (range 20)
             (pmap (fn [ignored]
                     (inc-fn))))))
    (is (= @test-atom 100))))


(deftest require-resolve
  (is (not (find-ns 'clojure.inspector)))
  (is (every? (complement nil?)
              (->> (range 100)
                   (pmap (fn [idx]
                           (parallel/require-resolve 'clojure.inspector/inspect))))))
  (is (find-ns 'clojure.inspector)))


(deftest indexed-map-reduce
  (let [sum
        (parallel-for/indexed-map-reduce
         1000
         (fn [idx n-args]
           n-args)
         (partial apply +))]
    (is (= sum 1000))))

overwrites a data array it should not.
If it hasn't yet then this number is 0.  We haven't blocked
This is true for any of the realized counts and it is important
to understand this ambiguity.
A very short sequence can result in an error if there are more threads than potential processors and
A very short sequence can result in an error if there are more threads than
threads causes an exception.
(ns tech.parallel-test
  (:require [clojure.test :refer :all]
            [tech.parallel :as parallel]
            [tech.parallel.for :as parallel-for]
            [clojure.tools.logging :as log]))


(deftest order-indexed-sequence
  (is (= (vec (range 1000))
         (vec
          (parallel/order-indexed-sequence identity
                                           (shuffle (range 1000)))))))


(deftest ordered-queued-sequence
  (let [result (:sequence (parallel/queued-sequence inc [(range 1000)]
                                                    :ordered? true))]
    (is (= (count (distinct result)) 1000))
    (is (= (vec result) (vec (range 1 1001))))))


(deftest queued-sequence-with-exception
  (let [result (:sequence (parallel/queued-sequence (fn [idx]
                                                      (when (= idx 100)
                                                        (throw (Exception. "Failure!!")))
                                                      (inc idx))
                                                    [(range 1000)]))]
    (is (thrown? RuntimeException (vec result)))))


(deftest queue-pmap-no-queue
  (let [result (vec (parallel/queued-pmap 0 inc (range 100)))]
    (is (= result (vec (range 1 101))))))


(deftest ordered-queued-sequence-items-in-flight
  (let [items-in-flight (atom 0)
        queue-depth 10
        input-sequence (range 1000)
        proc-fn (fn [item]
                  (swap! items-in-flight inc)
Add 1 to queue depth because we do n't decrement until dereference
                  (when-not (<= @items-in-flight (+ queue-depth 1))
                    (throw (ex-info (format "Items in flight invariant borked: %s"
                                            @items-in-flight)
                                    {:queue-depth queue-depth
                                     :items-in-flight @items-in-flight
                                     :item item})))
                  (inc item))
        result (map (fn [item]
                      (swap! items-in-flight dec)
                      item)
                    (:sequence (parallel/queued-sequence proc-fn [input-sequence]
                                                         :queue-depth 10
                                                         :ordered? true)))]
    (is (= (vec result) (vec (range 1 1001))))))


(deftest ordered-queue-sequence-items-in-flight-2
  (let [queue-depth 5
Infinite sequence of double arrays but only queue - depth + 1 repeating
        data-arrays (->> (repeatedly (+ queue-depth 1) #(double-array 1))
                         (repeat)
                         (mapcat identity))
        proc-fn (fn [idx ^doubles data-array]
                  (aset data-array 0 (double idx))
                  data-array)
if the invariant is n't met then we will get duplicates where one of the threads
        result (map (fn [^doubles data-array]
                      (long (aget data-array 0)))
                    (:sequence (parallel/queued-sequence proc-fn [(range 1000) data-arrays]
                                                         :queue-depth queue-depth
                                                         :ordered? true)))]
        (is (= (vec result) (vec (range 1000))))))


(deftest basic-parallel-for
  (let [result-ary (int-array 1000)]
    (parallel/parallel-for idx 1000
                           (aset result-ary idx (inc idx)))
    (is (= (count (distinct (vec result-ary))) 1000))))

(defmacro or=
  "Is a equal to b or c"
  [a b c]
  `(let [a# ~a]
     (or (= a# ~b)
         (= a# ~c))))

(deftest buffered-seq
  (let [realized-count (atom 0)
        test-seq (->> (parallel/queued-pmap
                       1
                       (fn [idx]
                         (swap! realized-count inc)
                         idx)
                       (range 100))
                      (parallel/buffered-seq 5)
                      (map identity))]
If the pmap thread has launched then this number is 1 .
on it yet so it could be legitimately be either 1 or 0 .
    (is (or= @realized-count 0 1))
    (is (= 0 (first test-seq)))
    (is (or= @realized-count 5 6))
    (is (= 1 (second test-seq)))
    (is (or= @realized-count 6 7))))


(def ^:dynamic *increment-count* 0)

(deftest thread-bindings
  (with-bindings {#'*increment-count* 1}
    (let [increment-atom (atom 0)]
      (->> (range 100)
           (parallel/queued-pmap 10 (fn [_]
                                      (swap! increment-atom (fn [val]
                                                              (+ val *increment-count*)))))
           dorun)
      (is (= @increment-atom 100)))))


(deftest precise-queue-behavior
  (let [access-count (atom 0)
        in-flight-max 2
        check-in-flight-max (fn []
                              (when (> @access-count in-flight-max)
                                (throw (ex-info "In flight max constraint violation"
                                                {:max-val @access-count}))))]
    (->> (range 1000)
         (parallel/queued-pmap (- in-flight-max 1)
                               (fn [_]
                                 (swap! access-count inc)
                                 (Thread/sleep 10)
                                 (check-in-flight-max)))
         (map (fn [_]
                (Thread/sleep 10)
                (check-in-flight-max)
                (swap! access-count dec)))
         vec)
    (is (= @access-count 0))))


(deftest short-sequence-clean-shutdown
the launch sequence does n't finish before one of the threads shuts down the sequence .
  (let [error
        (try
          (dotimes [iter 10]
            (->> (parallel/queued-pmap 100 identity [1 2 3])
                 doall)
            nil)
          (catch Throwable e e))]
    (is (nil? error))))


(deftest short-sequence-error-clean-shutdown
potential processors and the launch sequence does n't finish before one of the

  (dotimes [iter 20]
    (try
      (->> (parallel/queued-pmap 100 (fn [& args]
                                       (throw (ex-info "Sequence failure" {})))
                                 [1 2 3])
          doall)
      (throw (ex-info "should not make it here"))
      (catch Throwable e
        (is (not (instance? java.util.concurrent.RejectedExecutionException e)))
        nil))))


(deftest pskink-test
  (log/info "Logged exception expected!!")
  (let [test-atom (atom 0)]
    (parallel/psink!
     #(if (= 5 %)
        (throw (ex-info "hey" {}))
        (do
          (swap! test-atom inc)
          (Thread/sleep (rand-int 200))))
     (range 100))
    (is (= 99 @test-atom))))


(deftest parallel-memoize
  (let [test-atom (atom 0)]
    (dotimes [iter 100]
      (let [inc-fn (parallel/memoize #(swap! test-atom inc))]
        (->> (range 20)
             (pmap (fn [ignored]
                     (inc-fn))))))
    (is (= @test-atom 100))))


(deftest require-resolve
  (is (not (find-ns 'clojure.inspector)))
  (is (every? (complement nil?)
              (->> (range 100)
                   (pmap (fn [idx]
                           (parallel/require-resolve 'clojure.inspector/inspect))))))
  (is (find-ns 'clojure.inspector)))


(deftest indexed-map-reduce
  (let [sum
        (parallel-for/indexed-map-reduce
         1000
         (fn [idx n-args]
           n-args)
         (partial apply +))]
    (is (= sum 1000))))
