diff --git a/internal/network/configuration.go b/internal/network/configuration.go index 10032f2..bf18423 100644 --- a/internal/network/configuration.go +++ b/internal/network/configuration.go @@ -106,8 +106,20 @@ func (h *handler) handleClientInformation(pkt protocol.Packet) error { return err } + // view_distance drives the chunk streamer radius. The client sends 2..32; + // clamp into a sane server range so a huge view distance doesn't trigger a + // generation explosion. + vdInt := int(int8(vd)) + if vdInt < 2 { + vdInt = 2 + } + if vdInt > 16 { + vdInt = 16 + } + h.viewDistance = vdInt + h.log.Info("client information", - "locale", locale, "view_distance", int8(vd), + "locale", locale, "view_distance", vdInt, "chat_mode", chatMode, "main_hand", mainHand) return nil } diff --git a/internal/network/handler.go b/internal/network/handler.go index 2726c10..35409ba 100644 --- a/internal/network/handler.go +++ b/internal/network/handler.go @@ -1,6 +1,7 @@ package network import ( + "context" "errors" "io" "log/slog" @@ -11,28 +12,38 @@ import ( ) // handler drives one connection through its state machine until it closes. -// It is owned by a single goroutine (the read loop), so the play fields below -// need no synchronization. +// It is owned by a single goroutine (the read loop); play-phase chunk streaming +// is delegated to a background streamer goroutine. type handler struct { conn *Conn srv *server.Server log *slog.Logger - // Play-phase chunk streaming state. - loaded map[[2]int32]bool // chunks currently sent to the client - centerX int32 - centerZ int32 - hasCenter bool + // ctx is the connection lifetime context, set in serve(). It cancels when + // the read loop ends, stopping the streamer and any derived work. + ctx context.Context + + // Background chunk streamer (Play phase). requestRecenter pushes here. + streamer *streamer + // viewDistance is the client's requested view distance (from + // client_information), clamped; used to size the streamer. + viewDistance int // Creative inventory state for block placement. heldSlot int32 // selected hotbar index (0-8) hotbar [9]int32 // item network IDs per hotbar slot (-1 = empty) } -// serve runs the read/dispatch loop for a single connection. +// serve runs the read/dispatch loop for a single connection. It owns the +// streamer's lifecycle: the streamer context is cancelled when this loop exits, +// stopping generation and sending so no goroutine outlives the connection. func (h *handler) serve() { defer h.conn.Close() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() // stops the streamer when the read loop ends + h.ctx = ctx + for { pkt, err := h.conn.ReadPacket() if err != nil { @@ -49,7 +60,8 @@ func (h *handler) serve() { } } -// dispatch routes a packet to the handler for the current state. +// dispatch routes a packet to the handler for the current state. The Play +// handler reads h.ctx (the connection lifetime context) to launch the streamer. func (h *handler) dispatch(pkt protocol.Packet) error { switch h.conn.State() { case protocol.StateHandshaking: diff --git a/internal/network/play.go b/internal/network/play.go index 740ff20..d203501 100644 --- a/internal/network/play.go +++ b/internal/network/play.go @@ -18,13 +18,9 @@ const ( spawnZ = 8.5 ) -// chunkRadius is how many chunks around the player we send (a square of side -// 2*radius+1). Kept modest until streaming by view distance exists. -const chunkRadius = 4 - // beginPlay sends the join sequence once the client enters the Play phase and -// starts the keep-alive loop. In milestone 4a no chunks are sent, so the client -// reaches the "loading terrain" screen and waits. +// hands chunk streaming off to the background streamer. The streamer stops when +// h.ctx (the connection lifetime context) is cancelled. func (h *handler) beginPlay() error { for i := range h.hotbar { h.hotbar[i] = -1 // empty @@ -40,56 +36,24 @@ func (h *handler) beginPlay() error { if err := h.sendPlayerPosition(1); err != nil { return err } - if err := h.streamAround(0, 0); err != nil { - return err - } + // Launch the background chunk streamer. It owns generation + sending so the + // read loop stays free; requestRecenter is a non-blocking push. + h.streamer = newStreamer(h.srv.Chunks(), h.conn, h.log, h.viewDistance) + go h.streamer.run(h.ctx) + h.streamer.requestRecenter(0, 0) go h.keepAliveLoop() return nil } -// streamAround recenters the client's chunk cache on (centerX, centerZ) and -// sends the chunks newly in range. Chunks that fall out of range are dropped by -// the client automatically once it receives the new cache center, so we only -// send the difference and track the currently-loaded set. -func (h *handler) streamAround(centerX, centerZ int32) error { - cc := protocol.NewWriter(8) - cc.VarInt(centerX) - cc.VarInt(centerZ) - if err := h.conn.SendWriter(protocol.PlayChunkCacheCenter, cc); err != nil { - return err - } - - cache := h.srv.Chunks() - next := make(map[[2]int32]bool, (2*chunkRadius+1)*(2*chunkRadius+1)) - sent := 0 - for cx := centerX - chunkRadius; cx <= centerX+chunkRadius; cx++ { - for cz := centerZ - chunkRadius; cz <= centerZ+chunkRadius; cz++ { - key := [2]int32{cx, cz} - next[key] = true - if h.loaded[key] { - continue - } - if err := h.conn.SendFramed(cache.Frame(cx, cz)); err != nil { - return err - } - sent++ - } - } - h.loaded = next - h.centerX, h.centerZ, h.hasCenter = centerX, centerZ, true - h.log.Debug("streamed chunks", "center_x", centerX, "center_z", centerZ, "new", sent) - return nil -} - -// onPlayerMove recenters chunk streaming when the player crosses into a new -// chunk. X and Z are the player's block-precise coordinates. +// onPlayerMove recenters the streamer when the player crosses into a new chunk. +// It is a non-blocking push; the read loop never waits on generation. func (h *handler) onPlayerMove(x, z float64) error { cx := int32(int64(math.Floor(x)) >> 4) cz := int32(int64(math.Floor(z)) >> 4) - if h.hasCenter && cx == h.centerX && cz == h.centerZ { - return nil + if h.streamer != nil { + h.streamer.requestRecenter(cx, cz) } - return h.streamAround(cx, cz) + return nil } // sendPlayLogin writes the clientbound play "login" packet. Field layout was diff --git a/internal/network/streamer.go b/internal/network/streamer.go new file mode 100644 index 0000000..9559382 --- /dev/null +++ b/internal/network/streamer.go @@ -0,0 +1,348 @@ +package network + +import ( + "context" + "log/slog" + "runtime" + "sync" + + "regionio/internal/world" +) + +// 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. +// +// Ownership: +// - read loop: calls requestRecenter (non-blocking), owns nothing else here. +// - streamer goroutine: owns `loaded`, `centerX/Z`, the pool, and the sender. +// - conn write mutex: serializes every SendFramed (keep-alive, chunk frames, +// block updates all go through it). + +// recenterReq is a request to recenter streaming on a new chunk coordinate. +type recenterReq struct{ cx, cz int32 } + +// streamer streams chunks to one connection in the background. +type streamer struct { + cache *world.Cache + conn *Conn + log *slog.Logger + + recenter chan recenterReq + + // The loaded-set and current center are owned solely by the streamer's run + // goroutine — no other goroutine reads or writes them. + loaded map[[2]int32]bool + centerX int32 + centerZ int32 + hasCenter bool + + 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 +} + +// defaultViewRadius is used when the client hasn't sent client_information or +// sent an implausible value. Matches the legacy chunkRadius. +const defaultViewRadius = 4 + +// newStreamer constructs a streamer for the given cache/conn. viewDistance comes +// from the client's client_information (clamped to a safe range); genRadius is +// one ring wider so movement into fresh territory finds ready chunks. +func newStreamer(cache *world.Cache, conn *Conn, log *slog.Logger, viewDistance int) *streamer { + if viewDistance < 2 { + viewDistance = defaultViewRadius + } + if viewDistance > 16 { + viewDistance = 16 + } + pool := runtime.NumCPU() + if pool > 8 { + pool = 8 + } + if pool < 2 { + pool = 2 + } + return &streamer{ + cache: cache, + conn: conn, + log: log, + recenter: make(chan recenterReq, 4), + loaded: make(map[[2]int32]bool), + viewRadius: viewDistance, + genRadius: viewDistance + 1, + poolSize: pool, + } +} + +// requestRecenter asks the streamer to recenter on (cx, cz). Non-blocking: if +// the streamer is busy, the latest request wins (buffered channel drains the +// stale ones on next select). +func (s *streamer) requestRecenter(cx, cz int32) { + for { + select { + case s.recenter <- recenterReq{cx, cz}: + return + default: + // Channel full: a previous request is still queued. Drop it so the + // newest recenter is what the streamer acts on next. + select { + case <-s.recenter: + default: + // Another goroutine drained it concurrently; retry the send. + continue + } + } + } +} + +// run is the streamer's main loop. It blocks until ctx is cancelled (on +// 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 + ) + 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 + } + } + } +} + +// 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. +// It is the only place `loaded`/`centerX`/`centerZ` are mutated. +func (s *streamer) processRecenter(ctx context.Context, cx, cz int32) { + s.centerX, s.centerZ, s.hasCenter = cx, cz, true + + // 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)) + + // Split into "to send" (within viewRadius) and "pre-gen only" (the ring). + var toSend [][2]int32 + var toPreGen [][2]int32 + 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) { + toSend = append(toSend, key) + } else { + toPreGen = append(toPreGen, 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) + // 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) + + // 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] { + delete(s.loaded, key) + } + } +} + +// parallelSend generates the given chunks across the worker pool and sends each +// frame as soon as it is ready (order is best-effort; the client reassembles). +// Already-loaded chunks are skipped. Returns when all are sent or ctx cancels. +func (s *streamer) parallelSend(ctx context.Context, keys [][2]int32) { + var pending []frameJob + for _, k := range keys { + if s.loaded[k] { + continue + } + pending = append(pending, frameJob{k[0], k[1]}) + } + if len(pending) == 0 { + return + } + + jobs := make(chan frameJob, len(pending)) + results := make(chan frameResult, len(pending)) + + var wg sync.WaitGroup + workers := s.poolSize + if workers > len(pending) { + workers = len(pending) + } + for w := 0; w < workers; w++ { + wg.Add(1) + go func() { + defer wg.Done() + s.generateWorker(ctx, jobs, results) + }() + } + + // Feed the jobs. + go func() { + for _, j := range pending { + select { + case <-ctx.Done(): + close(jobs) + return + case jobs <- j: + } + } + close(jobs) + }() + + // Sender: drain results serially so writes don't interleave. + go func() { + wg.Wait() + close(results) + }() + for r := range results { + 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) + return + } + s.loaded[[2]int32{r.cx, r.cz}] = true + } +} + +// parallelGenerate warms the cache for the given chunks without sending them +// (used for the predictive ring). Errors are ignored. +func (s *streamer) parallelGenerate(ctx context.Context, keys [][2]int32) { + var pending []frameJob + for _, k := range keys { + if s.loaded[k] { + continue + } + pending = append(pending, frameJob{k[0], k[1]}) + } + if len(pending) == 0 { + return + } + + jobs := make(chan frameJob, len(pending)) + var wg sync.WaitGroup + workers := s.poolSize + if workers > len(pending) { + workers = len(pending) + } + for w := 0; w < workers; w++ { + wg.Add(1) + go func() { + defer wg.Done() + for j := range jobs { + select { + case <-ctx.Done(): + return + default: + } + _ = s.cache.Frame(j.cx, j.cz) // warm the cache; discard the frame + } + }() + } + for _, j := range pending { + select { + case <-ctx.Done(): + break + case jobs <- j: + } + } + close(jobs) + wg.Wait() +} + +// frameJob is one chunk coordinate awaiting generation. +type frameJob struct{ cx, cz int32 } + +// frameResult is a generated chunk frame plus any send error. +type frameResult struct { + cx, cz int32 + err error +} + +// generateWorker reads jobs, generates+frames the chunk via the (thread-safe) +// cache, and sends the frame to the conn. It exits when jobs closes. +func (s *streamer) generateWorker(ctx context.Context, jobs <-chan frameJob, results chan<- frameResult) { + for j := range jobs { + select { + case <-ctx.Done(): + return + default: + } + frame := s.cache.Frame(j.cx, j.cz) + 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 <-ctx.Done(): + return + } + } +} + +// spiralOrder returns chunk coordinates in a square of side (2*radius+1) around +// (cx, cz), ordered from the centre outward (Chebyshev rings). The centre is +// first, then ring 1, ring 2, … ring `radius`. Within a ring the order is +// deterministic but not otherwise constrained — nearest-first is what matters. +func spiralOrder(cx, cz int32, radius int) [][2]int32 { + if radius < 0 { + radius = 0 + } + out := make([][2]int32, 0, (2*radius+1)*(2*radius+1)) + out = append(out, [2]int32{cx, cz}) + for r := 1; r <= radius; r++ { + // Walk the perimeter of the ring at Chebyshev distance r. + for d := -r; d <= r; d++ { + out = append(out, [2]int32{cx + int32(d), cz - int32(r)}) // top edge + out = append(out, [2]int32{cx + int32(d), cz + int32(r)}) // bottom edge + } + for d := -r + 1; d <= r-1; d++ { + out = append(out, [2]int32{cx - int32(r), cz + int32(d)}) // left edge + out = append(out, [2]int32{cx + int32(r), cz + int32(d)}) // right edge + } + } + return out +} diff --git a/internal/network/streamer_test.go b/internal/network/streamer_test.go new file mode 100644 index 0000000..e62f30f --- /dev/null +++ b/internal/network/streamer_test.go @@ -0,0 +1,84 @@ +package network + +import ( + "testing" + "time" +) + +// TestSpiralOrderCenterFirst confirms the centre coordinate is returned first +// and the ring order expands outward (Chebyshev distance non-decreasing). +func TestSpiralOrderCenterFirst(t *testing.T) { + order := spiralOrder(0, 0, 2) + if order[0] != [2]int32{0, 0} { + t.Errorf("first = %v, want centre (0,0)", order[0]) + } + // Each entry's Chebyshev distance must not exceed the next's... actually it + // 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) + if len(order) != want { + t.Errorf("spiralOrder len = %d, want %d", len(order), want) + } + seen := make(map[[2]int32]bool, len(order)) + for _, c := range order { + if seen[c] { + t.Errorf("duplicate %v in spiral order", c) + } + seen[c] = true + } +} + +// TestSpiralOrderRingStructure checks that all distance-0 coords come before +// distance-1, which come before distance-2 (centre-outward ordering). +func TestSpiralOrderRingStructure(t *testing.T) { + order := spiralOrder(5, -3, 2) + cheb := func(c [2]int32) int32 { + dx := c[0] - 5 + if dx < 0 { + dx = -dx + } + dz := c[1] - -3 + if dz < 0 { + dz = -dz + } + if dx > dz { + return dx + } + return dz + } + // Track the max ring seen so far; it must never decrease (centre-first). + var maxRing int32 + for _, c := range order { + r := cheb(c) + if r < maxRing { + t.Errorf("ring %d appeared after ring %d — not centre-first", r, maxRing) + } + if r > maxRing { + maxRing = r + } + } + if maxRing != 2 { + t.Errorf("max ring = %d, want 2", maxRing) + } +} + +// TestRequestRecenterNonBlocking confirms requestRecenter never blocks the +// caller even when many requests are pushed rapidly (the streamer drains stale +// ones). This is the property the read loop relies on to stay responsive. +func TestRequestRecenterNonBlocking(t *testing.T) { + s := newStreamer(nil, nil, nil, 4) + done := make(chan struct{}) + go func() { + for i := 0; i < 1000; i++ { + s.requestRecenter(int32(i), int32(i)) + } + close(done) + }() + select { + case <-done: + // good: 1000 rapid requests returned without blocking + case <-time.After(2 * time.Second): + t.Fatal("requestRecenter blocked for 2s") + } +}