blockgrow.go raw
1 // Block-decomposed growth loop. Partitions the lattice into cache-local
2 // blocks, assigns each block exclusively to one worker per round, and
3 // uses boundary spillover queues as the only shared structure.
4 //
5 // Convergence: each round either bonds elements (reducing work) or
6 // saturates blocks (removing them from the active set). After 3
7 // consecutive stall rounds (no bonds, no reduction in spills), all
8 // remaining elements are expired.
9 package grow
10
11 import (
12 "context"
13 "math/rand/v2"
14 "os"
15 "sync"
16 "time"
17
18 "git.mleku.dev/mleku/dendrite/pkg/axiom"
19 "git.mleku.dev/mleku/dendrite/pkg/lattice"
20 )
21
22 // RunBlocked is the block-decomposed growth loop. Same channel protocol
23 // and event semantics as Run. For small lattices (< 2*BlockSize nodes),
24 // falls back to Run.
25 func RunBlocked(ctx context.Context, l *lattice.Lattice, solution <-chan axiom.Element, cfg Config, events chan<- Event) {
26 blockSize := cfg.BlockSize
27 if blockSize <= 0 {
28 blockSize = DefaultBlockSize
29 }
30
31 // Small lattice bypass — decomposition overhead exceeds benefit.
32 if l.Size() < 2*blockSize {
33 Run(ctx, l, solution, cfg, events)
34 return
35 }
36
37 bm := buildBlockMap(l, blockSize)
38
39 // Configure demand paging if requested.
40 if cfg.MaxResidentBlocks > 0 && cfg.MaxResidentBlocks < len(bm.Blocks) {
41 bm.maxResident = cfg.MaxResidentBlocks
42 bm.blockDir = cfg.BlockDir
43 bm.constraintFactory = cfg.ConstraintFactory
44 bm.lat = l
45 bm.residentCount = len(bm.Blocks) // all start resident
46
47 // Ensure block directory exists.
48 os.MkdirAll(bm.blockDir, 0o755)
49
50 // Evict excess blocks down to budget. Keep blocks with
51 // the lowest IDs resident (arbitrary but deterministic).
52 for i := len(bm.Blocks) - 1; i >= 0 && bm.residentCount > bm.maxResident; i-- {
53 bm.stripBlock(bm.Blocks[i])
54 }
55 }
56
57 // Drain solution channel into per-block element queues.
58 // Use global VacantByTag to find the best starting block for each element.
59 // The drain uses a short idle timeout: when no elements arrive within 50ms,
60 // we assume the injection goroutine has finished and move to processing.
61 // This avoids consuming the growth context budget on the drain phase.
62 blockQueues := make([][]axiom.Element, len(bm.Blocks))
63 robin := 0
64 idleTimeout := 500 * time.Millisecond
65 idleTimer := time.NewTimer(idleTimeout)
66 defer idleTimer.Stop()
67 drainLoop:
68 for {
69 select {
70 case <-ctx.Done():
71 break drainLoop
72 case elem, ok := <-solution:
73 if !ok {
74 break drainLoop
75 }
76 placed := false
77 // Try directed placement via the lattice's vascular index.
78 if n := l.VacantByTag(elem.Type()); n != nil {
79 bid := bm.NodeToBlock[n.ID()]
80 blockQueues[bid] = append(blockQueues[bid], elem)
81 placed = true
82 }
83 if !placed {
84 blockQueues[robin%len(bm.Blocks)] = append(blockQueues[robin%len(bm.Blocks)], elem)
85 robin++
86 }
87 // Reset idle timer on each received element.
88 if !idleTimer.Stop() {
89 select {
90 case <-idleTimer.C:
91 default:
92 }
93 }
94 idleTimer.Reset(idleTimeout)
95 case <-idleTimer.C:
96 // No elements received within idle window — injection complete.
97 break drainLoop
98 }
99 }
100
101 // Round loop.
102 workers := cfg.Workers
103 if workers <= 0 {
104 workers = WorkerCount()
105 }
106 maxRounds := cfg.MaxRounds
107 if maxRounds <= 0 {
108 maxRounds = 100 // safety cap
109 }
110
111 prevSpillCount := -1
112 stallRounds := 0
113 const maxStall = 3
114
115 for round := range maxRounds {
116 _ = round
117
118 select {
119 case <-ctx.Done():
120 return
121 default:
122 }
123
124 // Phase 0: LOAD — thaw stripped blocks that received deferred work
125 // or have queued elements from the solution distribution.
126 if bm.pagingEnabled() {
127 // Check blockQueues for stripped blocks — treat as deferred work.
128 for _, b := range bm.Blocks {
129 if b.state == BlockStripped && len(blockQueues[b.ID]) > 0 {
130 for _, elem := range blockQueues[b.ID] {
131 b.deferredSpills = append(b.deferredSpills, SpillItem{Element: elem})
132 }
133 blockQueues[b.ID] = nil
134 }
135 }
136 if err := bm.loadNeededBlocks(); err != nil {
137 return // disk error — bail
138 }
139 }
140
141 // Phase 1: Move spillover → inbound for resident blocks.
142 for _, b := range bm.Blocks {
143 if b.state == BlockResident {
144 b.drainSpill()
145 }
146 }
147
148 // Phase 2: Build demand maps (BFS from vacancies).
149 for _, b := range bm.Blocks {
150 if b.state == BlockResident {
151 b.buildDemandMap(bm)
152 }
153 }
154
155 // Phase 3: Propagate demand across block boundaries.
156 for _, b := range bm.Blocks {
157 if b.state == BlockResident {
158 b.propagateBoundaryDemand(bm)
159 }
160 }
161
162 // Phase 4: Merge block queues + inbound into work lists.
163 type blockWork struct {
164 block *Block
165 elements []axiom.Element
166 }
167 var active []blockWork
168 activeSet := make(map[int]bool)
169 for _, b := range bm.Blocks {
170 if b.state != BlockResident {
171 continue
172 }
173 var work []axiom.Element
174 if len(blockQueues[b.ID]) > 0 {
175 work = append(work, blockQueues[b.ID]...)
176 blockQueues[b.ID] = nil
177 }
178 if len(b.inbound) > 0 {
179 for _, item := range b.inbound {
180 work = append(work, item.Element)
181 }
182 b.inbound = b.inbound[:0]
183 }
184 if len(work) > 0 {
185 b.active = true
186 active = append(active, blockWork{block: b, elements: work})
187 activeSet[b.ID] = true
188 } else {
189 b.active = false
190 }
191 }
192
193 if len(active) == 0 {
194 break
195 }
196
197 // Phase 5: Process active blocks in parallel.
198 var wg sync.WaitGroup
199 sem := make(chan struct{}, workers)
200
201 var evMu sync.Mutex
202 var roundEvents []Event
203
204 for _, bw := range active {
205 wg.Add(1)
206 sem <- struct{}{} // acquire worker slot
207 go func(b *Block, elems []axiom.Element) {
208 defer func() { <-sem; wg.Done() }()
209 evs := processBlock(ctx, b, bm, elems, cfg.MaxSteps)
210 evMu.Lock()
211 roundEvents = append(roundEvents, evs...)
212 evMu.Unlock()
213 }(bw.block, bw.elements)
214 }
215 wg.Wait()
216
217 // Phase 6: Emit events.
218 for _, ev := range roundEvents {
219 select {
220 case events <- ev:
221 case <-ctx.Done():
222 return
223 }
224 }
225
226 // Phase 7: EVICT — strip exhausted blocks to make room.
227 if bm.pagingEnabled() {
228 if err := bm.evictExhaustedBlocks(activeSet); err != nil {
229 return // disk error — bail
230 }
231 }
232
233 // Phase 8: Convergence / stall detection.
234 spillCount := 0
235 for _, b := range bm.Blocks {
236 spillCount += len(b.spillover)
237 spillCount += len(b.deferredSpills)
238 }
239
240 if spillCount == 0 {
241 break // converged
242 }
243
244 if spillCount >= prevSpillCount && prevSpillCount >= 0 {
245 stallRounds++
246 } else {
247 stallRounds = 0
248 }
249 prevSpillCount = spillCount
250
251 if stallRounds >= maxStall {
252 // Expire all remaining spill items (resident + deferred).
253 for _, b := range bm.Blocks {
254 b.spillMu.Lock()
255 for _, item := range b.spillover {
256 select {
257 case events <- Event{Type: EventExpired, Element: item.Element}:
258 case <-ctx.Done():
259 b.spillMu.Unlock()
260 return
261 }
262 }
263 b.spillover = b.spillover[:0]
264 for _, item := range b.deferredSpills {
265 select {
266 case events <- Event{Type: EventExpired, Element: item.Element}:
267 case <-ctx.Done():
268 b.spillMu.Unlock()
269 return
270 }
271 }
272 b.deferredSpills = b.deferredSpills[:0]
273 b.spillMu.Unlock()
274 }
275 break
276 }
277 }
278
279 // Restore all stripped blocks so the lattice reflects the final state.
280 if bm.pagingEnabled() {
281 for _, b := range bm.Blocks {
282 if b.state == BlockStripped && b.hasDiskCopy {
283 thawBlock(b, bm.blockDir, bm.constraintFactory)
284 b.state = BlockResident
285 bm.residentCount++
286 }
287 }
288 }
289 }
290
291 // processBlock runs all walks for one block in one round.
292 // Returns events for bonded/rejected/expired elements. Spilled elements
293 // are deposited directly into neighbor blocks' spillover queues.
294 func processBlock(ctx context.Context, b *Block, bm *BlockMap, elements []axiom.Element, maxSteps int) []Event {
295 events := make([]Event, 0, len(elements))
296 for _, elem := range elements {
297 select {
298 case <-ctx.Done():
299 return events
300 default:
301 }
302 ev := blockWalk(ctx, b, bm, elem, maxSteps)
303 if ev != nil {
304 events = append(events, *ev)
305 }
306 }
307 return events
308 }
309
310 // blockWalk performs a wavefront-guided walk within a single block.
311 // The demand map (BFS distance to nearest matching vacancy) provides
312 // the gradient. The element follows the gradient downhill, bonding
313 // when it reaches a vacancy. If no demand exists in this block for the
314 // element's type, it spills to a neighbor block with closer demand.
315 //
316 // Returns nil for spilled elements (no event — continues next round).
317 func blockWalk(ctx context.Context, b *Block, bm *BlockMap, elem axiom.Element, maxSteps int) *Event {
318 tag := elem.Type()
319
320 // Find the best starting node: lowest demand distance for this tag.
321 var current *lattice.Node
322 bestDist := uint16(demandUnreachable)
323
324 dist := b.demandMap[tag]
325 if len(dist) > 0 {
326 for localIdx, d := range dist {
327 if d < bestDist {
328 bestDist = d
329 current = b.Nodes[localIdx]
330 }
331 }
332 }
333
334 // No demand for this tag in this block — spill to the neighbor
335 // block with the closest boundary demand.
336 if bestDist == demandUnreachable {
337 return spillToClosestDemand(b, bm, elem, tag)
338 }
339
340 // If the best starting point is the vacancy itself, bond directly.
341 if bestDist == 0 && current != nil && current.AdmitsUnsafe(elem) {
342 if current.BondUnsafe(elem) {
343 b.removeFromVacant(current)
344 ev := Event{Type: EventBonded, NodeID: current.ID(), Element: elem, Steps: 0}
345 return &ev
346 }
347 }
348
349 if current == nil {
350 ev := Event{Type: EventRejected, Element: elem}
351 return &ev
352 }
353
354 // Gradient descent: follow the demand map downhill.
355 for step := range maxSteps {
356 if step&0xF == 0 {
357 select {
358 case <-ctx.Done():
359 ev := Event{Type: EventExpired, Element: elem, Steps: step}
360 return &ev
361 default:
362 }
363 }
364
365 // Try bond at current node.
366 if current.AdmitsUnsafe(elem) {
367 if current.BondUnsafe(elem) {
368 b.removeFromVacant(current)
369 ev := Event{Type: EventBonded, NodeID: current.ID(), Element: elem, Steps: step}
370 return &ev
371 }
372 }
373
374 // Follow gradient: pick the neighbor with the lowest demand distance.
375 neighbors := current.NeighborsUnsafe()
376 curLocalIdx := int(current.ID()) - int(b.Start)
377 curDist := uint16(demandUnreachable)
378 if curLocalIdx >= 0 && curLocalIdx < len(dist) {
379 curDist = dist[curLocalIdx]
380 }
381
382 var bestNb *lattice.Node
383 bestNbDist := curDist // must improve on current
384 var spillCandidate *lattice.Node
385
386 for _, nb := range neighbors {
387 if !bm.isInBlock(nb.ID(), b.ID) {
388 spillCandidate = nb
389 continue
390 }
391 // Try immediate bond on neighbor.
392 if nb.AdmitsUnsafe(elem) {
393 if nb.BondUnsafe(elem) {
394 b.removeFromVacant(nb)
395 ev := Event{Type: EventBonded, NodeID: nb.ID(), Element: elem, Steps: step}
396 return &ev
397 }
398 }
399 nbLocalIdx := int(nb.ID()) - int(b.Start)
400 if nbLocalIdx >= 0 && nbLocalIdx < len(dist) {
401 d := dist[nbLocalIdx]
402 if d < bestNbDist {
403 bestNbDist = d
404 bestNb = nb
405 }
406 }
407 }
408
409 if bestNb != nil {
410 current = bestNb
411 continue
412 }
413
414 // No downhill neighbor — gradient bottomed out.
415 // Spill to adjacent block if there's a cross-block neighbor.
416 if spillCandidate != nil {
417 targetBlockID := bm.NodeToBlock[spillCandidate.ID()]
418 bm.Blocks[targetBlockID].pushSpill(SpillItem{Element: elem})
419 return nil
420 }
421
422 // Dead end — no in-block improvement and no cross-block escape.
423 // Try a random in-block neighbor to escape local minimum.
424 var inBlock [8]*lattice.Node
425 inBlockSlice := inBlock[:0]
426 for _, nb := range neighbors {
427 if bm.isInBlock(nb.ID(), b.ID) {
428 inBlockSlice = append(inBlockSlice, nb)
429 }
430 }
431 if len(inBlockSlice) > 0 {
432 current = inBlockSlice[rand.IntN(len(inBlockSlice))]
433 continue
434 }
435
436 ev := Event{Type: EventRejected, Element: elem, Steps: maxSteps}
437 return &ev
438 }
439
440 ev := Event{Type: EventExpired, Element: elem, Steps: maxSteps}
441 return &ev
442 }
443
444 // spillToClosestDemand spills an element to the neighbor block most
445 // likely to have demand for the element's type. Checks boundary nodes
446 // for cross-block neighbors and picks the block with any demand signal
447 // for the tag. Falls back to any neighbor block. Prefers resident blocks
448 // over stripped blocks to avoid deferred-queue ping-pong.
449 func spillToClosestDemand(b *Block, bm *BlockMap, elem axiom.Element, tag string) *Event {
450 // Pass 1: find a resident neighbor block with demand for this tag.
451 for _, node := range b.Nodes {
452 for _, nb := range node.NeighborsUnsafe() {
453 nbBlockID := bm.NodeToBlock[nb.ID()]
454 if nbBlockID == b.ID {
455 continue
456 }
457 target := bm.Blocks[nbBlockID]
458 if target.state != BlockResident {
459 continue
460 }
461 if targetDist := target.demandMap[tag]; len(targetDist) > 0 {
462 for _, d := range targetDist {
463 if d < demandUnreachable {
464 target.pushSpill(SpillItem{Element: elem})
465 return nil
466 }
467 }
468 }
469 }
470 }
471
472 // Pass 2: any resident neighbor block (even without demand for this tag).
473 for _, node := range b.Nodes {
474 for _, nb := range node.NeighborsUnsafe() {
475 nbBlockID := bm.NodeToBlock[nb.ID()]
476 if nbBlockID == b.ID {
477 continue
478 }
479 target := bm.Blocks[nbBlockID]
480 if target.state == BlockResident {
481 target.pushSpill(SpillItem{Element: elem})
482 return nil
483 }
484 }
485 }
486
487 // Pass 3: any neighbor block (may be stripped — triggers load next round).
488 for _, node := range b.Nodes {
489 for _, nb := range node.NeighborsUnsafe() {
490 nbBlockID := bm.NodeToBlock[nb.ID()]
491 if nbBlockID != b.ID {
492 bm.Blocks[nbBlockID].pushSpill(SpillItem{Element: elem})
493 return nil
494 }
495 }
496 }
497
498 // Completely isolated block — reject.
499 ev := Event{Type: EventRejected, Element: elem}
500 return &ev
501 }
502