From 28c11e2fc229b45b39dfdd1768901805ad0e7ab0 Mon Sep 17 00:00:00 2001 From: Daniar Mannanov Date: Mon, 10 Aug 2026 23:00:20 +0300 Subject: [PATCH] Bound connection outbound queues --- internal/network/conn.go | 150 ++++++++++++++++++++++++---- internal/network/conn_queue_test.go | 113 +++++++++++++++++++++ internal/network/play.go | 2 +- 3 files changed, 246 insertions(+), 19 deletions(-) create mode 100644 internal/network/conn_queue_test.go diff --git a/internal/network/conn.go b/internal/network/conn.go index b24d143..e924a94 100644 --- a/internal/network/conn.go +++ b/internal/network/conn.go @@ -4,6 +4,7 @@ package network import ( "bufio" + "errors" "io" "net" "sync" @@ -13,7 +14,18 @@ import ( "regionio/internal/server" ) -const networkWriteTimeout = 30 * time.Second +const ( + networkWriteTimeout = 30 * time.Second + outboundQueueSize = 256 + maxOutboundBytes = 8 * 1024 * 1024 +) + +var ErrOutboundFull = errors.New("network: outbound queue full") + +type outboundPacket struct { + frame []byte + done chan error +} // Conn wraps a TCP connection with buffered reads and tracks protocol state. type Conn struct { @@ -27,19 +39,27 @@ type Conn struct { // Profile is populated once the client identifies during login. Profile server.Profile - // writeMu serializes writes so a background sender (e.g. keep-alive) and - // the read-loop handler never interleave bytes on the wire. - writeMu sync.Mutex + outbound chan outboundPacket + done chan struct{} + writerDone chan struct{} + closeOnce sync.Once + outboundMu sync.Mutex + pendingBytes int } // NewConn wraps a raw TCP connection. func NewConn(raw net.Conn) *Conn { - return &Conn{ + c := &Conn{ raw: raw, br: bufio.NewReaderSize(raw, 4096), state: protocol.StateHandshaking, compressionThreshold: -1, + outbound: make(chan outboundPacket, outboundQueueSize), + done: make(chan struct{}), + writerDone: make(chan struct{}), } + go c.writeLoop() + return c } // State returns the current protocol state. @@ -68,16 +88,18 @@ func (c *Conn) ReadPacket() (protocol.Packet, error) { return pkt, err } -// Send writes a packet with the given ID and pre-encoded body. Safe for -// concurrent use. +// Send queues a packet and waits until the connection's sole writer has put it +// on the wire. Waiting preserves state-machine ordering while keeping all socket +// writes serialized through the bounded outbound queue. func (c *Conn) Send(id int32, body []byte) error { - c.writeMu.Lock() - defer c.writeMu.Unlock() - if err := c.raw.SetWriteDeadline(time.Now().Add(networkWriteTimeout)); err != nil { - return err - } - defer c.raw.SetWriteDeadline(time.Time{}) - return protocol.WritePacket(c.raw, c.compressionThreshold, id, body) + return c.sendFrame(protocol.AppendPacket(nil, c.compressionThreshold, id, body), true) +} + +// Enqueue adds a packet without waiting for the socket. Server broadcasts use +// it so one slow recipient cannot stall the world or another connection. A +// client that exceeds the bounded queue is disconnected. +func (c *Conn) Enqueue(id int32, body []byte) error { + return c.sendFrame(protocol.AppendPacket(nil, c.compressionThreshold, id, body), false) } // SendWriter writes a packet whose body was built with a protocol.Writer. @@ -89,8 +111,73 @@ func (c *Conn) SendWriter(id int32, w *protocol.Writer) error { // The frame must have been built for this connection's compression threshold. // Safe for concurrent use. func (c *Conn) SendFramed(frame []byte) error { - c.writeMu.Lock() - defer c.writeMu.Unlock() + return c.sendFrame(frame, true) +} + +func (c *Conn) sendFrame(frame []byte, wait bool) error { + packet := outboundPacket{frame: frame} + if wait { + packet.done = make(chan error, 1) + } + c.outboundMu.Lock() + if c.pendingBytes+len(frame) > maxOutboundBytes { + c.outboundMu.Unlock() + c.abort() + return ErrOutboundFull + } + select { + case <-c.done: + c.outboundMu.Unlock() + return net.ErrClosed + case c.outbound <- packet: + c.pendingBytes += len(frame) + c.outboundMu.Unlock() + default: + c.outboundMu.Unlock() + c.abort() + return ErrOutboundFull + } + if packet.done == nil { + return nil + } + select { + case err := <-packet.done: + return err + case <-c.done: + select { + case err := <-packet.done: + return err + default: + return net.ErrClosed + } + } +} + +func (c *Conn) writeLoop() { + defer close(c.writerDone) + for { + select { + case packet := <-c.outbound: + err := c.writeFrame(packet.frame) + c.outboundMu.Lock() + c.pendingBytes -= len(packet.frame) + c.outboundMu.Unlock() + if packet.done != nil { + packet.done <- err + } + if err != nil { + c.abort() + c.failPending(err) + return + } + case <-c.done: + c.failPending(net.ErrClosed) + return + } + } +} + +func (c *Conn) writeFrame(frame []byte) error { if err := c.raw.SetWriteDeadline(time.Now().Add(networkWriteTimeout)); err != nil { return err } @@ -108,8 +195,35 @@ func (c *Conn) SendFramed(frame []byte) error { return nil } +func (c *Conn) failPending(err error) { + for { + select { + case packet := <-c.outbound: + c.outboundMu.Lock() + c.pendingBytes -= len(packet.frame) + c.outboundMu.Unlock() + if packet.done != nil { + packet.done <- err + } + default: + return + } + } +} + +func (c *Conn) abort() { + c.closeOnce.Do(func() { + close(c.done) + _ = c.raw.Close() + }) +} + // CompressionThreshold returns the active threshold (-1 if disabled). func (c *Conn) CompressionThreshold() int32 { return c.compressionThreshold } -// Close closes the underlying connection. -func (c *Conn) Close() error { return c.raw.Close() } +// Close stops the writer and closes the underlying connection. +func (c *Conn) Close() error { + c.abort() + <-c.writerDone + return nil +} diff --git a/internal/network/conn_queue_test.go b/internal/network/conn_queue_test.go new file mode 100644 index 0000000..0e81846 --- /dev/null +++ b/internal/network/conn_queue_test.go @@ -0,0 +1,113 @@ +package network + +import ( + "bufio" + "bytes" + "errors" + "io" + "net" + "sync" + "testing" + "time" + + "regionio/internal/protocol" +) + +type blockedConn struct { + mu sync.Mutex + closed bool + release chan struct{} + once sync.Once +} + +func newBlockedConn() *blockedConn { return &blockedConn{release: make(chan struct{})} } + +func (c *blockedConn) Read([]byte) (int, error) { return 0, io.EOF } +func (c *blockedConn) Write(p []byte) (int, error) { + <-c.release + c.mu.Lock() + defer c.mu.Unlock() + if c.closed { + return 0, net.ErrClosed + } + return len(p), nil +} +func (c *blockedConn) Close() error { + c.mu.Lock() + c.closed = true + c.mu.Unlock() + c.once.Do(func() { close(c.release) }) + return nil +} +func (c *blockedConn) LocalAddr() net.Addr { return testAddr("local") } +func (c *blockedConn) RemoteAddr() net.Addr { return testAddr("remote") } +func (c *blockedConn) SetDeadline(time.Time) error { return nil } +func (c *blockedConn) SetReadDeadline(time.Time) error { return nil } +func (c *blockedConn) SetWriteDeadline(time.Time) error { return nil } + +func TestConnWriterPreservesPacketOrder(t *testing.T) { + raw := &recordingConn{} + conn := NewConn(raw) + defer conn.Close() + for id := int32(1); id <= 3; id++ { + if err := conn.Send(id, []byte{byte(id)}); err != nil { + t.Fatal(err) + } + } + raw.mu.Lock() + data := append([]byte(nil), raw.buf.Bytes()...) + raw.mu.Unlock() + reader := bufio.NewReader(bytes.NewReader(data)) + for want := int32(1); want <= 3; want++ { + packet, err := protocol.ReadPacket(reader, -1) + if err != nil { + t.Fatal(err) + } + if packet.ID != want || len(packet.Data) != 1 || packet.Data[0] != byte(want) { + t.Fatalf("packet %d = id %d data %v", want, packet.ID, packet.Data) + } + } +} + +func TestConnEnqueueDisconnectsAtByteBudget(t *testing.T) { + raw := newBlockedConn() + conn := NewConn(raw) + body := make([]byte, protocol.MaxPacketSize) + var err error + for i := 0; i < 8; i++ { + err = conn.Enqueue(1, body) + if err != nil { + break + } + } + if !errors.Is(err, ErrOutboundFull) { + t.Fatalf("enqueue error = %v, want ErrOutboundFull", err) + } + select { + case <-conn.done: + case <-time.After(time.Second): + t.Fatal("queue overflow did not close connection") + } + if err := conn.Close(); err != nil { + t.Fatal(err) + } +} + +func TestConnCloseUnblocksWaitingSend(t *testing.T) { + raw := newBlockedConn() + conn := NewConn(raw) + result := make(chan error, 1) + go func() { result <- conn.Send(1, []byte("blocked")) }() + time.Sleep(10 * time.Millisecond) + if err := conn.Close(); err != nil { + t.Fatal(err) + } + select { + case err := <-result: + if err == nil { + t.Fatal("blocked send succeeded after close") + } + case <-time.After(time.Second): + t.Fatal("Close did not unblock Send") + } +} diff --git a/internal/network/play.go b/internal/network/play.go index 5a49973..5ee8927 100644 --- a/internal/network/play.go +++ b/internal/network/play.go @@ -25,7 +25,7 @@ const ( // 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 { - session, err := h.srv.RegisterPlayer(h.conn.Profile, h.conn.Send) + session, err := h.srv.RegisterPlayer(h.conn.Profile, h.conn.Enqueue) if err != nil { return err }