diff --git a/Makefile b/Makefile index 5106361..20b9287 100644 --- a/Makefile +++ b/Makefile @@ -5,6 +5,6 @@ test: test-race: go test -race ./internal/network ./internal/server ./internal/world \ - -run 'Test(Integration|BoundaryEdit|PlayerInfo|PlayerRegistry|Concurrent|Incremental|EncodeLight|Cache|Store|Eviction|Region)' + -run 'Test(Integration|BoundaryEdit|PlayerInfo|PlayerRegistry|Concurrent|Incremental|EncodeLight|Cache|Store|Eviction|Region|Ticket|Streamer|LoadSixteen)' verify: test test-race diff --git a/README.md b/README.md index a2338e4..ae46433 100644 --- a/README.md +++ b/README.md @@ -13,8 +13,8 @@ block editing, persistent worlds, and an overworld generator built on the real - **Registries**: 28 synchronized registries + tags, captured verbatim from the 26.1.2 vanilla server and sent during configuration. - **World**: revisioned, concurrency-safe chunk snapshots; memoized - `level_chunk_with_light` frames; bounded LRU cache; Anvil `.mca` persistence - with autosave and seed metadata. + `level_chunk_with_light` frames; ticket-aware bounded LRU cache; shared frame + admission limit; Anvil `.mca` persistence with autosave and seed metadata. - **Generation**: vanilla-derived overworld terrain from the embedded datapack (`ImprovedNoise`/`PerlinNoise`/`BlendedNoise`/`NormalNoise` + the density function interpreter), 3D multi-noise biomes, surface-rule interpretation, @@ -24,7 +24,11 @@ block editing, persistent worlds, and an overworld generator built on the real creative block place/break; broadcast chat; and hotbar item→block mapping. - **Lighting**: stored vanilla nibble arrays for sky and block light; horizontal and cross-chunk propagation; incremental updates after edits; persisted - `SkyLight`/`BlockLight`; and chunk-scoped `light_update` broadcasts. + `SkyLight`/`BlockLight`; load-time border reconciliation; and chunk-scoped + `light_update` broadcasts. +- **Chunk lifecycle**: per-client view and prefetch tickets, strict near-first + ring streaming, stale-recenter cutoff, explicit client unload packets, and + eviction only after the final owner releases a chunk. - **Safety**: duplicate chunk generation is coalesced; corrupt stored chunks are not silently regenerated or overwritten; a world cannot reopen with another seed. @@ -45,7 +49,7 @@ Changing the seed for an existing world directory is rejected. ``` go test ./... go test -race ./internal/network ./internal/server ./internal/world \ - -run 'Test(Integration|BoundaryEdit|PlayerInfo|PlayerRegistry|Concurrent|Incremental|EncodeLight|Cache|Store|Eviction|Region)' + -run 'Test(Integration|BoundaryEdit|PlayerInfo|PlayerRegistry|Concurrent|Incremental|EncodeLight|Cache|Store|Eviction|Region|Ticket|Streamer|LoadSixteen)' # or run both gates: make verify ``` @@ -54,23 +58,27 @@ The integration suite exercises four clients across two visibility regions: join, movement, leaving, mob visibility, and local block/light updates. A two-client scenario separately covers shared block edits and chat. Concurrency tests cover simultaneous frame encoding, editing, autosave, cache misses, and -session movement/broadcasts. Lighting tests compare the initial flat chunk and a -31x31x31 glowstone propagation volume against fixtures captured from the -official vanilla 26.1.2 server. Optional terrain parity diagnostics compare -surface heights against `/tmp/vanilla_ground.json` when that capture is present. +session movement/broadcasts. A 16-client lifecycle test exercises overlapping +ticket ownership, bounded global frame work, packet output, and cleanup after +disconnect. Lighting tests compare the initial flat chunk and a 31x31x31 +glowstone propagation volume against fixtures captured from the official +vanilla 26.1.2 server. Optional terrain parity diagnostics compare surface +heights against `/tmp/vanilla_ground.json` when that capture is present. -## v0.3 scope +## v0.4 scope -RegionIO v0.3 is a small creative multiplayer server core, not a complete +RegionIO v0.4 is a small creative multiplayer server core, not a complete vanilla gameplay implementation. Player and mob visibility is chunk-scoped, but there is no interest prioritization or delta-movement compression yet. Lighting matches vanilla's block-state dampening, emission, and face-occlusion -properties and propagates across loaded chunk boundaries. Chunks outside the -live cache are recalculated exactly when loaded rather than retained as active -light-engine state. Structures, placed features, mob AI, authentication, -inventory, and survival mechanics remain intentionally partial. The density -router is vanilla-derived, while biome/surface/decoration layers still contain -approximations and require stricter parity fixtures. +properties and reconciles persisted borders when chunks re-enter the live +cache. Streaming prioritizes Chebyshev rings and abandons unstarted stale work; +an already admitted frame calculation completes atomically rather than being +interrupted halfway. Unowned clean chunks remain as an LRU warm cache until +capacity pressure evicts them. Structures, placed features, mob AI, +authentication, inventory, and survival mechanics remain intentionally partial. +The density router is vanilla-derived, while biome/surface/decoration layers +still contain approximations and require stricter parity fixtures. ## Project layout diff --git a/internal/network/multiplayer_test.go b/internal/network/multiplayer_test.go index 13bbae9..2b391f7 100644 --- a/internal/network/multiplayer_test.go +++ b/internal/network/multiplayer_test.go @@ -53,6 +53,25 @@ func (c *recordingConn) take(t *testing.T) []protocol.Packet { return packets } +func (c *recordingConn) countPacketID(id int32) int { + c.mu.Lock() + raw := append([]byte(nil), c.buf.Bytes()...) + c.mu.Unlock() + reader := bytes.NewReader(raw) + br := bufio.NewReader(reader) + count := 0 + for br.Buffered() > 0 || reader.Len() > 0 { + packet, err := protocol.ReadPacket(br, -1) + if err != nil { + return count + } + if packet.ID == id { + count++ + } + } + return count +} + type testAddr string func (a testAddr) Network() string { return "test" } diff --git a/internal/network/streamer.go b/internal/network/streamer.go index 288bbb7..dd6ead4 100644 --- a/internal/network/streamer.go +++ b/internal/network/streamer.go @@ -13,10 +13,9 @@ import ( // streamer.go is the per-connection background chunk streamer. The read loop // no longer generates or sends chunks inline; it pushes recenter requests here // and stays free to handle the player's packets (movement, chat, keep-alive -// acks). The streamer generates chunks in a worker pool (Cache.Frame is -// goroutine-safe), pre-generates a ring beyond the view distance so movement -// doesn't pop in, and sends finished frames serially under the conn's write -// mutex. +// acks). The streamer holds cache tickets for the view and one predictive ring, +// admits work in distance-priority batches, and sends finished frames serially +// under the conn's write mutex. // // Ownership: // - read loop: calls requestRecenter (non-blocking), owns nothing else here. @@ -45,6 +44,7 @@ type streamer struct { viewRadius int // chunks within this Chebyshev radius are sent to the client genRadius int // viewRadius + 1: pre-generated but not sent (predictive ring) poolSize int // parallel generation workers + tickets *world.TicketSet } // defaultViewRadius is used when the client hasn't sent client_information or @@ -68,7 +68,7 @@ func newStreamer(cache *world.Cache, conn *Conn, log *slog.Logger, viewDistance if pool < 2 { pool = 2 } - return &streamer{ + s := &streamer{ cache: cache, conn: conn, log: log, @@ -78,6 +78,10 @@ func newStreamer(cache *world.Cache, conn *Conn, log *slog.Logger, viewDistance genRadius: viewDistance + 1, poolSize: pool, } + if cache != nil { + s.tickets = cache.NewTicketSet() + } + return s } // requestRecenter asks the streamer to recenter on (cx, cz). Non-blocking: if @@ -105,41 +109,30 @@ func (s *streamer) requestRecenter(cx, cz int32) { // connection close). On each recenter it generates+sends the newly-in-range // chunks in spiral order (nearest first) and pre-generates the outer ring. func (s *streamer) run(ctx context.Context) { - var ( - // latest holds the most recent recenter request; processed when the - // previous batch finishes or on arrival if idle. - pending bool - next recenterReq - ) + if s.tickets != nil { + defer s.tickets.Close() + } for { - // If we have a pending recenter, process it; otherwise block waiting. - if pending { - select { - case <-ctx.Done(): - return - case req := <-s.recenter: - next = req // newer request supersedes the pending one - default: - // No newer request; process the one we have. - pending = false - s.processRecenter(ctx, next.cx, next.cz) - } - } else { - select { - case <-ctx.Done(): - return - case req := <-s.recenter: - pending = true - next = req + var req recenterReq + select { + case <-ctx.Done(): + return + case req = <-s.recenter: + } + for { + next, superseded := s.processRecenter(ctx, req.cx, req.cz) + if !superseded { + break } + req = next } } } // processRecenter generates and sends the chunks newly in range of (cx, cz), -// pre-generates the predictive ring, and drops chunks that left the gen radius. +// pre-generates the predictive ring, and drops chunks that left client view. // It is the only place `loaded`/`centerX`/`centerZ` are mutated. -func (s *streamer) processRecenter(ctx context.Context, cx, cz int32) { +func (s *streamer) processRecenter(ctx context.Context, cx, cz int32) (recenterReq, bool) { s.centerX, s.centerZ, s.hasCenter = cx, cz, true s.sendChunkCacheCenter(cx, cz) @@ -147,43 +140,97 @@ func (s *streamer) processRecenter(ctx context.Context, cx, cz int32) { // Build the desired set: everything within genRadius (the union of what we // send + the pre-gen ring). Sent = within viewRadius; pre-gen = the ring. order := spiralOrder(cx, cz, s.genRadius) - desired := make(map[[2]int32]bool, len(order)) + view := make(map[[2]int32]bool, (2*s.viewRadius+1)*(2*s.viewRadius+1)) // Split into "to send" (within viewRadius) and "pre-gen only" (the ring). var toSend [][2]int32 var toPreGen [][2]int32 + var viewTickets []world.ChunkPos + var prefetchTickets []world.ChunkPos for _, key := range order { - desired[key] = true - dx := key[0] - cx - if dx < 0 { - dx = -dx - } - dz := key[1] - cz - if dz < 0 { - dz = -dz - } - if dx <= int32(s.viewRadius) && dz <= int32(s.viewRadius) { + if chunkDistanceFrom(cx, cz, key) <= int32(s.viewRadius) { + view[key] = true toSend = append(toSend, key) + viewTickets = append(viewTickets, world.ChunkPos{X: key[0], Z: key[1]}) } else { toPreGen = append(toPreGen, key) + prefetchTickets = append(prefetchTickets, world.ChunkPos{X: key[0], Z: key[1]}) + } + } + if s.tickets != nil { + s.tickets.Replace(viewTickets, prefetchTickets) + } + + // Client residency follows viewRadius exactly. The prefetch ring is retained + // only server-side by tickets and never left loaded on the client. + for key := range s.loaded { + if !view[key] { + s.sendForgetLevelChunk(key[0], key[1]) + delete(s.loaded, key) } } - // Generate the send set in parallel, sending each frame as it completes. - // The pool guarantees `poolSize` concurrent cache.Frame calls; a single - // sender drains results and writes to the conn (serialized by writeMu). - s.parallelSend(ctx, toSend) + // Work is admitted in strict distance order. Each batch is at most poolSize, + // so a new recenter only waits for currently-running frames, not a full ring. + if next, superseded := s.streamPriority(ctx, cx, cz, toSend, true); superseded { + return next, true + } // Pre-generate the ring so the next recenter finds frames warm in the cache. - // Errors are irrelevant here (we don't send anything), so no sender. - s.parallelGenerate(ctx, toPreGen) + if next, superseded := s.streamPriority(ctx, cx, cz, toPreGen, false); superseded { + return next, true + } + return recenterReq{}, false +} - // Forget chunks that left the gen radius. The client drops them itself once - // it gets the new chunk-cache-center, but trimming our set keeps memory - // bounded and avoids re-sending. - for key := range s.loaded { - if !desired[key] { - s.sendForgetLevelChunk(key[0], key[1]) - delete(s.loaded, key) +func chunkDistanceFrom(cx, cz int32, key [2]int32) int32 { + dx := key[0] - cx + if dx < 0 { + dx = -dx + } + dz := key[1] - cz + if dz < 0 { + dz = -dz + } + if dz > dx { + return dz + } + return dx +} + +func (s *streamer) streamPriority(ctx context.Context, cx, cz int32, keys [][2]int32, send bool) (recenterReq, bool) { + for start := 0; start < len(keys); { + ring := chunkDistanceFrom(cx, cz, keys[start]) + end := start + for end < len(keys) && end-start < s.poolSize && chunkDistanceFrom(cx, cz, keys[end]) == ring { + end++ + } + if send { + s.parallelSend(ctx, keys[start:end]) + } else { + s.parallelGenerate(ctx, keys[start:end]) + } + if next, ok := s.latestRecenter(); ok { + return next, true + } + select { + case <-ctx.Done(): + return recenterReq{}, false + default: + } + start = end + } + return recenterReq{}, false +} + +func (s *streamer) latestRecenter() (recenterReq, bool) { + var latest recenterReq + found := false + for { + select { + case latest = <-s.recenter: + found = true + default: + return latest, found } } } @@ -261,9 +308,19 @@ func (s *streamer) parallelSend(ctx context.Context, keys [][2]int32) { if r.err != nil { // Send failed — the connection is likely closing. Bail out; the // serve loop will tear us down via ctx cancel. - s.log.Debug("streamer send failed", "cx", r.cx, "cz", r.cz, "err", r.err) + if s.log != nil { + s.log.Debug("streamer frame failed", "cx", r.cx, "cz", r.cz, "err", r.err) + } return } + if s.conn != nil { + if err := s.conn.SendFramed(r.frame); err != nil { + if s.log != nil { + s.log.Debug("streamer send failed", "cx", r.cx, "cz", r.cz, "err", err) + } + return + } + } s.loaded[[2]int32{r.cx, r.cz}] = true } } @@ -298,7 +355,7 @@ func (s *streamer) parallelGenerate(ctx context.Context, keys [][2]int32) { return default: } - _, _ = s.cache.FrameErr(j.cx, j.cz) // warm the cache; discard the frame + _, _ = s.cache.FrameErrContext(ctx, j.cx, j.cz) // warm cache; discard frame } }() } @@ -320,6 +377,7 @@ type frameJob struct{ cx, cz int32 } // frameResult is a generated chunk frame plus any send error. type frameResult struct { cx, cz int32 + frame []byte err error } @@ -332,7 +390,7 @@ func (s *streamer) generateWorker(ctx context.Context, jobs <-chan frameJob, res return default: } - frame, err := s.cache.FrameErr(j.cx, j.cz) + frame, err := s.cache.FrameErrContext(ctx, j.cx, j.cz) if err != nil { select { case results <- frameResult{cx: j.cx, cz: j.cz, err: err}: @@ -340,15 +398,8 @@ func (s *streamer) generateWorker(ctx context.Context, jobs <-chan frameJob, res } return } - if err := s.conn.SendFramed(frame); err != nil { - select { - case results <- frameResult{cx: j.cx, cz: j.cz, err: err}: - case <-ctx.Done(): - } - return - } select { - case results <- frameResult{cx: j.cx, cz: j.cz}: + case results <- frameResult{cx: j.cx, cz: j.cz, frame: frame}: case <-ctx.Done(): return } diff --git a/internal/network/streamer_test.go b/internal/network/streamer_test.go index e62f30f..65a73ef 100644 --- a/internal/network/streamer_test.go +++ b/internal/network/streamer_test.go @@ -1,8 +1,14 @@ package network import ( + "context" + "sync" + "sync/atomic" "testing" "time" + + "regionio/internal/protocol" + "regionio/internal/world" ) // TestSpiralOrderCenterFirst confirms the centre coordinate is returned first @@ -16,7 +22,7 @@ func TestSpiralOrderCenterFirst(t *testing.T) { // only needs to be non-decreasing within ring blocks. Verify the max // distance of the first k entries grows as expected by checking the full // set contains exactly the (2*2+1)² = 49 distinct coords. - want := (2*2 + 1) * (2 * 2 + 1) + want := (2*2 + 1) * (2*2 + 1) if len(order) != want { t.Errorf("spiralOrder len = %d, want %d", len(order), want) } @@ -82,3 +88,174 @@ func TestRequestRecenterNonBlocking(t *testing.T) { t.Fatal("requestRecenter blocked for 2s") } } + +func TestStreamerSendsStrictlyNearFirst(t *testing.T) { + cache := world.NewCache(-1, func(cx, cz int32) *world.Chunk { + return world.NewChunk(cx, cz, world.BiomePlains) + }) + recorder := &recordingConn{} + s := newStreamer(cache, NewConn(recorder), nil, 2) + s.genRadius = s.viewRadius + s.poolSize = 4 + if _, superseded := s.processRecenter(context.Background(), 7, -3); superseded { + t.Fatal("unexpected recenter supersession") + } + defer s.tickets.Close() + + lastDistance := int32(-1) + chunks := 0 + for _, packet := range recorder.take(t) { + if packet.ID != protocol.PlayLevelChunk { + continue + } + r := packet.Body() + x, err := r.Int32() + if err != nil { + t.Fatal(err) + } + z, err := r.Int32() + if err != nil { + t.Fatal(err) + } + distance := chunkDistanceFrom(7, -3, [2]int32{x, z}) + if distance < lastDistance { + t.Fatalf("chunk (%d,%d) at distance %d arrived after distance %d", x, z, distance, lastDistance) + } + lastDistance = distance + chunks++ + } + if chunks != 25 { + t.Fatalf("level chunks sent = %d, want 25", chunks) + } + if got := cache.Stats().Tickets; got != 25 { + t.Fatalf("tickets = %d, want 25", got) + } +} + +func TestStreamerQueuedRecenterStopsOldOuterRings(t *testing.T) { + cache := world.NewCache(-1, func(cx, cz int32) *world.Chunk { + return world.NewChunk(cx, cz, world.BiomePlains) + }) + s := newStreamer(cache, nil, nil, 2) + s.genRadius = s.viewRadius + s.poolSize = 4 + s.requestRecenter(100, 100) + next, superseded := s.processRecenter(context.Background(), 0, 0) + defer s.tickets.Close() + if !superseded || next != (recenterReq{cx: 100, cz: 100}) { + t.Fatalf("superseded=%v next=%+v, want latest (100,100)", superseded, next) + } + if len(s.loaded) != 1 || !s.loaded[[2]int32{0, 0}] { + t.Fatalf("old loaded set = %v, want only old center", s.loaded) + } +} + +func TestStreamerRecenterForgetsViewAndReplacesPrefetchTickets(t *testing.T) { + cache := world.NewCacheWithLimit(-1, func(cx, cz int32) *world.Chunk { + return world.NewChunk(cx, cz, world.BiomePlains) + }, nil, 9) + recorder := &recordingConn{} + s := newStreamer(cache, NewConn(recorder), nil, 2) + s.viewRadius, s.genRadius, s.poolSize = 0, 1, 4 + defer s.tickets.Close() + if _, superseded := s.processRecenter(context.Background(), 0, 0); superseded { + t.Fatal("unexpected first recenter supersession") + } + if len(s.loaded) != 1 || cache.Stats().Tickets != 9 { + t.Fatalf("first lifecycle loaded=%v stats=%+v", s.loaded, cache.Stats()) + } + recorder.take(t) + + if _, superseded := s.processRecenter(context.Background(), 10, 10); superseded { + t.Fatal("unexpected second recenter supersession") + } + if len(s.loaded) != 1 || !s.loaded[[2]int32{10, 10}] { + t.Fatalf("second loaded set = %v, want only (10,10)", s.loaded) + } + stats := cache.Stats() + if stats.Tickets != 9 || stats.Chunks > 9 || hasLevelChunkPacket(recorder.take(t), protocol.PlayForgetLevelChunk) != 1 { + t.Fatalf("second lifecycle stats=%+v; want 9 tickets, <=9 chunks and one forget", stats) + } +} + +func hasLevelChunkPacket(packets []protocol.Packet, id int32) int { + count := 0 + for _, packet := range packets { + if packet.ID == id { + count++ + } + } + return count +} + +func TestLoadSixteenClientStreamersBoundedAndReleasesTickets(t *testing.T) { + var active, peak atomic.Int32 + gen := func(cx, cz int32) *world.Chunk { + now := active.Add(1) + for { + old := peak.Load() + if now <= old || peak.CompareAndSwap(old, now) { + break + } + } + time.Sleep(200 * time.Microsecond) + active.Add(-1) + return world.NewChunk(cx, cz, world.BiomePlains) + } + cache := world.NewCacheWithLimit(-1, gen, nil, 32) + ctx, cancel := context.WithCancel(context.Background()) + var wg sync.WaitGroup + const clients = 16 + const groups = 4 + recorders := make([]*recordingConn, clients) + for i := 0; i < clients; i++ { + recorders[i] = &recordingConn{} + s := newStreamer(cache, NewConn(recorders[i]), nil, 2) + // A 3x3 view produces 144 ticket claims. Four spawn regions exercise + // both shared tickets and concurrent independent generation. + s.viewRadius, s.genRadius, s.poolSize = 1, 1, 4 + wg.Add(1) + go func(index int) { + defer wg.Done() + group := int32(index % groups) + s.requestRecenter(group*64, group*64) + s.run(ctx) + }(i) + } + + deadline := time.Now().Add(120 * time.Second) + for (cache.Stats().Frames < groups*9 || cache.Stats().Tickets < clients*9 || clientsWithPacket(recorders, protocol.PlayLevelChunk) < clients) && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + stats := cache.Stats() + if stats.Frames < groups*9 || stats.Tickets != clients*9 || clientsWithPacket(recorders, protocol.PlayLevelChunk) != clients { + cancel() + wg.Wait() + t.Fatalf("loaded stats = %+v, want at least %d frames and %d tickets", stats, groups*9, clients*9) + } + if got := peak.Load(); got > 8 { + cancel() + wg.Wait() + t.Fatalf("peak concurrent generators = %d, want <= shared limit 8", got) + } + + cancel() + wg.Wait() + deadline = time.Now().Add(5 * time.Second) + for cache.Stats().Tickets != 0 && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + if stats = cache.Stats(); stats.Tickets != 0 || stats.Chunks > 32 { + t.Fatalf("after disconnect stats = %+v, want zero tickets and <=32 chunks", stats) + } +} + +func clientsWithPacket(recorders []*recordingConn, id int32) int { + count := 0 + for _, recorder := range recorders { + if recorder.countPacketID(id) > 0 { + count++ + } + } + return count +} diff --git a/internal/world/cache.go b/internal/world/cache.go index 24b506f..62f0cf3 100644 --- a/internal/world/cache.go +++ b/internal/world/cache.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "log/slog" + "runtime" "sync" "time" @@ -38,11 +39,16 @@ type Cache struct { store *Store // nil = in-memory only (tests, flat worlds) maxChunks int // LRU capacity; 0 = unbounded - mu sync.Mutex - lightMu sync.Mutex - chunks map[[2]int32]*Chunk - frames map[[2]int32][]byte - dirty map[[2]int32]uint64 + mu sync.Mutex + lightMu sync.Mutex + chunks map[[2]int32]*Chunk + frames map[[2]int32][]byte + dirty map[[2]int32]uint64 + tickets map[[2]int32]int + inflight map[[2]int32]int + // frameSlots bounds expensive load/light/encode work across all players. + // Streamers may have their own workers, but they share this admission gate. + frameSlots chan struct{} // LRU bookkeeping: order is MRU(front)→LRU(back); index gives O(1) lookup. order *list.List // elements are *[2]int32; nil when maxChunks==0 index map[[2]int32]*list.Element @@ -61,13 +67,23 @@ type chunkLoad struct { // threshold using gen to produce missing chunks. It has no persistence and no // eviction limit (unbounded; for tests/flat worlds). func NewCache(threshold int32, gen Generator) *Cache { + workers := runtime.GOMAXPROCS(0) + if workers > 8 { + workers = 8 + } + if workers < 2 { + workers = 2 + } c := &Cache{ - threshold: threshold, - gen: gen, - chunks: make(map[[2]int32]*Chunk), - frames: make(map[[2]int32][]byte), - dirty: make(map[[2]int32]uint64), - loads: make(map[[2]int32]*chunkLoad), + threshold: threshold, + gen: gen, + chunks: make(map[[2]int32]*Chunk), + frames: make(map[[2]int32][]byte), + dirty: make(map[[2]int32]uint64), + tickets: make(map[[2]int32]int), + inflight: make(map[[2]int32]int), + loads: make(map[[2]int32]*chunkLoad), + frameSlots: make(chan struct{}, workers), } return c } @@ -124,6 +140,11 @@ func (c *Cache) evictIfNeeded() { return } key := *back.Value.(*[2]int32) + if c.tickets[key] > 0 || c.inflight[key] > 0 { + c.order.MoveToFront(back) + checked++ + continue + } // Never drop a dirty chunk: it has unsaved edits. Bump it to MRU and // stop evicting this cycle; the autosave flush will clear it and the // next eviction pass can reclaim it. @@ -209,6 +230,25 @@ func (c *Cache) Frame(cx, cz int32) []byte { // FrameErr returns a framed chunk packet while preserving storage failures. // Callers serving clients should prefer it to Frame so corruption is observable. func (c *Cache) FrameErr(cx, cz int32) ([]byte, error) { + return c.FrameErrContext(context.Background(), cx, cz) +} + +// FrameErrContext is FrameErr with cancellable admission to the shared frame +// worker budget. Cancellation prevents obsolete streamer jobs from starting; +// an operation already admitted completes so cache state is never half-built. +func (c *Cache) FrameErrContext(ctx context.Context, cx, cz int32) ([]byte, error) { + select { + case c.frameSlots <- struct{}{}: + defer func() { <-c.frameSlots }() + case <-ctx.Done(): + return nil, ctx.Err() + } + release := c.beginUse([2]int32{cx, cz}) + defer release() + return c.frameErr(cx, cz) +} + +func (c *Cache) frameErr(cx, cz int32) ([]byte, error) { key := [2]int32{cx, cz} for { @@ -258,6 +298,8 @@ func (c *Cache) GetBlock(x, y, z int) uint16 { } cx := int32(x >> 4) cz := int32(z >> 4) + release := c.beginUse([2]int32{cx, cz}) + defer release() ch, err := c.chunkAtErr(cx, cz) if err != nil { return StateAir @@ -267,6 +309,8 @@ func (c *Cache) GetBlock(x, y, z int) uint16 { // LightUpdate returns the standalone light_update body for a loaded chunk. func (c *Cache) LightUpdate(cx, cz int32) ([]byte, error) { + release := c.beginUse([2]int32{cx, cz}) + defer release() ch, err := c.chunkAtErr(cx, cz) if err != nil { return nil, err @@ -298,6 +342,8 @@ func (c *Cache) SetBlockWithLight(x, y, z int, state uint16) (bool, []ChunkPos) } cx := int32(x >> 4) cz := int32(z >> 4) + release := c.beginUse([2]int32{cx, cz}) + defer release() ch, err := c.chunkAtErr(cx, cz) if err != nil { return false, nil @@ -321,6 +367,7 @@ func (c *Cache) SetBlockWithLight(x, y, z int, state uint16) (bool, []ChunkPos) // its light from the authoritative blocks. ch.mu.Lock() ch.lightReady = false + ch.lightValidated = false ch.mu.Unlock() lightChanged = []ChunkPos{{X: cx, Z: cz}} } diff --git a/internal/world/cache_light.go b/internal/world/cache_light.go index 3bcf900..350383f 100644 --- a/internal/world/cache_light.go +++ b/internal/world/cache_light.go @@ -18,7 +18,7 @@ func (c *Cache) ensureLight(chunk *Chunk) error { func (c *Cache) ensureLightLocked(chunk *Chunk) error { for { center, revision := chunk.snapshot() - if center.lightReady { + if center.lightReady && center.lightValidated { return nil } @@ -48,11 +48,18 @@ func (c *Cache) ensureLightLocked(chunk *Chunk) error { chunk.mu.Unlock() continue } - chunk.installLight(volume) + changed := chunk.installLight(volume) + if center.lightReady && changed { + revision = chunk.revision.Add(1) + } chunk.mu.Unlock() c.mu.Lock() - delete(c.frames, [2]int32{chunk.X, chunk.Z}) + key := [2]int32{chunk.X, chunk.Z} + delete(c.frames, key) + if c.store != nil && center.lightReady && changed { + c.dirty[key] = revision + } c.mu.Unlock() return nil } diff --git a/internal/world/cache_tickets.go b/internal/world/cache_tickets.go new file mode 100644 index 0000000..a800304 --- /dev/null +++ b/internal/world/cache_tickets.go @@ -0,0 +1,124 @@ +package world + +import "sync" + +func (c *Cache) beginUse(key [2]int32) func() { + c.mu.Lock() + c.inflight[key]++ + c.mu.Unlock() + return func() { + c.mu.Lock() + c.inflight[key]-- + if c.inflight[key] <= 0 { + delete(c.inflight, key) + } + c.evictIfNeeded() + c.mu.Unlock() + } +} + +// TicketLevel describes why a streamer keeps a chunk resident. View tickets +// are client-visible; Prefetch tickets retain the predictive outer ring. +type TicketLevel uint8 + +const ( + TicketView TicketLevel = iota + TicketPrefetch +) + +// TicketSet is one owner's atomic chunk-residency claim. A streamer replaces +// the complete set on recenter and closes it on disconnect. Multiple sets may +// overlap; a chunk becomes evictable only after its final owner releases it. +type TicketSet struct { + mu sync.Mutex + cache *Cache + held map[[2]int32]TicketLevel + closed bool +} + +// NewTicketSet creates an empty ticket set associated with this cache. +func (c *Cache) NewTicketSet() *TicketSet { + return &TicketSet{cache: c, held: make(map[[2]int32]TicketLevel)} +} + +// Replace atomically changes the ticket set. View entries take precedence when +// a coordinate is present in both slices. +func (t *TicketSet) Replace(view, prefetch []ChunkPos) { + if t == nil || t.cache == nil { + return + } + desired := make(map[[2]int32]TicketLevel, len(view)+len(prefetch)) + for _, pos := range prefetch { + desired[[2]int32{pos.X, pos.Z}] = TicketPrefetch + } + for _, pos := range view { + desired[[2]int32{pos.X, pos.Z}] = TicketView + } + + t.mu.Lock() + defer t.mu.Unlock() + if t.closed { + return + } + t.cache.mu.Lock() + for key := range t.held { + if _, keep := desired[key]; keep { + continue + } + t.cache.tickets[key]-- + if t.cache.tickets[key] <= 0 { + delete(t.cache.tickets, key) + } + } + for key := range desired { + if _, alreadyHeld := t.held[key]; !alreadyHeld { + t.cache.tickets[key]++ + } + } + t.held = desired + t.cache.evictIfNeeded() + t.cache.mu.Unlock() +} + +// Close releases every ticket. It is safe to call more than once. +func (t *TicketSet) Close() { + if t == nil || t.cache == nil { + return + } + t.mu.Lock() + defer t.mu.Unlock() + if t.closed { + return + } + t.cache.mu.Lock() + for key := range t.held { + t.cache.tickets[key]-- + if t.cache.tickets[key] <= 0 { + delete(t.cache.tickets, key) + } + } + clear(t.held) + t.closed = true + t.cache.evictIfNeeded() + t.cache.mu.Unlock() +} + +// CacheStats is a concurrency-safe lifecycle snapshot used by diagnostics and +// load tests. +type CacheStats struct { + Chunks int + Frames int + TicketedChunks int + Tickets int +} + +// Stats returns current cache and ticket counts. +func (c *Cache) Stats() CacheStats { + c.mu.Lock() + defer c.mu.Unlock() + stats := CacheStats{Chunks: len(c.chunks), Frames: len(c.frames), TicketedChunks: len(c.tickets)} + for _, count := range c.tickets { + stats.Tickets += count + } + return stats +} diff --git a/internal/world/cache_tickets_test.go b/internal/world/cache_tickets_test.go new file mode 100644 index 0000000..8145fea --- /dev/null +++ b/internal/world/cache_tickets_test.go @@ -0,0 +1,78 @@ +package world + +import ( + "context" + "errors" + "testing" +) + +func TestTicketPinsChunkUntilRelease(t *testing.T) { + cache := NewCacheWithLimit(-1, func(cx, cz int32) *Chunk { + return NewChunk(cx, cz, BiomePlains) + }, nil, 2) + tickets := cache.NewTicketSet() + tickets.Replace([]ChunkPos{{X: 0, Z: 0}}, nil) + for x := int32(0); x < 4; x++ { + if _, err := cache.FrameErr(x, 0); err != nil { + t.Fatal(err) + } + } + if !hasChunk(cache, 0, 0) { + t.Fatal("ticketed chunk was evicted") + } + if got := cache.Stats().Tickets; got != 1 { + t.Fatalf("ticket count = %d, want 1", got) + } + + tickets.Close() + if got := cache.Stats(); got.Tickets != 0 || got.Chunks > 2 { + t.Fatalf("after release stats = %+v, want no tickets and <=2 chunks", got) + } +} + +func TestOverlappingTicketSetsRequireFinalRelease(t *testing.T) { + cache := NewCacheWithLimit(-1, func(cx, cz int32) *Chunk { + return NewChunk(cx, cz, BiomePlains) + }, nil, 1) + first, second := cache.NewTicketSet(), cache.NewTicketSet() + pos := []ChunkPos{{X: 0, Z: 0}} + first.Replace(pos, nil) + second.Replace(nil, pos) + if _, err := cache.FrameErr(0, 0); err != nil { + t.Fatal(err) + } + first.Close() + if got := cache.Stats().Tickets; got != 1 { + t.Fatalf("tickets after first close = %d, want 1", got) + } + if _, err := cache.FrameErr(1, 0); err != nil { + t.Fatal(err) + } + if !hasChunk(cache, 0, 0) { + t.Fatal("chunk was evicted while second owner still held it") + } + + second.Close() + second.Close() // idempotent + if got := cache.Stats(); got.Tickets != 0 || got.Chunks > 1 { + t.Fatalf("after final close stats = %+v", got) + } +} + +func TestFrameAdmissionCanBeCancelled(t *testing.T) { + cache := NewCache(-1, func(cx, cz int32) *Chunk { + return NewChunk(cx, cz, BiomePlains) + }) + for i := 0; i < cap(cache.frameSlots); i++ { + cache.frameSlots <- struct{}{} + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + _, err := cache.FrameErrContext(ctx, 0, 0) + if !errors.Is(err, context.Canceled) { + t.Fatalf("FrameErrContext error = %v, want context.Canceled", err) + } + for i := 0; i < cap(cache.frameSlots); i++ { + <-cache.frameSlots + } +} diff --git a/internal/world/encode.go b/internal/world/encode.go index a316313..fc50a5a 100644 --- a/internal/world/encode.go +++ b/internal/world/encode.go @@ -77,7 +77,10 @@ type Chunk struct { skyLight [SectionCount]*[2048]byte blockLight [SectionCount]*[2048]byte lightReady bool - biome uint16 // fallback uniform biome when biomes[si] is nil + // lightValidated is runtime-only. Persisted arrays are ready to read but are + // reconciled with current neighbor blocks once after entering a live cache. + lightValidated bool + biome uint16 // fallback uniform biome when biomes[si] is nil } // NewChunk returns an empty (all-air) chunk at (x, z) with the given biome. @@ -140,6 +143,7 @@ func (c *Chunk) SetBlock(lx, y, lz int, state uint16) { if changed { c.mu.Lock() c.lightReady = false + c.lightValidated = false c.mu.Unlock() } } @@ -243,6 +247,7 @@ func (c *Chunk) snapshot() (*Chunk, uint64) { } } clone.lightReady = c.lightReady + clone.lightValidated = c.lightValidated return clone, revision } diff --git a/internal/world/light.go b/internal/world/light.go index b686ccd..b7e6782 100644 --- a/internal/world/light.go +++ b/internal/world/light.go @@ -331,6 +331,7 @@ func (c *Chunk) installLight(v *lightVolume) bool { c.skyLight = sky c.blockLight = block c.lightReady = true + c.lightValidated = true return changed } diff --git a/internal/world/store_test.go b/internal/world/store_test.go index a59418a..4b1abbd 100644 --- a/internal/world/store_test.go +++ b/internal/world/store_test.go @@ -185,6 +185,85 @@ func TestStoreLightRoundTrip(t *testing.T) { } } +func TestCacheReconcilesPersistedLightWithLoadedNeighbor(t *testing.T) { + dir := t.TempDir() + store, err := NewStore(dir) + if err != nil { + t.Fatal(err) + } + glowstone := nameToStateID("minecraft:glowstone", nil) + left := NewChunk(0, 0, BiomePlains) + left.SetBlock(15, 100, 8, glowstone) + if err := store.SaveChunk(left); err != nil { + t.Fatal(err) + } + // Simulate a chunk saved before its unloaded neighbor gained a light source. + right := NewChunk(1, 0, BiomePlains) + right.lightReady = true + if err := store.SaveChunk(right); err != nil { + t.Fatal(err) + } + if err := store.Close(); err != nil { + t.Fatal(err) + } + + store, err = NewStore(dir) + if err != nil { + t.Fatal(err) + } + defer store.Close() + cache := NewCacheWithLimit(-1, func(cx, cz int32) *Chunk { + return NewChunk(cx, cz, BiomePlains) + }, store, 1) + tickets := cache.NewTicketSet() + tickets.Replace([]ChunkPos{{X: 1, Z: 0}}, nil) + loaded, err := cache.chunkAtErr(1, 0) + if err != nil { + t.Fatal(err) + } + if _, block, ready := loaded.LightAt(0, 100, 8); !ready || block != 0 { + t.Fatalf("persisted pre-reconcile light = %d ready=%v, want stale zero", block, ready) + } + if _, err := cache.FrameErr(1, 0); err != nil { + t.Fatal(err) + } + if _, block, ready := loaded.LightAt(0, 100, 8); !ready || block != 14 { + t.Fatalf("reconciled border light = %d ready=%v, want 14", block, ready) + } + if err := cache.SaveAll(); err != nil { + t.Fatal(err) + } + tickets.Close() + if _, err := cache.FrameErr(10, 0); err != nil { + t.Fatal(err) + } + if hasChunk(cache, 1, 0) { + t.Fatal("released light chunk remained resident after LRU replacement") + } + reloadedAfterEviction, err := cache.chunkAtErr(1, 0) + if err != nil { + t.Fatal(err) + } + if _, block, ready := reloadedAfterEviction.LightAt(0, 100, 8); !ready || block != 14 { + t.Fatalf("light after ticket unload/reload = %d ready=%v, want 14", block, ready) + } + if err := store.Close(); err != nil { + t.Fatal(err) + } + store, err = NewStore(dir) + if err != nil { + t.Fatal(err) + } + defer store.Close() + reloaded, err := store.LoadChunk(1, 0) + if err != nil { + t.Fatal(err) + } + if _, block, ready := reloaded.LightAt(0, 100, 8); !ready || block != 14 { + t.Fatalf("persisted reconciled light = %d ready=%v, want 14", block, ready) + } +} + // TestStoreSaveLoadIntegration is the end-to-end "world survives restart" test: // generate a chunk via a store-backed cache, edit a block, SaveAll, then open a // fresh cache over the same store and confirm the edit is present.