diff options
Diffstat (limited to 'internal/net')
| -rw-r--r-- | internal/net/conn.go | 17 | ||||
| -rw-r--r-- | internal/net/server.go | 108 | ||||
| -rw-r--r-- | internal/net/session_close_test.go | 73 | ||||
| -rw-r--r-- | internal/net/wsconn.go | 14 |
4 files changed, 183 insertions, 29 deletions
diff --git a/internal/net/conn.go b/internal/net/conn.go index 6d78d8f..a7c53ec 100644 --- a/internal/net/conn.go +++ b/internal/net/conn.go @@ -92,6 +92,23 @@ func (t *tcpConn) SetEcho(on bool) error { return t.writeTelnetCmd(will, echo) } +// EchoSequence returns the telnet negotiation bytes that switch the client's +// echo state. Queued sessions transmit these through the session writer so a +// stalled client cannot block the caller (see Session.SetEcho); the local +// echo state is applied separately via SetEchoState. +func (t *tcpConn) EchoSequence(on bool) []byte { + if on { + return []byte{iac, wont, echo} + } + return []byte{iac, will, echo} +} + +// SetEchoState records the echo state without writing to the network; used +// when the negotiation bytes are queued instead of written inline. +func (t *tcpConn) SetEchoState(on bool) { + t.echoOff = !on +} + const ( iac = 255 will = 251 diff --git a/internal/net/server.go b/internal/net/server.go index adc828c..dd65f6e 100644 --- a/internal/net/server.go +++ b/internal/net/server.go @@ -89,7 +89,12 @@ type Session struct { // 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 + outbox chan string + // closeCh is closed by Close() to tell the writer goroutine to drain the + // queue and then close the transport. It is ONLY ever closed, never sent + // to: enqueues select on it to drop late writes, so a close can never + // race a concurrent send the way closing outbox itself did. + closeCh chan struct{} writerOnce sync.Once closeOnce sync.Once writeDone chan struct{} @@ -143,16 +148,18 @@ const ( // 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. +// deadlines. The goroutine exits after Close() — draining whatever was +// queued first — and closes the transport, 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.closeCh = make(chan struct{}) sess.writeDone = make(chan struct{}) go func() { defer close(sess.writeDone) - for msg := range sess.outbox { + writeOne := func(msg string) bool { if dl, ok := sess.Conn.(interface { SetWriteDeadline(time.Time) error }); ok { @@ -164,32 +171,52 @@ func (sess *Session) StartWriter() { // goroutine notices the closed connection on its next // read and performs normal removal. sess.Conn.Close() - return + return false } + return true } - // 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)) + for { + select { + case msg := <-sess.outbox: + if !writeOne(msg) { + return + } + case <-sess.closeCh: + // Drain whatever was queued before the close, then + // flush any final bytes and close the transport. + for { + select { + case msg := <-sess.outbox: + if !writeOne(msg) { + return + } + default: + if dl, ok := sess.Conn.(interface { + SetWriteDeadline(time.Time) error + }); ok { + _ = dl.SetWriteDeadline(time.Now().Add(closeFlushTimeout)) + } + sess.Conn.Close() + return + } + } + } } - 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. -// 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. +// never wait on a single slow client. Writes racing a Close() are dropped, +// never panicked on: closeCh is only ever closed (never sent to), and outbox +// itself is never closed, so no close/send race exists. func (sess *Session) enqueue(msg string) { - // A broadcast can race Close(): the queue may already be (or become) - // closed mid-send. Such late writes are dropped, never panicked on — - // Close() closes the queue so the writer goroutine drains whatever was - // queued before the close and then flushes the transport. - defer func() { _ = recover() }() + select { + case <-sess.closeCh: + return + default: + } select { case sess.outbox <- msg: default: @@ -628,6 +655,31 @@ func (sess *Session) dispatch(msg string) { sess.Conn.Write([]byte(msg)) } +// echoController is implemented by transports whose echo negotiation can be +// expressed as bytes: queued sessions send those bytes through the session +// writer so a stalled client cannot block the caller (the game owner), +// while the local echo state is applied immediately via SetEchoState. +type echoController interface { + EchoSequence(on bool) []byte + SetEchoState(on bool) +} + +// SetEcho toggles client echo. Queued sessions transmit the negotiation +// through the outbound queue (keeping prompt bookkeeping untouched); direct +// sessions and transports without byte-expressible negotiation write inline. +func (sess *Session) SetEcho(on bool) { + if sess.outbox != nil { + if ec, ok := sess.Conn.(echoController); ok { + ec.SetEchoState(on) + sess.writeMu.Lock() + defer sess.writeMu.Unlock() + sess.dispatch(string(ec.EchoSequence(on))) + return + } + } + _ = sess.Conn.SetEcho(on) +} + // Write sends raw text with no wrapping. Used for prompts and partial lines. func (sess *Session) Write(msg string) { sess.writeMu.Lock() @@ -690,15 +742,17 @@ func (sess *Session) Reprompt(prompt string) { 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 +// Close stops the session. In queued mode it signals the writer goroutine, +// which drains pending messages and then closes the transport; in direct // mode it closes the transport immediately. Idempotent. func (sess *Session) Close() error { - if sess.outbox != nil { - sess.closeOnce.Do(func() { close(sess.outbox) }) - return nil - } var err error - sess.closeOnce.Do(func() { err = sess.Conn.Close() }) + sess.closeOnce.Do(func() { + if sess.closeCh != nil { + close(sess.closeCh) + return + } + err = sess.Conn.Close() + }) return err } diff --git a/internal/net/session_close_test.go b/internal/net/session_close_test.go new file mode 100644 index 0000000..7cb54b6 --- /dev/null +++ b/internal/net/session_close_test.go @@ -0,0 +1,73 @@ +package net + +// Close/broadcast race regression: a broadcast racing Session.Close() used +// to send on the outbound channel while Close closed the same channel — a +// genuine data race (closechan vs chansend) that the old panic-recover only +// hid. Closing is now signaled through a separate channel that is only ever +// closed, never sent to. + +import ( + stdnet "net" + "sync" + "testing" + "time" +) + +// newTestSocketPair returns a synchronous, unbuffered connection pair: a +// write blocks until the peer reads, which is exactly the stall condition +// the echo test has to reproduce. +func newTestSocketPair() (stdnet.Conn, stdnet.Conn) { return stdnet.Pipe() } + +func testTimeout() <-chan time.Time { return time.After(200 * time.Millisecond) } + +func TestConcurrentWriteCloseIsRaceFree(t *testing.T) { + for i := 0; i < 50; i++ { + sess := &Session{Conn: &testConn{}} + sess.StartWriter() + var wg sync.WaitGroup + wg.Add(2) + start := make(chan struct{}) + go func() { + defer wg.Done() + <-start + for j := 0; j < 100; j++ { + sess.WriteLine("broadcast") + } + }() + go func() { + defer wg.Done() + <-start + sess.Close() + }() + close(start) + wg.Wait() + <-sess.writeDone + } +} + +// Echo negotiation regression: login flips telnet echo on the game owner's +// goroutine. Writing the negotiation inline made that write a direct socket +// I/O, so a client that never read (full receive window, stalled socket) +// blocked the entire game loop behind one connection. Queued sessions must +// route the negotiation through the outbound writer instead. +func TestSetEchoQueuedDoesNotBlockCaller(t *testing.T) { + local, peer := newTestSocketPair() + defer local.Close() + defer peer.Close() + sess := &Session{Conn: newTCPConn(local)} + sess.StartWriter() + defer sess.Close() + + done := make(chan struct{}) + go func() { + sess.SetEcho(false) // what the login path calls for password entry + close(done) + }() + select { + case <-done: + case <-testTimeout(): + peer.Close() // unblock the writer so the test can exit + <-done + t.Fatal("SetEcho blocked the caller instead of queueing the negotiation") + } +} diff --git a/internal/net/wsconn.go b/internal/net/wsconn.go index 34663ba..29d1947 100644 --- a/internal/net/wsconn.go +++ b/internal/net/wsconn.go @@ -38,8 +38,18 @@ func (w *wsConn) Close() error { } func (w *wsConn) SetEcho(on bool) error { + return w.conn.WriteMessage(websocket.TextMessage, w.EchoSequence(on)) +} + +// EchoSequence returns the browser echo-control bytes for the websocket +// transport; queued sessions send these through the session writer so a +// stalled browser cannot block the caller (see Session.SetEcho). +func (w *wsConn) EchoSequence(on bool) []byte { if on { - return w.conn.WriteMessage(websocket.TextMessage, []byte("\x1b]10;0\x07")) + return []byte("\x1b]10;0\x07") } - return w.conn.WriteMessage(websocket.TextMessage, []byte("\x1b]10;1\x07")) + return []byte("\x1b]10;1\x07") } + +// SetEchoState is a no-op: the websocket transport keeps no local echo state. +func (w *wsConn) SetEchoState(bool) {} |