0 | ||| Log-Structured Merge RRB Vector (LSMRRBVector)
  1 | module Data.LSMRRBVector
  2 |
  3 | import public Data.LSMRRBVector.Internal
  4 | import public Data.LSMRRBVector.Types
  5 | import public Data.RRBVector
  6 |
  7 | import Control.Monad.Elin
  8 | import Control.Monad.MCancel
  9 | import Control.Monad.ST
 10 | import Data.Array
 11 | import Data.Array.Core
 12 | import Data.Array.Index
 13 | import Data.Array.Indexed
 14 | import Data.Bits
 15 | import Data.Linear.Ref1
 16 | import Data.List
 17 | import Data.List1
 18 | import Data.Maybe
 19 | import Data.RRBVector
 20 | import Data.SortedMap
 21 | import Data.SnocList
 22 | import Data.Vect
 23 | import Data.Zippable
 24 | import IO.Async
 25 | import IO.Async.Core
 26 | import IO.Async.Loop.Poller
 27 | import IO.Async.Loop.Posix
 28 | import IO.Async.Posix
 29 | import IO.Async.Service
 30 | import Syntax.T1 as T1
 31 | import System.Concurrency
 32 | import System.Posix.Timer
 33 | import System.Posix.Timer.Prim
 34 |
 35 | %hide Control.Monad.Elin.Elin.(.run)
 36 | %hide Control.Monad.Elin.Elin.run
 37 | %hide Prelude.null
 38 | %hide Prelude.Ops.infixr.(<|)
 39 | %hide Prelude.Ops.infixl.(|>)
 40 |
 41 | %default total
 42 |
 43 | --------------------------------------------------------------------------------
 44 | --          Mutation Operations
 45 | --------------------------------------------------------------------------------
 46 |
 47 | ||| Appends a value onto the logical end of the vector.
 48 | |||
 49 | ||| Effect:
 50 | ||| - Adds an Append operation to the thread-local buffer.
 51 | |||
 52 | export
 53 | append :  LSMRRBVector World a
 54 |        -> RebuildService Poll
 55 |        -> ThreadId
 56 |        -> a
 57 |        -> Async Poll [Errno] ()
 58 | append lsmrrbvector svc tid x = do
 59 |   shouldtrigger <- liftIO (enqueueOperation lsmrrbvector.buffers lsmrrbvector.combinedsnapshotstate tid (Append x))
 60 |   scheduleIfNeeded lsmrrbvector svc shouldtrigger
 61 |   
 62 | ||| Prepends a value onto the logical beginning of the vector.
 63 | |||
 64 | ||| Effect:
 65 | ||| - Adds a Prepend operation to the thread-local buffer.
 66 | |||
 67 | export
 68 | prepend :  LSMRRBVector World a
 69 |         -> RebuildService Poll
 70 |         -> ThreadId
 71 |         -> a
 72 |         -> Async Poll [Errno] ()
 73 | prepend lsmrrbvector svc tid x = do
 74 |   shouldtrigger <- liftIO (enqueueOperation lsmrrbvector.buffers lsmrrbvector.combinedsnapshotstate tid (Prepend x))
 75 |   scheduleIfNeeded lsmrrbvector svc shouldtrigger
 76 |
 77 | ||| Inserts a value at a specified logical index.
 78 | |||
 79 | ||| Effect:
 80 | ||| - Adds an Insert operation to the thread-local buffer.
 81 | |||
 82 | export
 83 | insert :  LSMRRBVector World a
 84 |        -> RebuildService Poll
 85 |        -> ThreadId
 86 |        -> Nat
 87 |        -> a
 88 |        -> Async Poll [Errno] ()
 89 | insert lsmrrbvector svc tid i x = do
 90 |   shouldtrigger <- liftIO (enqueueOperation lsmrrbvector.buffers lsmrrbvector.combinedsnapshotstate tid (Insert i x))
 91 |   scheduleIfNeeded lsmrrbvector svc shouldtrigger
 92 |
 93 | ||| Removes a value at a specified logical index.
 94 | |||
 95 | ||| Effect:
 96 | ||| - Adds a Delete operation to the thread-local buffer.
 97 | |||
 98 | export
 99 | delete :  LSMRRBVector World a
100 |        -> RebuildService Poll
101 |        -> ThreadId
102 |        -> Nat
103 |        -> Async Poll [Errno] ()
104 | delete lsmrrbvector svc tid i = do
105 |   shouldtrigger <- liftIO (enqueueOperation lsmrrbvector.buffers lsmrrbvector.combinedsnapshotstate tid (Delete i))
106 |   scheduleIfNeeded lsmrrbvector svc shouldtrigger
107 |
108 | ||| Replaces a value at a specified logical index.
109 | |||
110 | ||| Effect:
111 | ||| - Adds an Update operation to the thread-local buffer.
112 | |||
113 | export
114 | update :  LSMRRBVector World a
115 |        -> RebuildService Poll
116 |        -> ThreadId
117 |        -> Nat
118 |        -> a
119 |        -> Async Poll [Errno] ()
120 | update lsmrrbvector svc tid i x = do
121 |   shouldtrigger <- liftIO (enqueueOperation lsmrrbvector.buffers lsmrrbvector.combinedsnapshotstate tid (Update i x))
122 |   scheduleIfNeeded lsmrrbvector svc shouldtrigger
123 |
124 | --------------------------------------------------------------------------------
125 | --          Read Operations
126 | --------------------------------------------------------------------------------
127 |
128 | ||| Converts the current published snapshot into a list.
129 | |||
130 | ||| Behavior:
131 | ||| - Reads the current immutable snapshot.
132 | ||| - Converts the snapshot contents into a List.
133 | |||
134 | ||| Properties:
135 | ||| - Observes a consistent snapshot.
136 | ||| - Does not block writers or rebuild activity.
137 | ||| - Reader participation is cleaned up automatically.
138 | |||
139 | ||| Notes:
140 | ||| - Concurrent writes published after acquisition are not visible.
141 | |||
142 | ||| Complexity:
143 | ||| - Snapshot acquisition: O(1)
144 | ||| - Conversion: O(n)
145 | |||
146 | export
147 | toList :  LSMRRBVector World a
148 |        -> ThreadId
149 |        -> IO (List a)
150 | toList lsmrrbvector tid =
151 |   readSnapshotWithGeneration lsmrrbvector tid (\(_, v) => Data.RRBVector.toList v)
152 |
153 | ||| Returns the number of elements in the current published snapshot.
154 | |||
155 | ||| Behavior:
156 | ||| - Reads the current immutable snapshot.
157 | ||| - Returns its logical length.
158 | |||
159 | ||| Properties:
160 | ||| - Observes a consistent snapshot.
161 | ||| - Does not block writers or rebuild activity.
162 | ||| - Reader participation is cleaned up automatically.
163 | |||
164 | ||| Notes:
165 | ||| - Concurrent writes published after acquisition are not visible.
166 | |||
167 | ||| Complexity:
168 | ||| - O(1)
169 | |||
170 | export
171 | length :  LSMRRBVector World a
172 |        -> ThreadId
173 |        -> IO Nat
174 | length lsmrrbvector tid =
175 |   readSnapshotWithGeneration lsmrrbvector tid (\(_, v) => Data.RRBVector.length v)
176 |
177 | ||| Looks up an element by index.
178 | |||
179 | ||| Behavior:
180 | ||| - Reads the current immutable snapshot.
181 | ||| - Returns Nothing if the index is out of bounds.
182 | |||
183 | ||| Properties:
184 | ||| - Observes a consistent snapshot.
185 | ||| - Does not block writers or rebuild activity.
186 | ||| - Reader participation is cleaned up automatically.
187 | |||
188 | ||| Notes:
189 | ||| - Concurrent writes published after acquisition are not visible.
190 | |||
191 | ||| Complexity:
192 | ||| - O(log n)
193 | |||
194 | export
195 | lookup :  LSMRRBVector World a
196 |        -> ThreadId
197 |        -> Nat
198 |        -> IO (Maybe a)
199 | lookup lsmrrbvector tid i =
200 |   readSnapshotWithGeneration lsmrrbvector tid (\(_, v) => Data.RRBVector.lookup i v)
201 |
202 | ||| Tests whether the current published snapshot is empty.
203 | |||
204 | ||| Behavior:
205 | ||| - Reads the current immutable snapshot.
206 | ||| - Returns True when no elements exist.
207 | |||
208 | ||| Properties:
209 | ||| - Observes a consistent snapshot.
210 | ||| - Does not block writers or rebuild activity.
211 | ||| - Reader participation is cleaned up automatically.
212 | |||
213 | ||| Notes:
214 | ||| - Concurrent writes published after acquisition are not visible.
215 | |||
216 | ||| Complexity:
217 | ||| - O(1)
218 | |||
219 | export
220 | null :  LSMRRBVector World a
221 |      -> ThreadId
222 |      -> IO Bool
223 | null lsmrrbvector tid =
224 |   readSnapshotWithGeneration lsmrrbvector tid (\(_, v) => Data.RRBVector.null v)
225 |
226 | --------------------------------------------------------------------------------
227 | --          Default Config
228 | --------------------------------------------------------------------------------
229 |
230 | ||| Default log-structured merge vector configuration.
231 | |||
232 | ||| Current defaults favor balanced throughput and latency.
233 | |||
234 | export
235 | defaultconfig : LSMRRBVectorConfig
236 | defaultconfig = MkLSMRRBVectorConfig 64
237 |
238 | --------------------------------------------------------------------------------
239 | --          Creating Log-Structured Merge RRB-Vectors
240 | --------------------------------------------------------------------------------
241 |
242 | ||| Run an empty log-structured merge vector using a user-provided configuration.
243 | |||
244 | ||| Parameters:
245 | ||| - initialbatchwindow: Starting adaptive batching target.
246 | |||
247 | ||| Notes:
248 | ||| - Smaller values rebuild more aggressively.
249 | ||| - Larger values favor write throughput.
250 | |||
251 | export covering
252 | runEmptyWith :  Ord (Entry a)
253 |              => LSMRRBVectorConfig
254 |              -> List (LSMRRBVector World a -> RebuildService Poll -> RebuildServiceState -> Async Poll [Errno] ())
255 |              -> List (LSMRRBVector World a -> Async Poll [Errno] ())
256 |              -> IO ()
257 | runEmptyWith config rebuilderactions lsmrrbvectoractions = do
258 |   buffers                 <- newref Data.SortedMap.empty
259 |   combinedsnapshotstate   <- newref (MkCombinedSnapshotState (MkSnapshotState Z Empty) [] Data.SortedMap.empty 0 False config.initialbatchwindow)
260 |   rebuildscheduled        <- newref False
261 |   let lsmrrbvector        = MkLSMRRBVector buffers combinedsnapshotstate rebuildscheduled
262 |   let rebuilderservice    = rebuilderService lsmrrbvector initialRebuildServiceState rebuilderactions
263 |   let lsmrrbvectorservice = lsmrrbvectorService lsmrrbvector lsmrrbvectoractions
264 |   n                       <- asyncThreads
265 |   app n [SIGINT] posixPoller $ handle handlers (rebuilderAndLSMRRBVectorService rebuilderservice lsmrrbvectorservice)
266 |   where
267 |     handlers : All (Handler () Poll) [Errno]
268 |     handlers = [\x => stderrLn "Error: \{errorText x} (\{errorName x})"]
269 |
270 | ||| Runs an empty log-structured merge vector tuned for high sustained write throughput.
271 | |||
272 | ||| Configuration:
273 | ||| - Initial adaptive batch window: 512
274 | |||
275 | ||| Behavior:
276 | ||| - Favors larger rebuild batches.
277 | ||| - Reduces rebuild frequency under heavy write load.
278 | ||| - May increase visibility latency for newly written values.
279 | |||
280 | ||| Notes:
281 | ||| - Intended for write-heavy workloads.
282 | |||
283 | export covering
284 | runFastWritesEmpty :  Ord (Entry a)
285 |                    => List (LSMRRBVector World a -> RebuildService Poll -> RebuildServiceState -> Async Poll [Errno] ())
286 |                    -> List (LSMRRBVector World a -> Async Poll [Errno] ())
287 |                    -> IO ()
288 | runFastWritesEmpty rebuilderactions lsmrrbvectoractions =
289 |   runEmptyWith (MkLSMRRBVectorConfig 512) rebuilderactions lsmrrbvectoractions
290 |
291 | ||| Runs an empty log-structured merge vector tuned for low publication latency.
292 | |||
293 | ||| Configuration:
294 | ||| - Initial adaptive batch window: 16
295 | |||
296 | ||| Behavior:
297 | ||| - Favors frequent rebuild cycles.
298 | ||| - Reduces time between writes and publication.
299 | ||| - May increase rebuild overhead under heavy load.
300 | |||
301 | ||| Notes:
302 | ||| - Intended for latency-sensitive workloads.
303 | |||
304 | export covering
305 | runLowLatencyEmpty :  Ord (Entry a)
306 |                    => List (LSMRRBVector World a -> RebuildService Poll -> RebuildServiceState -> Async Poll [Errno] ())
307 |                    -> List (LSMRRBVector World a -> Async Poll [Errno] ())
308 |                    -> IO ()
309 | runLowLatencyEmpty rebuilderactions lsmrrbvectoractions =
310 |   runEmptyWith (MkLSMRRBVectorConfig 16) rebuilderactions lsmrrbvectoractions
311 |
312 | ||| Runs an empty log-structured merge vector.
313 | |||
314 | export covering
315 | runEmpty :  Ord (Entry a)
316 |          => List (LSMRRBVector World a -> RebuildService Poll -> RebuildServiceState -> Async Poll [Errno] ())
317 |          -> List (LSMRRBVector World a -> Async Poll [Errno] ())
318 |          -> IO ()
319 | runEmpty rebuilderactions lsmrrbvectoractions = 
320 |   runEmptyWith defaultconfig rebuilderactions lsmrrbvectoractions
321 |