aboutsummaryrefslogtreecommitdiff
path: root/internal/net
diff options
context:
space:
mode:
Diffstat (limited to 'internal/net')
-rw-r--r--internal/net/conn.go10
-rw-r--r--internal/net/server.go202
-rw-r--r--internal/net/server_test.go11
-rw-r--r--internal/net/web.go1
-rw-r--r--internal/net/wsconn.go8
5 files changed, 206 insertions, 26 deletions
diff --git a/internal/net/conn.go b/internal/net/conn.go
index 2fff0bc..6d78d8f 100644
--- a/internal/net/conn.go
+++ b/internal/net/conn.go
@@ -6,12 +6,18 @@ import (
"io"
"net"
"strings"
+ "time"
)
type Conn interface {
ReadMessage() (string, error)
io.Writer
io.Closer
+ // SetWriteDeadline is optional for transports: sessions that support it
+ // enforce per-write deadlines so a stalled peer surfaces as a timeout
+ // instead of a hung writer goroutine. Implementations without deadline
+ // support return nil (deadlines are best-effort).
+ SetWriteDeadline(t time.Time) error
SetEcho(bool) error
}
@@ -70,6 +76,10 @@ func (t *tcpConn) Write(b []byte) (int, error) {
return t.conn.Write(b)
}
+func (t *tcpConn) SetWriteDeadline(ts time.Time) error {
+ return t.conn.SetWriteDeadline(ts)
+}
+
func (t *tcpConn) Close() error {
return t.conn.Close()
}
diff --git a/internal/net/server.go b/internal/net/server.go
index 9b5be18..b0407fc 100644
--- a/internal/net/server.go
+++ b/internal/net/server.go
@@ -1,6 +1,7 @@
package net
import (
+ "context"
"crypto/rand"
"crypto/rsa"
"crypto/tls"
@@ -73,6 +74,19 @@ type Session struct {
promptVisible bool
writeMu sync.Mutex
+ // Outbound queueing. Transport sessions (telnet/TLS/websocket) are put
+ // into queued mode by startWriter before their input goroutine begins:
+ // Write* then enqueues into outbox, and the dedicated writer goroutine
+ // performs the network I/O with a per-message write deadline. A stalled
+ // or slow client therefore blocks at most its own queue, never the
+ // world; an overflowing queue disconnects only that client. Sessions
+ // without a writer (tests, synthetic conns) keep the legacy synchronous
+ // path so message construction is shared and behavior identical.
+ outbox chan string
+ writerOnce sync.Once
+ closeOnce sync.Once
+ writeDone chan struct{}
+
// seq is a monotonically increasing sequence number assigned when the
// session is added to a Hub. Hub.AllSessions and Hub.PlayersInRoom sort
// sessions by seq so per-tick consumers iterate players in a stable,
@@ -97,6 +111,10 @@ type Server struct {
httpsTLSConfig *tls.Config
hub *Hub
handler func(*Session, string)
+
+ done chan struct{}
+ shutdownOnce sync.Once
+ listenErr error
}
type Hub struct {
@@ -107,6 +125,64 @@ type Hub struct {
onDisconnect func(*Session)
}
+// writeQueueBounds bound the outbound queue: maxQueuedMessages before a
+// client is disconnected for falling behind, writeDeadline per message, and
+// closeFlushTimeout to drain the queue when the session is closing.
+const (
+ maxQueuedMessages = 512
+ writeDeadline = 5 * time.Second
+ closeFlushTimeout = 2 * time.Second
+)
+
+// startWriter puts the session into queued mode: writes are enqueued and a
+// dedicated goroutine drains the queue to the network with per-message write
+// deadlines. The goroutine closes the transport after the queue drains, so a
+// Close() during shutdown flushes pending output before the socket drops.
+// Must be called before the session's input goroutine starts.
+func (sess *Session) StartWriter() {
+ sess.writerOnce.Do(func() {
+ sess.outbox = make(chan string, maxQueuedMessages)
+ sess.writeDone = make(chan struct{})
+ go func() {
+ defer close(sess.writeDone)
+ for msg := range sess.outbox {
+ if dl, ok := sess.Conn.(interface {
+ SetWriteDeadline(time.Time) error
+ }); ok {
+ _ = dl.SetWriteDeadline(time.Now().Add(writeDeadline))
+ }
+ if _, err := sess.Conn.Write([]byte(msg)); err != nil {
+ log.Printf("session write error: %v", err)
+ // Abandon the queue and drop the transport; the input
+ // goroutine notices the closed connection on its next
+ // read and performs normal removal.
+ sess.Conn.Close()
+ return
+ }
+ }
+ // Queue closed and drained: flush any final bytes and close.
+ if dl, ok := sess.Conn.(interface {
+ SetWriteDeadline(time.Time) error
+ }); ok {
+ _ = dl.SetWriteDeadline(time.Now().Add(closeFlushTimeout))
+ }
+ sess.Conn.Close()
+ }()
+ })
+}
+
+// enqueue appends msg to the outbound queue. When the client's queue is full
+// the session is disconnected instead of blocking the caller: the world must
+// never wait on a single slow client.
+func (sess *Session) enqueue(msg string) {
+ select {
+ case sess.outbox <- msg:
+ default:
+ log.Printf("session write queue overflow; disconnecting slow client")
+ sess.Close()
+ }
+}
+
func NewHub() *Hub {
return &Hub{
sessions: make(map[*Session]bool),
@@ -145,24 +221,34 @@ func (h *Hub) Remove(s *Session) {
}
return
}
- h.hardRemoveLocked(s)
+ callback := h.hardRemoveLocked(s)
h.mu.Unlock()
+ // The removal callback runs outside the Hub mutex: it performs saves
+ // and game-state cleanup, and must never serialize behind (or hold up)
+ // Hub membership operations.
+ if callback != nil {
+ callback(s)
+ }
}
func (h *Hub) HardRemove(s *Session) {
h.mu.Lock()
- defer h.mu.Unlock()
- h.hardRemoveLocked(s)
+ callback := h.hardRemoveLocked(s)
+ h.mu.Unlock()
+ if callback != nil {
+ callback(s)
+ }
}
-func (h *Hub) hardRemoveLocked(s *Session) {
+// hardRemoveLocked removes the session from all membership structures and
+// returns the removal callback for the caller to invoke AFTER releasing the
+// Hub mutex (nil when none is registered).
+func (h *Hub) hardRemoveLocked(s *Session) func(*Session) {
delete(h.sessions, s)
for _, room := range h.rooms {
delete(room, s)
}
- if h.onRemove != nil {
- h.onRemove(s)
- }
+ return h.onRemove
}
func (h *Hub) EnterRoom(s *Session, roomID int) {
@@ -187,6 +273,28 @@ func (h *Hub) leaveRoomLocked(s *Session) {
}
}
+// RekeyRooms remaps Hub room membership buckets per idMap (oldID -> newID).
+// Uses delete-all-then-set-all so a swap (A<->B) cannot clobber an entry
+// mid-rekey. Sessions' own player state is migrated by the game layer; this
+// only moves the membership buckets.
+func (h *Hub) RekeyRooms(idMap map[int]int) {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ if len(idMap) == 0 {
+ return
+ }
+ moved := make(map[int]map[*Session]bool, len(h.rooms))
+ for id, set := range h.rooms {
+ if newID, ok := idMap[id]; ok && newID != id {
+ moved[newID] = set
+ delete(h.rooms, id)
+ }
+ }
+ for newID, set := range moved {
+ h.rooms[newID] = set
+ }
+}
+
func (h *Hub) AllSessions() []*Session {
h.mu.Lock()
defer h.mu.Unlock()
@@ -215,6 +323,7 @@ func NewServer(cfg *config.Config) (*Server, error) {
s := &Server{
config: cfg,
hub: NewHub(),
+ done: make(chan struct{}),
}
if cfg.Telnet.Enabled {
@@ -337,7 +446,38 @@ func (s *Server) ListenAndServe(handler func(*Session, string)) error {
return fmt.Errorf("no listeners configured")
}
- select {}
+ // Block until Shutdown is called (or a fatal listener error occurs).
+ <-s.done
+ return s.listenErr
+}
+
+// Shutdown stops all transports: HTTP(S) servers drain, listeners close, and
+// every session is closed (its outbound queue drains first). ListenAndServe
+// returns after Shutdown completes. It is safe to call more than once.
+func (s *Server) Shutdown() {
+ s.shutdownOnce.Do(func() {
+ if s.httpServer != nil {
+ ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+ defer cancel()
+ _ = s.httpServer.Shutdown(ctx)
+ }
+ if s.httpsServer != nil {
+ ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+ defer cancel()
+ _ = s.httpsServer.Shutdown(ctx)
+ }
+ if s.telnetLn != nil {
+ s.telnetLn.Close()
+ }
+ if s.telnetTLSLn != nil {
+ s.telnetTLSLn.Close()
+ }
+ for _, sess := range s.hub.AllSessions() {
+ sess.Close()
+ }
+ close(s.done)
+ })
+ <-s.done
}
func (s *Server) serveTelnet() {
@@ -399,7 +539,7 @@ func (s *Server) handleSession(sess *Session, handler func(*Session, string)) {
sess.Close()
}()
- sess.Conn.Write([]byte("\033[2J\033[H"))
+ sess.Write("\033[2J\033[H")
sess.Write(WelcomeBanner())
sess.Write("What's your account name? ")
@@ -439,15 +579,26 @@ func (sess *Session) ClearPrompt() {
sess.promptVisible = false
}
+// dispatch delivers a fully-formatted message to the transport: enqueued in
+// queued mode (the writer goroutine owns the network), written synchronously
+// in direct mode (tests, synthetic conns). Caller holds writeMu.
+func (sess *Session) dispatch(msg string) {
+ if sess.outbox != nil {
+ sess.enqueue(msg)
+ return
+ }
+ sess.Conn.Write([]byte(msg))
+}
+
// Write sends raw text with no wrapping. Used for prompts and partial lines.
func (sess *Session) Write(msg string) {
sess.writeMu.Lock()
defer sess.writeMu.Unlock()
if sess.promptVisible {
- sess.Conn.Write([]byte("\r\n"))
+ sess.dispatch("\r\n")
sess.promptVisible = false
}
- sess.Conn.Write([]byte(msg))
+ sess.dispatch(msg)
}
func (sess *Session) WriteLine(msg string) {
@@ -455,10 +606,10 @@ func (sess *Session) WriteLine(msg string) {
sess.writeMu.Lock()
defer sess.writeMu.Unlock()
if sess.promptVisible {
- sess.Conn.Write([]byte("\r\n"))
+ sess.dispatch("\r\n")
sess.promptVisible = false
}
- sess.Conn.Write([]byte(msg + "\r\n"))
+ sess.dispatch(msg + "\r\n")
}
func (sess *Session) WriteLines(lines ...string) {
@@ -466,10 +617,10 @@ func (sess *Session) WriteLines(lines ...string) {
defer sess.writeMu.Unlock()
for i, l := range lines {
if i == 0 && sess.promptVisible {
- sess.Conn.Write([]byte("\r\n"))
+ sess.dispatch("\r\n")
sess.promptVisible = false
}
- sess.Conn.Write([]byte(sess.wrap(l) + "\r\n"))
+ sess.dispatch(sess.wrap(l) + "\r\n")
}
}
@@ -478,10 +629,10 @@ func (sess *Session) Writef(format string, args ...interface{}) {
sess.writeMu.Lock()
defer sess.writeMu.Unlock()
if sess.promptVisible {
- sess.Conn.Write([]byte("\r\n"))
+ sess.dispatch("\r\n")
sess.promptVisible = false
}
- sess.Conn.Write([]byte(msg))
+ sess.dispatch(msg)
}
func (sess *Session) WritePrompt(prompt string) {
@@ -490,19 +641,26 @@ func (sess *Session) WritePrompt(prompt string) {
if sess.promptVisible {
return
}
- sess.Conn.Write([]byte(prompt))
+ sess.dispatch(prompt)
sess.promptVisible = true
}
func (sess *Session) Reprompt(prompt string) {
sess.writeMu.Lock()
defer sess.writeMu.Unlock()
- sess.Conn.Write([]byte(prompt))
+ sess.dispatch(prompt)
sess.promptVisible = true
}
+// Close stops the session. In queued mode it closes the queue so the writer
+// goroutine drains pending messages and then closes the transport; in direct
+// mode it closes the transport immediately. Idempotent.
func (sess *Session) Close() error {
- sess.writeMu.Lock()
- defer sess.writeMu.Unlock()
- return sess.Conn.Close()
+ if sess.outbox != nil {
+ sess.closeOnce.Do(func() { close(sess.outbox) })
+ return nil
+ }
+ var err error
+ sess.closeOnce.Do(func() { err = sess.Conn.Close() })
+ return err
}
diff --git a/internal/net/server_test.go b/internal/net/server_test.go
index b106c53..c91c701 100644
--- a/internal/net/server_test.go
+++ b/internal/net/server_test.go
@@ -3,6 +3,7 @@ package net
import (
"sync"
"testing"
+ "time"
)
func TestHubConcurrency(t *testing.T) {
@@ -38,10 +39,12 @@ func TestHubConcurrency(t *testing.T) {
type testConn struct{ data []byte }
-func (c *testConn) ReadMessage() (string, error) { return "", nil }
-func (c *testConn) Write(b []byte) (int, error) { c.data = append(c.data, b...); return len(b), nil }
-func (c *testConn) Close() error { return nil }
-func (c *testConn) SetEcho(bool) error { return nil }
+func (c *testConn) ReadMessage() (string, error) { return "", nil }
+func (c *testConn) Write(b []byte) (int, error) { c.data = append(c.data, b...); return len(b), nil }
+func (c *testConn) SetWriteDeadline(time.Time) error { return nil }
+
+func (c *testConn) Close() error { return nil }
+func (c *testConn) SetEcho(bool) error { return nil }
func TestSessionWriteConcurrency(t *testing.T) {
conn := &testConn{}
diff --git a/internal/net/web.go b/internal/net/web.go
index e573aba..233928f 100644
--- a/internal/net/web.go
+++ b/internal/net/web.go
@@ -64,6 +64,7 @@ func newHTTPMux(s *Server) *http.ServeMux {
Conn: newWSConn(ws),
State: StateAccountName,
}
+ sess.StartWriter()
s.hub.Add(sess)
go s.handleSession(sess, s.handler)
})
diff --git a/internal/net/wsconn.go b/internal/net/wsconn.go
index 4990c0b..34663ba 100644
--- a/internal/net/wsconn.go
+++ b/internal/net/wsconn.go
@@ -1,6 +1,8 @@
package net
import (
+ "time"
+
"github.com/gorilla/websocket"
)
@@ -25,6 +27,12 @@ func (w *wsConn) Write(b []byte) (int, error) {
return len(b), err
}
+// SetWriteDeadline delegates to the underlying websocket connection, which
+// supports deadlines natively.
+func (w *wsConn) SetWriteDeadline(ts time.Time) error {
+ return w.conn.SetWriteDeadline(ts)
+}
+
func (w *wsConn) Close() error {
return w.conn.Close()
}
generated by cgit v1.2.3 (git 2.25.1) at 2026-10-24 12:13:30 +0000