1 // Block-decomposed lattice partitioning. Divides a lattice into
2 // contiguous blocks for cache-local growth. Each block is owned
3 // exclusively by one worker during a round — no per-node locks.
4 //
5 // The only shared structure between blocks is the spillover queue:
6 // an append-only list of elements that walked to a block boundary.
7 package grow
8 9 import (
10 "math/rand/v2"
11 "sync"
12 13 "git.mleku.dev/mleku/dendrite/pkg/axiom"
14 "git.mleku.dev/mleku/dendrite/pkg/lattice"
15 )
16 17 // DefaultBlockSize is the number of nodes per block. Tuned for L2 cache:
18 // 1024 nodes * ~200 bytes = ~200KB, fitting comfortably in per-core L2.
19 const DefaultBlockSize = 1024
20 21 // BlockState tracks whether a block's nodes carry full payload or have
22 // been stripped to skeleton (id + neighbors only) with payload on disk.
23 type BlockState uint8
24 25 const (
26 // BlockResident means node payloads are in memory; the block
27 // participates in rounds normally.
28 BlockResident BlockState = iota
29 30 // BlockStripped means node payloads have been serialized to disk
31 // and the in-memory nodes carry only id + neighbor pointers.
32 // The block holds deferred work queues until it is loaded.
33 BlockStripped
34 )
35 36 // Block is a contiguous partition of the lattice. During a round,
37 // exactly one worker owns a block — no locks needed on nodes within it.
38 type Block struct {
39 ID int
40 Start lattice.NodeID // inclusive
41 End lattice.NodeID // exclusive
42 Nodes []*lattice.Node
43 44 // Block-local vascular index: tag → local indices into Nodes slice.
45 vacantIdx map[string][]int
46 47 // Spillover: elements that walked to a boundary edge during this round.
48 // Protected by spillMu — the only lock in the system during rounds.
49 spillover []SpillItem
50 spillMu sync.Mutex
51 52 // Inbound: spillover from neighbors, drained at round start.
53 inbound []SpillItem
54 55 // active tracks whether this block has work to do.
56 active bool
57 58 // demandMap: tag → []uint16 indexed by local node offset.
59 // Value = BFS distance to nearest matching vacancy. demandUnreachable = no path.
60 // Rebuilt each round before walks begin.
61 demandMap map[string][]uint16
62 63 // boundaryDemand: demand signals received from neighboring blocks.
64 // Integrated into the demand map during buildDemandMap.
65 boundaryDemand []BoundarySignal
66 67 // --- Demand paging fields ---
68 69 // state tracks whether the block's nodes carry full payload.
70 state BlockState
71 72 // hasDiskCopy is true after the block has been frozen at least once.
73 // Avoids redundant freezes when evicting a block that hasn't changed.
74 hasDiskCopy bool
75 76 // deferredSpills collects spill items arriving while the block is
77 // stripped. Transferred to spillover on load.
78 deferredSpills []SpillItem
79 80 // deferredDemand collects boundary demand signals arriving while the
81 // block is stripped. Transferred to boundaryDemand on load.
82 deferredDemand []BoundarySignal
83 }
84 85 // SpillItem is an element that crossed a block boundary.
86 type SpillItem struct {
87 Element axiom.Element
88 }
89 90 // BoundarySignal carries demand from a neighboring block's vacancy.
91 // The wavefront propagates outward from vacant sites; when it hits a
92 // block boundary, the signal crosses to the adjacent block so the
93 // demand gradient extends across the full lattice.
94 type BoundarySignal struct {
95 Tag string
96 Distance uint16
97 EntryID lattice.NodeID // node in THIS block where demand enters
98 }
99 100 const demandUnreachable = 0xFFFF
101 102 // BlockMap holds the full decomposition of a lattice into blocks.
103 type BlockMap struct {
104 Blocks []*Block
105 NodeToBlock []int // NodeID → block index; flat array, L1-cacheable
106 BlockSize int
107 108 // --- Demand paging fields (zero values = paging disabled) ---
109 110 // residentCount is the number of blocks currently in BlockResident state.
111 residentCount int
112 113 // maxResident is the memory budget: maximum simultaneous resident blocks.
114 // 0 means all blocks stay resident (no paging).
115 maxResident int
116 117 // blockDir is the directory for block snapshot files.
118 blockDir string
119 120 // constraintFactory reconstructs Constraint objects from tag strings
121 // during block thaw.
122 constraintFactory func(string) axiom.Constraint
123 124 // lat is a back-reference to the lattice for node range access.
125 lat *lattice.Lattice
126 }
127 128 // buildBlockMap partitions the lattice into blocks of the given size.
129 // O(N) in lattice size. Builds per-block vacant indices.
130 func buildBlockMap(l *lattice.Lattice, blockSize int) *BlockMap {
131 if blockSize <= 0 {
132 blockSize = DefaultBlockSize
133 }
134 nodes := l.Nodes()
135 n := len(nodes)
136 if n == 0 {
137 return &BlockMap{BlockSize: blockSize}
138 }
139 140 numBlocks := (n + blockSize - 1) / blockSize
141 bm := &BlockMap{
142 Blocks: make([]*Block, numBlocks),
143 NodeToBlock: make([]int, n),
144 BlockSize: blockSize,
145 }
146 147 for i := range numBlocks {
148 start := i * blockSize
149 end := start + blockSize
150 if end > n {
151 end = n
152 }
153 b := &Block{
154 ID: i,
155 Start: lattice.NodeID(start),
156 End: lattice.NodeID(end),
157 Nodes: nodes[start:end],
158 vacantIdx: make(map[string][]int),
159 active: true,
160 state: BlockResident,
161 }
162 bm.Blocks[i] = b
163 164 // Build block-local vacant index.
165 for localIdx, node := range b.Nodes {
166 bm.NodeToBlock[node.ID()] = i
167 if !node.OccupiedUnsafe() {
168 for _, c := range node.Constraints() {
169 tag := c.Tag()
170 b.vacantIdx[tag] = append(b.vacantIdx[tag], localIdx)
171 }
172 }
173 }
174 }
175 176 return bm
177 }
178 179 // vacantByTag returns a vacant node within this block matching the tag.
180 // Lazily compacts stale entries. No lock needed — caller owns the block.
181 func (b *Block) vacantByTag(tag string) *lattice.Node {
182 ids := b.vacantIdx[tag]
183 for attempts := len(ids); attempts > 0 && len(ids) > 0; attempts-- {
184 idx := rand.IntN(len(ids))
185 localIdx := ids[idx]
186 n := b.Nodes[localIdx]
187 if !n.OccupiedUnsafe() {
188 return n
189 }
190 // Stale — swap-remove.
191 ids[idx] = ids[len(ids)-1]
192 ids = ids[:len(ids)-1]
193 b.vacantIdx[tag] = ids
194 }
195 return nil
196 }
197 198 // randomNode returns a random node within this block.
199 func (b *Block) randomNode() *lattice.Node {
200 if len(b.Nodes) == 0 {
201 return nil
202 }
203 return b.Nodes[rand.IntN(len(b.Nodes))]
204 }
205 206 // removeFromVacant removes a node from the block's vacant index
207 // after it has been bonded. Called by the walker after a successful bond.
208 func (b *Block) removeFromVacant(n *lattice.Node) {
209 localIdx := int(n.ID()) - int(b.Start)
210 for tag, ids := range b.vacantIdx {
211 for i, id := range ids {
212 if id == localIdx {
213 ids[i] = ids[len(ids)-1]
214 ids = ids[:len(ids)-1]
215 b.vacantIdx[tag] = ids
216 break
217 }
218 }
219 }
220 }
221 222 // pushSpill adds a spill item to this block's spillover queue.
223 // Thread-safe — this is the only lock taken during a round.
224 // If the block is stripped, the item goes to the deferred queue
225 // and will be transferred on the next load.
226 func (b *Block) pushSpill(item SpillItem) {
227 b.spillMu.Lock()
228 if b.state == BlockStripped {
229 b.deferredSpills = append(b.deferredSpills, item)
230 } else {
231 b.spillover = append(b.spillover, item)
232 }
233 b.spillMu.Unlock()
234 }
235 236 // drainSpill moves spillover into inbound and clears spillover.
237 // Called between rounds, no concurrent access.
238 func (b *Block) drainSpill() {
239 b.inbound = append(b.inbound[:0], b.spillover...)
240 b.spillover = b.spillover[:0]
241 }
242 243 // isInBlock reports whether a node belongs to this block.
244 func (bm *BlockMap) isInBlock(nodeID lattice.NodeID, blockID int) bool {
245 return bm.NodeToBlock[nodeID] == blockID
246 }
247 248 // buildDemandMap runs multi-source BFS from all vacant sites within this
249 // block for each constraint tag. The result is a distance map: for each
250 // node, how many hops to the nearest vacancy of that tag. O(nodes × tags).
251 //
252 // Also integrates boundary demand signals from neighboring blocks, seeding
253 // the BFS with external demand sources so the gradient extends across
254 // block boundaries.
255 func (b *Block) buildDemandMap(bm *BlockMap) {
256 n := len(b.Nodes)
257 if n == 0 {
258 return
259 }
260 261 // Collect all tags that have vacancies in this block.
262 tags := make(map[string]bool)
263 for tag := range b.vacantIdx {
264 tags[tag] = true
265 }
266 // Also include tags from boundary demand signals.
267 for _, sig := range b.boundaryDemand {
268 tags[sig.Tag] = true
269 }
270 271 if b.demandMap == nil {
272 b.demandMap = make(map[string][]uint16, len(tags))
273 }
274 275 for tag := range tags {
276 // Get or allocate the distance array for this tag.
277 dist := b.demandMap[tag]
278 if cap(dist) >= n {
279 dist = dist[:n]
280 } else {
281 dist = make([]uint16, n)
282 }
283 for i := range dist {
284 dist[i] = demandUnreachable
285 }
286 287 // BFS queue — local indices.
288 queue := make([]int, 0, 64)
289 290 // Seed: all vacant nodes matching this tag.
291 for _, localIdx := range b.vacantIdx[tag] {
292 if localIdx < n && !b.Nodes[localIdx].OccupiedUnsafe() {
293 dist[localIdx] = 0
294 queue = append(queue, localIdx)
295 }
296 }
297 298 // Seed: boundary demand signals for this tag.
299 for _, sig := range b.boundaryDemand {
300 if sig.Tag != tag {
301 continue
302 }
303 localIdx := int(sig.EntryID) - int(b.Start)
304 if localIdx < 0 || localIdx >= n {
305 continue
306 }
307 if sig.Distance < dist[localIdx] {
308 dist[localIdx] = sig.Distance
309 queue = append(queue, localIdx)
310 }
311 }
312 313 // BFS expansion — only within this block.
314 for len(queue) > 0 {
315 cur := queue[0]
316 queue = queue[1:]
317 curDist := dist[cur]
318 next := curDist + 1
319 320 for _, nb := range b.Nodes[cur].NeighborsUnsafe() {
321 if !bm.isInBlock(nb.ID(), b.ID) {
322 continue // cross-block neighbor, skip
323 }
324 localNb := int(nb.ID()) - int(b.Start)
325 if localNb < 0 || localNb >= n {
326 continue
327 }
328 if next < dist[localNb] {
329 dist[localNb] = next
330 queue = append(queue, localNb)
331 }
332 }
333 }
334 335 b.demandMap[tag] = dist
336 }
337 338 // Clear boundary signals after integration.
339 b.boundaryDemand = b.boundaryDemand[:0]
340 }
341 342 // demandAt returns the BFS distance to the nearest vacancy matching tag
343 // at the given local index. Returns demandUnreachable if no path.
344 func (b *Block) demandAt(tag string, localIdx int) uint16 {
345 dist := b.demandMap[tag]
346 if localIdx < 0 || localIdx >= len(dist) {
347 return demandUnreachable
348 }
349 return dist[localIdx]
350 }
351 352 // propagateBoundaryDemand pushes demand signals to adjacent blocks.
353 // For each boundary node in this block that has a finite demand distance,
354 // emit a signal to each cross-block neighbor so the demand wavefront
355 // extends across block boundaries.
356 func (b *Block) propagateBoundaryDemand(bm *BlockMap) {
357 n := len(b.Nodes)
358 for tag, dist := range b.demandMap {
359 for localIdx := range n {
360 d := dist[localIdx]
361 if d == demandUnreachable {
362 continue
363 }
364 // Check if this node has any cross-block neighbors.
365 for _, nb := range b.Nodes[localIdx].NeighborsUnsafe() {
366 nbBlock := bm.NodeToBlock[nb.ID()]
367 if nbBlock == b.ID {
368 continue // same block
369 }
370 // Push demand signal to the neighboring block.
371 // The entry point is the neighbor node in that block.
372 sig := BoundarySignal{
373 Tag: tag,
374 Distance: d + 1,
375 EntryID: nb.ID(),
376 }
377 target := bm.Blocks[nbBlock]
378 if target.state == BlockStripped {
379 target.deferredDemand = append(target.deferredDemand, sig)
380 } else {
381 target.boundaryDemand = append(target.boundaryDemand, sig)
382 }
383 }
384 }
385 }
386 }
387 388 // rebuildVacantIdx reconstructs the block-local vacant index from the
389 // current node state. Called after thawing a block from disk.
390 func (b *Block) rebuildVacantIdx() {
391 b.vacantIdx = make(map[string][]int)
392 for localIdx, node := range b.Nodes {
393 if !node.OccupiedUnsafe() {
394 for _, c := range node.Constraints() {
395 tag := c.Tag()
396 b.vacantIdx[tag] = append(b.vacantIdx[tag], localIdx)
397 }
398 }
399 }
400 }
401 402 // --- BlockMap paging operations ---
403 404 // pagingEnabled reports whether demand paging is active.
405 func (bm *BlockMap) pagingEnabled() bool {
406 return bm.maxResident > 0 && bm.maxResident < len(bm.Blocks)
407 }
408 409 // loadNeededBlocks scans for stripped blocks that have deferred work
410 // and loads them. Called between rounds (Phase 0).
411 func (bm *BlockMap) loadNeededBlocks() error {
412 for _, b := range bm.Blocks {
413 if b.state != BlockStripped {
414 continue
415 }
416 if len(b.deferredSpills) == 0 && len(b.deferredDemand) == 0 {
417 continue
418 }
419 if err := bm.loadBlock(b); err != nil {
420 return err
421 }
422 }
423 return nil
424 }
425 426 // loadBlock thaws a stripped block from disk. If at capacity, evicts
427 // a victim block first.
428 func (bm *BlockMap) loadBlock(b *Block) error {
429 // Make room if at capacity.
430 if bm.residentCount >= bm.maxResident {
431 victim := bm.pickEvictionVictim(b.ID)
432 if victim != nil {
433 if err := bm.stripBlock(victim); err != nil {
434 return err
435 }
436 }
437 }
438 439 // Thaw from disk.
440 if err := thawBlock(b, bm.blockDir, bm.constraintFactory); err != nil {
441 return err
442 }
443 444 // Transfer deferred work to normal queues.
445 b.spillMu.Lock()
446 b.spillover = append(b.spillover, b.deferredSpills...)
447 b.deferredSpills = b.deferredSpills[:0]
448 b.spillMu.Unlock()
449 450 b.boundaryDemand = append(b.boundaryDemand, b.deferredDemand...)
451 b.deferredDemand = b.deferredDemand[:0]
452 453 // Rebuild block-local state.
454 b.rebuildVacantIdx()
455 b.state = BlockResident
456 b.active = true
457 bm.residentCount++
458 return nil
459 }
460 461 // stripBlock freezes a block to disk and strips its nodes to skeleton.
462 // The block transitions to BlockStripped state.
463 func (bm *BlockMap) stripBlock(b *Block) error {
464 if b.state != BlockResident {
465 return nil // already stripped
466 }
467 468 // Freeze to disk (always freeze — correctness over avoiding writes).
469 if err := freezeBlock(b, bm.blockDir); err != nil {
470 return err
471 }
472 b.hasDiskCopy = true
473 474 // Strip node payloads in place — preserves id + neighbor pointers.
475 for _, n := range b.Nodes {
476 n.StripForEvictionUnsafe()
477 }
478 479 // Move any in-flight spillover to deferred so it survives stripping.
480 b.spillMu.Lock()
481 if len(b.spillover) > 0 {
482 b.deferredSpills = append(b.deferredSpills, b.spillover...)
483 b.spillover = b.spillover[:0]
484 }
485 if len(b.inbound) > 0 {
486 for _, item := range b.inbound {
487 b.deferredSpills = append(b.deferredSpills, item)
488 }
489 b.inbound = b.inbound[:0]
490 }
491 b.spillMu.Unlock()
492 493 // Clear block-level caches.
494 b.vacantIdx = nil
495 b.demandMap = nil
496 b.state = BlockStripped
497 b.active = false
498 bm.residentCount--
499 return nil
500 }
501 502 // pickEvictionVictim selects a resident block to evict. Prefers blocks
503 // that are not active and have no pending spills. Avoids the block
504 // being loaded (excludeID).
505 func (bm *BlockMap) pickEvictionVictim(excludeID int) *Block {
506 for _, b := range bm.Blocks {
507 if b.ID == excludeID {
508 continue
509 }
510 if b.state != BlockResident {
511 continue
512 }
513 if !b.active && len(b.spillover) == 0 && len(b.inbound) == 0 {
514 return b
515 }
516 }
517 // All resident blocks are active — evict the first non-excluded resident.
518 for _, b := range bm.Blocks {
519 if b.ID == excludeID {
520 continue
521 }
522 if b.state == BlockResident {
523 return b
524 }
525 }
526 return nil
527 }
528 529 // evictExhaustedBlocks strips blocks that were not active this round
530 // and have no pending work. Called after processing (Phase 7).
531 func (bm *BlockMap) evictExhaustedBlocks(activeSet map[int]bool) error {
532 if bm.residentCount <= bm.maxResident {
533 return nil
534 }
535 for _, b := range bm.Blocks {
536 if bm.residentCount <= bm.maxResident {
537 break
538 }
539 if b.state != BlockResident {
540 continue
541 }
542 if activeSet[b.ID] {
543 continue
544 }
545 if len(b.spillover) > 0 || len(b.inbound) > 0 {
546 continue
547 }
548 if err := bm.stripBlock(b); err != nil {
549 return err
550 }
551 }
552 return nil
553 }
554