0 | ||| Resource Pool Internals
  1 | module Data.Pool.Internal
  2 |
  3 | import Data.Array.Core
  4 | import Data.Linear.Ref1
  5 | import Data.Nat
  6 | import Data.So
  7 | import Data.SortedSet
  8 | import System.Concurrency
  9 | import System.Posix.Timer
 10 | import System.Posix.Timer.Prim
 11 |
 12 | %default total
 13 |
 14 | ||| Configuration of a Pool.
 15 | |||
 16 | ||| Constraints:
 17 | ||| - poolmaxresources -> The smallest acceptable value is 1.
 18 | ||| - poolnumstripes -> The smallest acceptable value is 1, poolnumstripes must not be larger than poolmaxresources.
 19 | |||
 20 | public export
 21 | record PoolConfig a where
 22 |   constructor MkPoolConfig
 23 |   createresource   : IO a
 24 |   freeresource     : a -> IO ()
 25 |   poolcachettl     : Clock Duration
 26 |   poolmaxresources : (maxres ** LTE 1 maxres)
 27 |   poolnumstripes   : (n ** (LTE 1 n, LTE n (fst poolmaxresources)))
 28 |   poolconfiglabel  : String
 29 |
 30 | ||| A simple (persistent) FIFO queue.
 31 | |||
 32 | ||| This is used to maintain an ordered collection of waiting threads.
 33 | ||| Elements are appended at the tail and removed from the head.
 34 | |||
 35 | ||| Notes:
 36 | ||| - This representation has O(n) append.
 37 | ||| - Under contention (with CAS updates), appends may be retried, so this structure favors simplicity over performance.
 38 | ||| - It is typically used together with a secondary "reversed" queue to amortize costs (two-list queue pattern).
 39 | |||
 40 | public export
 41 | data Queue a
 42 |   = QNode a (Queue a)
 43 |   | QEnd
 44 |
 45 | ||| Result of waking a waiting thread.
 46 | |||
 47 | ||| This represents the outcome delivered to a blocked waiter through its wake channel.
 48 | |||
 49 | ||| Variants:
 50 | ||| - `Deliver a`
 51 | |||  - A resource was directly handed off to the waiter.
 52 | |||
 53 | ||| - `Create`
 54 | |||  - No reusable resource was available, but the waiter should proceed by creating a fresh resource using an already-reserved capacity slot.
 55 | |||
 56 | ||| - `Cancelled`
 57 | |||  - The waiter was cancelled before receiving a resource.
 58 | |||  - This is used to distinguish cancellation from normal wakeup semantics.
 59 | |||
 60 | ||| Design Notes:
 61 | ||| - This replaces the older `Maybe a` wake protocol, which overloaded `Nothing` to represent multiple meanings.
 62 | ||| - Explicit wake states improve clarity and correctness of the Stripe state machine.
 63 | |||
 64 | ||| Guarantees:
 65 | ||| - Each waiter receives at most one `WakeResult`.
 66 | ||| - Wake results correspond only to committed Stripe transitions.
 67 | ||| - No wake result is delivered more than once.
 68 | |||
 69 | ||| Invariants:
 70 | ||| - `Deliver a` carries ownership transfer of exactly one resource.
 71 | ||| - `Create` implies capacity has already been reserved.
 72 | ||| - `Cancelled` does not transfer ownership of a resource.
 73 | |||
 74 | public export
 75 | data WakeResult a
 76 |   = Deliver a
 77 |   | Create
 78 |   | Cancelled
 79 |
 80 | ||| A pure waiting token representing a blocked thread.
 81 | |||
 82 | ||| This contains no mutable state. All lifecycle tracking is handled
 83 | ||| by the Stripe during dequeue / cancellation.
 84 | |||
 85 | ||| Fields:
 86 | ||| - `id`   : unique identifier for cancellation tracking
 87 | ||| - `wake` : channel used to unblock the thread
 88 | |||
 89 | ||| Invariants:
 90 | ||| - Waiter is immutable
 91 | ||| - Wake is single-use
 92 | ||| - Cancellation is handled by Stripe (not locally)
 93 | |||
 94 | public export
 95 | data Waiter : (a : Type) -> Type where
 96 |   MkWaiter :  (id   : Nat)
 97 |            -> (wake : Channel (WakeResult a))
 98 |            -> Waiter a
 99 |
100 | ||| An existing resource currently sitting in a pool.
101 | |||
102 | public export
103 | data Entry : (a : Type) -> Type where
104 |   MkEntry :  (entry    : a)
105 |           -> (lastused : IClock CLOCK_MONOTONIC)
106 |           -> Entry a
107 |
108 | ||| A Stripe error.
109 | |||
110 | ||| Fields:
111 | ||| - 'id'                    : The identifier of the Stripe the error originated from
112 | ||| - `fnname`                : The function the error originated from
113 | ||| - `errormessage`          : The error formatted as a String
114 | ||| - `errormessagetimestamp` : The timestamp the error occurred at
115 | |||
116 | public export
117 | record StripeError where
118 |   constructor MkStripeError
119 |   id                    : Nat
120 |   fnname                : String
121 |   errormessage          : String
122 |   errormessagetimestamp : Maybe (IClock CLOCK_REALTIME)
123 |
124 | public export
125 | Show StripeError where
126 |   show (MkStripeError id fnname errormessage (Just errormessagetimestamp)) =
127 |     "MkStripeError " ++
128 |     (show id)        ++
129 |     " "              ++
130 |     fnname           ++
131 |     " "              ++
132 |     errormessage     ++
133 |     " "              ++
134 |     (asctime $ fromUTC errormessagetimestamp)
135 |   show (MkStripeError id fnname errormessage Nothing)                      =
136 |     "MkStripeError " ++
137 |     (show id)        ++
138 |     " "              ++
139 |     fnname           ++
140 |     " "              ++
141 |     errormessage     ++
142 |     " "              ++
143 |     "IClock CLOCK_REALTIME"
144 |
145 | ||| Stripe is the only concurrent state machine in the system.
146 | |||
147 | ||| It owns:
148 | ||| - Resource availability.
149 | ||| - Cached resources.
150 | ||| - All waiting threads.
151 | ||| - Cancellation tracking.
152 | |||
153 | ||| All mutations occur via CAS on an enclosing Ref, `Stripe1 s a`.
154 | |||
155 | ||| Fields:
156 | ||| - `available` : number of available resources
157 | ||| - `cache`     : reusable resources
158 | ||| - `queue`     : primary FIFO of waiters
159 | ||| - `queuer`    : secondary FIFO (amortized append)
160 | ||| - `nextId`    : fresh waiter id supply
161 | ||| - `cancelled` : sorted set of cancelled waiter ids
162 | |||
163 | ||| Invariants:
164 | ||| - Stripe is immutable between CAS updates.
165 | ||| - Queue ordering is authoritative.
166 | ||| - Cancelled waiters are lazily skipped.
167 | |||
168 | public export
169 | data Stripe : (a : Type) -> Type where
170 |   MkStripe :  (available : Nat)
171 |            -> (cache     : List (Entry a))
172 |            -> (queue     : Queue (Waiter a))
173 |            -> (queuer    : Queue (Waiter a))
174 |            -> (nextid    : Nat)
175 |            -> (cancelled : SortedSet Nat)
176 |            -> Stripe a
177 |
178 | ||| A linear mutable stripe.
179 | |||
180 | public export
181 | data Stripe1 : (s : Type) -> (a : Type) -> Type where
182 |   MkStripe1 :  Ref s (Stripe a)
183 |             -> Stripe1 s a
184 |
185 | ||| Effects emitted by a Stripe transition.
186 | |||
187 | ||| These are collected during CAS computation and executed
188 | ||| only after a successful CAS commit.
189 | |||
190 | ||| Guarantees:
191 | ||| - No IO inside CAS.
192 | ||| - No duplicated effects on retry.
193 | ||| - Deterministic state transitions.
194 | |||
195 | public export
196 | data StripeEffect a
197 |   = Wake (Channel (WakeResult a)) (WakeResult a)
198 |   | WakeMany (List (Channel (WakeResult a), WakeResult a))
199 |   | InsertWithTimestamp a
200 |   | FreeMany (a -> IO ()) (List a)
201 |   | None
202 |
203 | ||| Result of a Stripe state transition.
204 | |||
205 | ||| Represents:
206 | ||| - The new Stripe state (to be CAS'd).
207 | ||| - Effects to run AFTER successful commit.
208 | |||
209 | ||| Invariants:
210 | ||| - This must be treated as a one-shot value.
211 | ||| - If CAS fails, this MUST be discarded.
212 | ||| - Effects must NEVER be run unless CAS succeeds.
213 | |||
214 | public export
215 | record StripeStep a where
216 |   constructor MkStripeStep
217 |   stripe  : Stripe a
218 |   effects : List (StripeEffect a)
219 |
220 | ||| A single, local pool based on a linear mutable stripe.
221 | |||
222 | public export
223 | data LocalPool1 : (s : Type) -> (a : Type) -> Type where
224 |   MkLocalPool1 :  (stripeid  : Nat)
225 |                -> (stripevar : Stripe1 s a)
226 |                -> LocalPool1 s a
227 |
228 | ||| A Pool1 error.
229 | |||
230 | ||| Fields:
231 | ||| - `fnname`                : The function the error originated from
232 | ||| - `errormessage`          : The error formatted as a String
233 | ||| - `errormessagetimestamp` : The timestamp the error occurred at
234 | |||
235 | public export
236 | record Pool1Error where
237 |   constructor MkPool1Error
238 |   fnname                : String
239 |   errormessage          : String
240 |   errormessagetimestamp : Maybe (IClock CLOCK_REALTIME)
241 |
242 | public export
243 | Show Pool1Error where
244 |   show (MkPool1Error fnname errormessage (Just errormessagetimestamp)) =
245 |     "MkPool1Error " ++
246 |     fnname          ++
247 |     " "             ++
248 |     errormessage    ++
249 |     " "             ++
250 |     (asctime $ fromUTC errormessagetimestamp)
251 |   show (MkPool1Error fnname errormessage Nothing)                      =
252 |     "MkPool1Error " ++
253 |     fnname          ++
254 |     " "             ++
255 |     errormessage    ++
256 |     " "             ++
257 |     "IClock CLOCK_REALTIME"
258 |
259 | ||| Striped resource pool based on linear mutable references.
260 | |||
261 | public export
262 | data Pool1 : (s : Type) -> (n : Nat) -> (a : Type) -> Type where
263 |   MkPool1 :  (poolconfig : PoolConfig a)
264 |           -> (localpools : (MArray s n (LocalPool1 s a)))
265 |           -> Pool1 s n a
266 |