blob: 369bc8507dff8945dd925343ecbbd6173fbe1518 (
plain)
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
|
(.module:
[lux #*
[abstract
["." monad (#+ do)]]
[control
["." io]
[concurrency
["/" semaphore]
["." promise (#+ Promise)]
["." atom (#+ Atom)]]]
[data
["." maybe]
["." text ("#;." equivalence monoid)
format]
[collection
["." list ("#;." functor)]]]
[math
["r" random]]]
lux/test)
## (def: (wait-many-times times semaphore)
## (-> Nat /.Semaphore (Promise Any))
## (loop [steps times]
## (if (n/> 0 steps)
## (do promise.monad
## [_ (/.wait semaphore)]
## (recur (dec steps)))
## (:: promise.monad wrap []))))
## (context: "Semaphore."
## (<| (times 100)
## (do @
## [open-positions (|> r.nat (:: @ map (|>> (n/% 10) (n/max 1))))]
## ($_ seq
## (let [semaphore (/.semaphore open-positions)]
## (wrap (do promise.monad
## [_ (wait-many-times open-positions semaphore)]
## (assert "Can wait on a semaphore up to the number of open positions without blocking."
## true))))
## (let [semaphore (/.semaphore open-positions)]
## (wrap (do promise.monad
## [result (<| (promise.time-out 100)
## (wait-many-times (inc open-positions) semaphore))]
## (assert "Waiting on a semaphore more than the number of open positions blocks the process."
## (case result
## (#.Some _)
## false
## #.None
## true)))))
## (let [semaphore (/.semaphore open-positions)]
## (wrap (do promise.monad
## [_ (: (Promise Any)
## (loop [steps (n/* 2 open-positions)]
## (if (n/> 0 steps)
## (do @
## [_ (/.wait semaphore)
## _ (/.signal semaphore)]
## (recur (dec steps)))
## (wrap []))))]
## (assert "Signaling a semaphore replenishes its open positions."
## true))))
## (let [semaphore (/.semaphore open-positions)]
## (wrap (do promise.monad
## [#let [resource (atom.atom "")
## blocked (do @
## [_ (wait-many-times open-positions semaphore)
## _ (/.wait semaphore)
## #let [_ (io.run (atom.update (|>> (format "B"))
## resource))]]
## (wrap []))]
## _ (promise.wait 100)
## _ (exec (io.run (atom.update (|>> (format "A"))
## resource))
## (/.signal semaphore))
## _ blocked]
## (assert "A blocked process can be un-blocked by a signal somewhere else."
## (text;= "BA"
## (io.run (atom.read resource)))))))
## ))))
## (context: "Mutex."
## (<| (times 100)
## (do @
## [repetitions (|> r.nat (:: @ map (|>> (n/% 100) (n/max 10))))]
## ($_ seq
## (let [mutex (/.mutex [])]
## (wrap (do promise.monad
## [#let [resource (atom.atom "")
## expected-As (text.join-with "" (list.repeat repetitions "A"))
## expected-Bs (text.join-with "" (list.repeat repetitions "B"))
## processA (<| (/.synchronize mutex)
## io.io
## promise.future
## (do io.monad
## [_ (<| (monad.seq @)
## (list.repeat repetitions)
## (atom.update (|>> (format "A")) resource))]
## (wrap [])))
## processB (<| (/.synchronize mutex)
## io.io
## promise.future
## (do io.monad
## [_ (<| (monad.seq @)
## (list.repeat repetitions)
## (atom.update (|>> (format "B")) resource))]
## (wrap [])))]
## _ processA
## _ processB
## #let [outcome (io.run (atom.read resource))]]
## (assert "Mutexes only allow one process to execute at a time."
## (or (text;= (format expected-As expected-Bs)
## outcome)
## (text;= (format expected-Bs expected-As)
## outcome))))))
## ))))
## (def: (waiter resource barrier id)
## (-> (Atom Text) /.Barrier Nat (Promise Any))
## (do promise.monad
## [_ (/.block barrier)
## #let [_ (io.run (atom.update (|>> (format (%n id))) resource))]]
## (wrap [])))
## (context: "Barrier."
## (let [limit 10
## barrier (/.barrier (maybe.assume (/.limit limit)))
## resource (atom.atom "")]
## ($_ seq
## (wrap (do promise.monad
## [#let [ids (list.n/range 0 (dec limit))
## waiters (list;map (function (_ id)
## (let [process (waiter resource barrier id)]
## (exec (io.run (atom.update (|>> (format "_")) resource))
## process)))
## ids)]
## _ (monad.seq @ waiters)
## #let [outcome (io.run (atom.read resource))]]
## (assert "A barrier can stop all processes from acting, until an amount of waiting processes is reached, and then the barrier is un-blocked for all."
## (and (text.ends-with? "__________" outcome)
## (list.every? (function (_ id)
## (text.contains? (%n id) outcome))
## ids)
## )))))))
|
