/
parallel_test.clj
205 lines (173 loc) · 7.45 KB
/
parallel_test.clj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
(ns tech.parallel-test
(:require [clojure.test :refer :all]
[tech.parallel :as parallel]
[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 don'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 isn'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 doesn'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 doesn'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)))