remo/RegionIO
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Background predictive chunk streaming

Browse source 
Chunk generation and sending no longer block the read loop. The read
loop pushes a non-blocking recenter request and stays free to handle
movement, chat, and keep-alive acks immediately; a per-connection
streamer goroutine generates chunks in a worker pool and sends them
serially under the write mutex.

- network/streamer.go: per-connection streamer. spiralOrder emits
  chunks centre-outward; parallelSend fans Cache.Frame across a worker
  pool (Cache.Frame is already goroutine-safe), parallelGenerate warms
  a one-ring predictive border so movement finds ready chunks; the
  loaded-set is owned solely by the streamer.
- network/play.go: beginPlay launches the streamer and pushes an
  initial recenter instead of the old blocking streamAround;
  onPlayerMove now just calls requestRecenter (non-blocking).
- network/handler.go: ctx (connection lifetime) + streamer field; the
  streamer stops when the read loop ends (cancel on serve exit).
- network/configuration.go: client view_distance is saved (clamped
  2..16) and drives the streamer radius.
- Tests: spiral order (centre-first, ring structure) and the
  non-blocking recenter guarantee the read loop relies on.
This commit is contained in:
Master290 2026-06-25 00:54:57 +03:00
parent d1cc29bb60
commit d7c80a8858
5 changed files with 478 additions and 58 deletions
Download patch file Download diff file Expand all files Collapse all files

14
internal/network/configuration.go
View file

@ -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
}

30
internal/network/handler.go
View file

@ -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:

60
internal/network/play.go
View file

@ -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

348
internal/network/streamer.go Normal file
View file

@ -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
}

84
internal/network/streamer_test.go Normal file
View file

@ -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")
}
}