diff --git a/README.md b/README.md index 4ae338ea..6471ea1e 100644 --- a/README.md +++ b/README.md @@ -274,7 +274,28 @@ type Session interface { - **`Conn()`**: Get the underlying network connection object - **`IsClosed()`**: Check if the session is closed - **`Close()`**: Close the session connection -- **`Reset()`**: Reset session state +- **`Reset()`**: Close and reset the session after managed work and connection cleanup finish. If any managed callback of this session is active at entry, Reset schedules cleanup and returns immediately, including calls from external goroutines. Otherwise it waits. The signature is unchanged, but completion semantics change while callbacks are active. +- **`TryReset()`** (optional `TryResetter`): Return `ErrResetDuringCallback` without requesting reset if any callback of this session is active at entry; otherwise wait and return the reset result. +- **`ResetAsync()`** (optional `AsyncResetter`): Request cleanup without blocking. Each call returns a separate buffered channel that receives one error (`nil` on success) and then closes. Pending requests share one cleanup operation, but later resets cannot overwrite their results. + +`Session` keeps its original `Reset()` signature. The optional interfaces do not require existing custom Session implementations to add methods. Stop starting new operations before waiting for completion or reusing the session. Never wait for reset inside a callback of the session being reset; request it and return instead: + +```go +func (l *listener) OnClose(s getty.Session) { + async, ok := s.(getty.AsyncResetter) + if !ok { + return + } + done := async.ResetAsync() + go func() { + if err := <-done; err != nil { + return // Reset failed; do not recycle the session. + } + // Session cleanup and reset completed successfully. + }() + // Return so the reset worker can finish waiting for this callback. +} +``` ##### Configuration Settings - **`SetName(string)`**: Set session name diff --git a/README_CN.md b/README_CN.md index 2d5bce45..7a3ca9cd 100644 --- a/README_CN.md +++ b/README_CN.md @@ -274,7 +274,28 @@ type Session interface { - **`Conn()`**: 获取底层网络连接对象 - **`IsClosed()`**: 检查会话是否已关闭 - **`Close()`**: 关闭会话连接 -- **`Reset()`**: 重置会话状态 +- **`Reset()`**: 关闭会话，等待受管理工作和连接清理结束后重置状态。如果调用时本 Session 有任何受管理回调正在执行，则发起重置后立即返回，外部 goroutine 的调用也一样；否则同步等待。原有签名保持不变，但回调活跃期间的完成语义发生变化。 +- **`TryReset()`**（可选 `TryResetter`）：如果调用时本 Session 有回调正在执行，返回 `ErrResetDuringCallback`，不发起重置；否则等待并返回本次重置结果。 +- **`ResetAsync()`**（可选 `AsyncResetter`）：不阻塞地请求清理。每次调用返回独立的缓冲 channel，发送一次 error（成功为 `nil`）后关闭。并发未完成请求共享一次清理，但结果不会被后续重置覆盖。 + +`Session` 保持原有 `Reset()` 签名，可选接口无需现有自定义 Session 实现新增方法。在等待完成或复用会话前，应停止发起新操作。不能在被重置 Session 的回调内等待完成，应发起请求后返回： + +```go +func (l *listener) OnClose(s getty.Session) { + async, ok := s.(getty.AsyncResetter) + if !ok { + return + } + done := async.ResetAsync() + go func() { + if err := <-done; err != nil { + return // Reset failed; do not recycle the session. + } + // 会话清理和重置成功完成，现在可以回收。 + }() + // 返回，让重置任务能够结束对本回调的等待。 +} +``` ##### 配置设置 - **`SetName(string)`**: 设置会话名称 diff --git a/transport/client_test.go b/transport/client_test.go index 9043a06f..f84bfbee 100644 --- a/transport/client_test.go +++ b/transport/client_test.go @@ -1053,7 +1053,11 @@ func TestUDPClient(t *testing.T) { clt.Close() assert.True(t, clt.IsClosed()) - msgHandler.array[0].Reset() + async, ok := msgHandler.array[0].(AsyncResetter) + if !ok { + t.Fatal("built-in session does not implement AsyncResetter") + } + waitForReset(t, async.ResetAsync()) assert.Nil(t, msgHandler.array[0].Conn()) // ss.WritePkg([]byte("hello"), 0) } diff --git a/transport/getty.go b/transport/getty.go index ccedb367..c1eecbf1 100644 --- a/transport/getty.go +++ b/transport/getty.go @@ -24,9 +24,13 @@ import ( ) var ( - ErrSessionClosed = perrors.New("session Already Closed") - ErrSessionBlocked = perrors.New("session Full Blocked") - ErrNullPeerAddr = perrors.New("peer address is nil") + // ErrResetDuringCallback is returned by TryResetter.TryReset when it is + // called while any managed callback of the target session is active. Use ResetAsync and + // observe completion outside the callback instead. + ErrResetDuringCallback = perrors.New("cannot synchronously reset a session while a callback is active") + ErrSessionClosed = perrors.New("session Already Closed") + ErrSessionBlocked = perrors.New("session Full Blocked") + ErrNullPeerAddr = perrors.New("peer address is nil") ) // NewSessionCallback will be invoked when server accepts a new client connection or client connects to server successfully. @@ -64,7 +68,10 @@ type ReadWriter interface { Writer } -// EventListener is used to process pkg that received from remote session +// EventListener processes session events. Inside a callback, Session.Reset +// schedules the reset and returns; it must not wait for the callback to finish. +// Use an AsyncResetter to observe completion, or a TryResetter to get +// ErrResetDuringCallback while any callback of this session is active. type EventListener interface { // OnOpen invoked when session opened // If the return error is not nil, @Session will be closed. @@ -93,6 +100,22 @@ type EventListener interface { OnMessage(Session, any) } +// AsyncResetter requests reset without blocking. Each call returns a separate +// buffered channel that receives one error (nil on success) and then closes. +// Concurrent pending requests share cleanup work, but retain their own result. +// Success means managed work, connection cleanup, and field reset have finished. +// Never wait for the result inside a callback of the session being reset. +type AsyncResetter interface { + ResetAsync() <-chan error +} + +// TryResetter reports ErrResetDuringCallback without requesting reset if any +// managed callback of this session is active at entry, regardless of the caller's +// goroutine. Otherwise it waits for reset and returns that operation's result. +type TryResetter interface { + TryReset() error +} + // EndPoint represents the identity of the client/server type EndPoint interface { // ID get EndPoint ID diff --git a/transport/session.go b/transport/session.go index 80249ec5..6e8a25e7 100644 --- a/transport/session.go +++ b/transport/session.go @@ -91,6 +91,11 @@ func init() { // Session wrap connection between the server and the client type Session interface { Connection + // Reset closes and clears the session after its managed work finishes. + // If this session has any active callback when called, Reset schedules + // cleanup and returns, including calls from other goroutines. Otherwise + // it waits for completion. Use AsyncResetter to observe completion; never + // wait for reset inside a callback of the session being reset. Reset() Conn() net.Conn Stat() string @@ -160,10 +165,13 @@ type session struct { grWG sync.WaitGroup lock sync.RWMutex packetLock sync.RWMutex + resetLock sync.Mutex + resetRun *sessionResetRun // callbacks closeCallback callbacks closeCallbackMutex sync.RWMutex + callbackDepth uatomic.Int32 } type sessionLifecycle struct { @@ -260,12 +268,89 @@ func newWSSession(conn *websocket.Conn, endPoint EndPoint) Session { return session } +// sessionResetRun groups pending requests into one cleanup operation. Each +// caller receives its own buffered result, so later resets cannot overwrite it. +type sessionResetRun struct { + results []chan error // guarded by resetLock +} + +// Reset requests cleanup and waits only if no callback of this session is +// active at entry. The count covers all goroutines, not just the caller. +// Later count changes do not change this call's decision to wait. Reset does +// not return failures; the worker logs them. Use TryReset or ResetAsync when +// the caller needs the result before reusing the session. func (s *session) Reset() { + async := s.callbackDepth.Load() > 0 + result := s.ResetAsync() + if !async { + <-result // The worker logs failures for both synchronous and async callers. + } +} + +// TryReset waits for cleanup unless a callback of this session is active at +// entry, in which case it returns ErrResetDuringCallback without requesting reset. +func (s *session) TryReset() error { + if s.callbackDepth.Load() > 0 { + return ErrResetDuringCallback + } + return <-s.ResetAsync() +} + +// ResetAsync requests cleanup without waiting. Pending requests share one +// worker, but each gets a channel containing its own result (nil on success), +// followed by closure. Callbacks must return without waiting for that result. +// Stop starting new operations before awaiting completion and reusing a session. +func (s *session) ResetAsync() <-chan error { + s.resetLock.Lock() + defer s.resetLock.Unlock() + result := make(chan error, 1) + if s.resetRun != nil { + s.resetRun.results = append(s.resetRun.results, result) + return result + } + run := &sessionResetRun{results: []chan error{result}} + s.resetRun = run + go func() { + var resetErr error + defer func() { + if r := recover(); r != nil { + resetErr = perrors.Errorf("session reset panic: %v", r) + log.Error(resetErr) + } + s.resetLock.Lock() + for _, result := range run.results { + result <- resetErr + close(result) + } + s.resetRun = nil + s.resetLock.Unlock() + }() + s.reset() + }() + return result +} + +var ( + _ AsyncResetter = (*session)(nil) + _ TryResetter = (*session)(nil) +) + +// reset runs only in the reset worker, never on a callback's goroutine. +func (s *session) reset() { + s.stop() + // Close the transport to release blocked reads, including idle codecs and + // WebSockets whose read loop cannot recover from deadline errors. + if conn := s.Conn(); conn != nil { + _ = conn.Close() + } lifecycle := s.stopHeartbeat() s.grWG.Wait() + s.gc() if lifecycle != nil { lifecycle.wait() } + s.packetLock.Lock() + defer s.packetLock.Unlock() // #105: Reset() previously did `*s = session{...}` without holding s.lock, // racing with concurrent readers. Reset the fields individually under the // lock instead; replacing the whole struct would also copy the mutexes @@ -281,6 +366,7 @@ func (s *session) Reset() { s.reader = nil s.writer = nil s.maxMsgLen = 0 + s.maxWriteMsgLen = 0 s.once = &sync.Once{} s.done = make(chan struct{}) s.period = period @@ -292,6 +378,20 @@ func (s *session) Reset() { s.closeCallback = callbacks{} } +// invokeCallback counts active user callbacks across this session. The defer +// restores the count on panic and supports nested and concurrent callbacks. +func (s *session) invokeCallback(f func()) { + s.callbackDepth.Add(1) + defer s.callbackDepth.Add(-1) + f() +} + +// readPackage applies the reset callback contract to application decoders. +func (s *session) readPackage(data []byte) (pkg any, n int, err error) { + s.invokeCallback(func() { pkg, n, err = s.reader.Read(s, data) }) + return +} + func (s *session) Conn() net.Conn { // #112: gc()/Reset() nil s.Connection under s.lock, so read it under s.lock // too. Only the field access is guarded; the returned net.Conn is immutable @@ -544,6 +644,8 @@ func (s *session) sessionToken() string { name, endPoint.EndPointType(), conn.ID(), conn.LocalAddr(), conn.RemoteAddr()) } +// WritePkg encodes and sends a package, keeping the encoder and write operation +// in the current lifecycle until both finish so reset cannot clear their state. func (s *session) WritePkg(pkg any, timeout time.Duration) (pkgBytesLenth int, successCount int, err error) { if pkg == nil { return 0, 0, fmt.Errorf("@pkg is nil") @@ -551,6 +653,13 @@ func (s *session) WritePkg(pkg any, timeout time.Duration) (pkgBytesLenth int, s if s.IsClosed() { return 0, 0, ErrSessionClosed } + s.lock.RLock() + lifecycle, writer := s.lifecycle, s.writer + s.lock.RUnlock() + if lifecycle == nil || !lifecycle.acquire() { + return 0, 0, ErrSessionClosed + } + defer lifecycle.release() defer func() { if r := recover(); r != nil { @@ -562,7 +671,8 @@ func (s *session) WritePkg(pkg any, timeout time.Duration) (pkgBytesLenth int, s } }() - pkgBytes, err := s.writer.Write(s, pkg) + var pkgBytes []byte + s.invokeCallback(func() { pkgBytes, err = writer.Write(s, pkg) }) if err != nil { log.Warnf("%s, [session.WritePkg] session.writer.Write(@pkg:%#v) = error:%+v", s.Stat(), pkg, err) return len(pkgBytes), 0, perrors.WithStack(err) @@ -814,6 +924,7 @@ func (s *session) WriteBytesArray(pkgs ...[]byte) (int, error) { return wlg, nil } +// heartbeat schedules OnCron only while the captured session lifecycle is open. func heartbeat(_ gxtime.TimerID, _ time.Time, arg any) error { ctx, _ := arg.(*heartbeatContext) if ctx == nil || ctx.session == nil || ctx.lifecycle == nil { @@ -843,7 +954,7 @@ func heartbeat(_ gxtime.TimerID, _ time.Time, arg any) error { } } - listener.OnCron(ss) + ss.invokeCallback(func() { listener.OnCron(ss) }) } // if enable task pool, run @f asynchronously. @@ -877,11 +988,23 @@ func (s *session) callOnOpen() (err error) { } }() - return s.listener.OnOpen(s) + s.invokeCallback(func() { err = s.listener.OnOpen(s) }) + return err } -// func (s *session) RunEventLoop() { +// run opens the session and starts its heartbeat and read loop. Startup itself +// is tracked so reset requests from OnOpen wait for initialization to unwind. func (s *session) run() { + // Register startup before OnOpen can request reset, and exclude new starts + // once a reset worker has begun waiting on grWG. + s.resetLock.Lock() + if s.resetRun != nil { + s.resetLock.Unlock() + return + } + s.grWG.Add(1) + s.resetLock.Unlock() + defer s.grWG.Done() if s.Connection == nil || s.listener == nil || s.writer == nil { errStr := fmt.Sprintf("session{name:%s, conn:%#v, listener:%#v, writer:%#v}", s.name, s.Connection, s.listener, s.writer) @@ -964,7 +1087,7 @@ func (s *session) addTask(pkg any) { if listener == nil { return } - listener.OnMessage(s, pkg) + s.invokeCallback(func() { listener.OnMessage(s, pkg) }) s.IncReadPkgNum() } if endPoint != nil { @@ -980,6 +1103,7 @@ func (s *session) handlePackage() { var err error defer func() { + defer s.gc() if r := recover(); r != nil { const size = 64 << 10 rBuf := make([]byte, size) @@ -992,13 +1116,12 @@ func (s *session) handlePackage() { if err != nil { log.Errorf("%s, [session.handlePackage] error:%+v", s.sessionToken(), perrors.WithStack(err)) if s != nil && s.listener != nil { - s.listener.OnError(s, err) + s.invokeCallback(func() { s.listener.OnError(s, err) }) } } if s.listener != nil { - s.listener.OnClose(s) + s.invokeCallback(func() { s.listener.OnClose(s) }) } - s.gc() }() if _, ok := s.Connection.(*gettyTCPConn); ok { @@ -1099,7 +1222,7 @@ func (s *session) handleTCPPackage() error { log.Warnf("%s, pktBuf.WriteNextEnd(%d) error:%+v", s.sessionToken(), bufLen, perrors.WithStack(werr)) } for pktBuf.Len() > 0 { - pkg, pkgLen, err = s.reader.Read(s, pktBuf.Bytes()) + pkg, pkgLen, err = s.readPackage(pktBuf.Bytes()) // SetMaxMsgLen uses the same lock; use one snapshot for both checks. s.lock.RLock() maxMsgLen := s.maxMsgLen @@ -1185,7 +1308,7 @@ func (s *session) handleUDPPackage() error { continue } - pkg, pkgLen, err = s.reader.Read(s, buf[:bufLen]) + pkg, pkgLen, err = s.readPackage(buf[:bufLen]) if log.IsDebugEnabled() { log.Debugf("s.reader.Read() = pkg:%#v, pkgLen:%d, err:%+v", pkg, pkgLen, perrors.WithStack(err)) } @@ -1242,7 +1365,7 @@ func (s *session) handleWSPackage() error { } s.UpdateActive() if s.reader != nil { - unmarshalPkg, length, err = s.reader.Read(s, pkg) + unmarshalPkg, length, err = s.readPackage(pkg) if err == nil && maxMsgLen > 0 && length > int(maxMsgLen) { err = perrors.Errorf("Message Too Long, length %d, session max message len %d", length, maxMsgLen) } @@ -1269,6 +1392,7 @@ func (s *session) handleWSPackage() error { return nil } +// stop closes the current lifecycle once and schedules its close callbacks. func (s *session) stop() { // #129: a repeat stop() used to return as soon as s.done was closed, while // the first caller was still inside the once body. handlePackage's @@ -1276,7 +1400,10 @@ func (s *session) stop() { // before the body read sessionClientKey/ignoreReconnectKey, silently // dropping the reconnect. Always going through once.Do makes repeat callers // wait for the body, so the attributes are still readable when it finishes. - s.once.Do(func() { + s.lock.RLock() + once := s.once + s.lock.RUnlock() + once.Do(func() { // let read/Write timeout asap now := time.Now() if conn := s.Conn(); conn != nil { @@ -1306,13 +1433,13 @@ func (s *session) stop() { const size = 64 << 10 rBuf := make([]byte, size) rBuf = rBuf[:runtime.Stack(rBuf, false)] - err := perrors.WithStack(fmt.Errorf("[session.invokeCloseCallbacks] panic session %s: err=%v\n%s", + err := perrors.WithStack(fmt.Errorf("[session.closeCallbacks] panic session %s: err=%v\n%s", sessionToken, r, rBuf)) log.Error(err) } }() - closeCallbacks.Invoke() + s.invokeCallback(closeCallbacks.Invoke) }(s.sessionToken(), closeCallbacks, lifecycle) clt, cltFound := s.GetAttribute(sessionClientKey).(*client) @@ -1339,6 +1466,8 @@ func (s *session) stopHeartbeat() *sessionLifecycle { return lifecycle } +// gc detaches session resources and registers asynchronous connection cleanup +// before making the detached state visible to a reset waiter. func (s *session) gc() { var conn Connection var wait time.Duration @@ -1351,11 +1480,13 @@ func (s *session) gc() { s.Connection = nil wait = s.wait lifecycle = s.lifecycle + if conn != nil && lifecycle != nil { + // Register cleanup before another gc/reset can see the detached + // connection and begin waiting for lifecycle completion. + lifecycle.addClosingTask() + } } s.lock.Unlock() - if conn != nil && lifecycle != nil { - lifecycle.addClosingTask() - } go func(conn Connection, wait time.Duration, lifecycle *sessionLifecycle) { if conn != nil && lifecycle != nil { diff --git a/transport/session_callback.go b/transport/session_callback.go index 5fe91eba..73464d95 100644 --- a/transport/session_callback.go +++ b/transport/session_callback.go @@ -28,6 +28,9 @@ package getty // - If the session is already closed, this addition will be ignored // - The combination of handler and key must be unique, otherwise it will override previous callbacks // - Callback functions will be executed in the order they were added when the session closes +// - Use an AsyncResetter to request reset from a callback and observe its +// completion outside the callback. Session.Reset from a callback schedules +// the reset and returns; a TryResetter reports ErrResetDuringCallback. func (s *session) AddCloseCallback(handler, key any, f CallBackFunc) { if f == nil { return @@ -61,23 +64,6 @@ func (s *session) RemoveCloseCallback(handler, key any) { s.closeCallback.Remove(handler, key) } -// invokeCloseCallbacks executes all registered close callback functions -// -// Function description: -// - Executes all registered close callbacks in the order they were added -// - Uses read lock to protect the callback list, ensuring concurrency safety -// - This method is typically called automatically when the session closes -// -// Notes: -// - This is an internal method, not recommended for external direct calls -// - If panic occurs during callback execution, it will be caught and logged -// - Callback functions should avoid long blocking operations, async processing is recommended for time-consuming tasks -func (s *session) invokeCloseCallbacks() { - s.closeCallbackMutex.RLock() - defer s.closeCallbackMutex.RUnlock() - s.closeCallback.Invoke() -} - // CallBackFunc defines the callback function type when Session closes // // Usage notes: diff --git a/transport/session_callback_test.go b/transport/session_callback_test.go index 1bb538fa..77c73402 100644 --- a/transport/session_callback_test.go +++ b/transport/session_callback_test.go @@ -122,7 +122,7 @@ func TestSessionCallback(t *testing.T) { callbackMutex.Unlock() }) - // Test invokeCloseCallbacks functionality + // Test close callback invocation t.Run("InvokeCloseCallbacks", func(t *testing.T) { s := &session{ once: &sync.Once{}, @@ -162,7 +162,7 @@ func TestSessionCallback(t *testing.T) { t.Errorf("Expected callback count is %d, but got %d", expectedCount, s.closeCallback.Len()) } - // Manually invoke close callbacks (simulate invokeCloseCallbacks) + // Manually invoke close callbacks (simulate session close) callbackMutex.Lock() callbackResults = nil // Clear previous results callbackMutex.Unlock() @@ -238,12 +238,10 @@ func TestSessionCallback(t *testing.T) { }) } -func TestResetWaitsForCloseCallbacks(t *testing.T) { - s := &session{ - once: &sync.Once{}, - done: make(chan struct{}), - closeCallback: callbacks{}, - } +// TestResetAsyncWaitsForCloseCallbacks verifies that reset completion includes +// registered close callbacks even when no read goroutine was started. +func TestResetAsyncWaitsForCloseCallbacks(t *testing.T) { + s := newTCPSession(&eofDataNetConn{}, nil).(*session) callbackStarted := make(chan struct{}) releaseCallback := make(chan struct{}) @@ -252,34 +250,20 @@ func TestResetWaitsForCloseCallbacks(t *testing.T) { <-releaseCallback }) - callbackDone := make(chan struct{}) - go func() { - s.invokeCloseCallbacks() - close(callbackDone) - }() + s.Close() <-callbackStarted - resetDone := make(chan struct{}) - go func() { - s.Reset() - close(resetDone) - }() + resetDone := s.ResetAsync() select { case <-resetDone: close(releaseCallback) - <-callbackDone t.Fatal("Reset returned while a close callback was still running") case <-time.After(50 * time.Millisecond): } close(releaseCallback) select { - case <-callbackDone: - case <-time.After(time.Second): - t.Fatal("close callback did not finish") - } - select { case <-resetDone: case <-time.After(time.Second): t.Fatal("Reset did not finish after the close callback completed") diff --git a/transport/session_reset_test.go b/transport/session_reset_test.go new file mode 100644 index 00000000..339c73d6 --- /dev/null +++ b/transport/session_reset_test.go @@ -0,0 +1,511 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package getty + +import ( + "errors" + "strings" + "sync" + "testing" + "time" +) + +import ( + gxsync "github.com/dubbogo/gost/sync" +) + +// resetContractListener invokes a test hook from a selected lifecycle callback. +type resetContractListener struct { + target string + hook func(Session) +} + +// call dispatches only the callback selected by the test. +func (l *resetContractListener) call(name string, s Session) { + if name == l.target { + l.hook(s) + } +} + +// OnOpen exercises reset before the read goroutine has started. +func (l *resetContractListener) OnOpen(s Session) error { l.call("open", s); return nil } + +// OnMessage exercises reset while the package loop owns lifecycle work. +func (l *resetContractListener) OnMessage(s Session, _ any) { l.call("message", s) } + +// OnError exercises reset in the package loop's shutdown defer. +func (l *resetContractListener) OnError(s Session, _ error) { l.call("error", s) } + +// OnClose exercises the original self-wait regression. +func (l *resetContractListener) OnClose(s Session) { l.call("close", s) } + +// OnCron exercises reset while the heartbeat owns lifecycle work. +func (l *resetContractListener) OnCron(s Session) { l.call("cron", s) } + +// resetContractCodec can request reset from a reader or writer callback. +type resetContractCodec struct { + hook func(Session) + fail bool +} + +// Read returns a complete frame or the test error after invoking the hook. +func (c resetContractCodec) Read(s Session, data []byte) (any, int, error) { + if c.hook != nil { + c.hook(s) + } + if c.fail { + return nil, 0, errTestReadFailure + } + return string(data), len(data), nil +} + +// Write invokes the test hook before returning the encoded payload. +func (c resetContractCodec) Write(s Session, _ any) ([]byte, error) { + if c.hook != nil { + c.hook(s) + } + return []byte("payload"), nil +} + +// waitForReset bounds a test's wait for asynchronous cleanup and reset. +func waitForReset[T any](t *testing.T, done <-chan T) { + t.Helper() + select { + case result := <-done: + if err, ok := any(result).(error); ok && err != nil { + t.Fatalf("reset failed: %v", err) + } + case <-time.After(3 * time.Second): + t.Fatal("reset did not complete") + } +} + +// TestResetCallbackContract checks prompt rejection of synchronous reset and +// deferred completion of asynchronous reset in every managed callback path. +func TestResetCallbackContract(t *testing.T) { + for _, target := range []string{"open", "message", "error", "close", "cron", "registered close", "reader", "writer", "pooled message", "pooled cron"} { + t.Run(target, func(t *testing.T) { + endpoint := newServer(TCP_SERVER) + if target == "pooled message" || target == "pooled cron" { + pool := gxsync.NewTaskPoolSimple(1) + t.Cleanup(pool.Close) + endpoint = newServer(TCP_SERVER, WithServerTaskPool(pool)) + } + ss := newTCPSession(&eofDataNetConn{data: []byte("frame")}, endpoint).(*session) + oldLifecycle := ss.lifecycle + release := make(chan struct{}) + var releaseOnce sync.Once + unblock := func() { releaseOnce.Do(func() { close(release) }) } + defer unblock() + type request struct { + err error + done, again <-chan error + } + entered := make(chan request, 1) + hook := func(s Session) { + err := s.(TryResetter).TryReset() + // The compatibility path must return rather than wait for the + // callback that is calling it; it starts the asynchronous reset. + s.Reset() + done := s.(AsyncResetter).ResetAsync() + entered <- request{err, done, s.(AsyncResetter).ResetAsync()} + <-release + } + listener := &resetContractListener{target: target, hook: hook} + switch target { + case "pooled message": + listener.target = "message" + case "pooled cron": + listener.target = "cron" + } + ss.SetEventListener(listener) + ss.SetPkgHandler(resetContractCodec{fail: target == "error"}) + switch target { + case "registered close": + ss.AddCloseCallback("reset", "test", func() { hook(ss) }) + case "reader": + ss.SetReader(resetContractCodec{hook: hook}) + case "writer": + ss.SetWriter(resetContractCodec{hook: hook}) + } + operationDone := make(chan struct{}) + go func() { + defer close(operationDone) + switch target { + case "cron", "pooled cron": + _ = heartbeat(0, time.Time{}, &heartbeatContext{session: ss, lifecycle: oldLifecycle}) + case "pooled message": + ss.addTask("frame") + case "writer": + _, _, _ = ss.WritePkg("payload", 0) + default: + ss.run() + } + }() + var req request + select { + case req = <-entered: + case <-time.After(3 * time.Second): + t.Fatal("callback did not return from Reset/ResetAsync") + } + if !errors.Is(req.err, ErrResetDuringCallback) { + t.Fatalf("TryReset error = %v, want ErrResetDuringCallback", req.err) + } + if req.done == req.again { + t.Fatal("concurrent reset requests must have independent results") + } + // Session-wide activity also makes external Reset asynchronous. + externalReset := make(chan struct{}) + go func() { + ss.Reset() + close(externalReset) + }() + select { + case <-externalReset: + case <-time.After(3 * time.Second): + t.Fatal("external Reset blocked while a callback was active") + } + select { + case <-req.done: + t.Fatal("reset completed before its requesting callback returned") + default: + } + ss.lock.RLock() + sameGeneration := ss.lifecycle == oldLifecycle && ss.listener == listener + ss.lock.RUnlock() + if !sameGeneration { + t.Fatal("reset cleared the active callback's generation") + } + unblock() + waitForReset(t, req.done) + waitForReset(t, req.again) + waitForReset(t, operationDone) + waitForReset(t, externalReset) + if ss.Connection != nil || ss.listener != nil || ss.lifecycle == oldLifecycle { + t.Fatal("completion notification preceded state reset") + } + if ss.callbackDepth.Load() != 0 || ss.grNum.Load() != 0 { + t.Fatal("reset completed with active callbacks or read goroutines") + } + if oldLifecycle.acquire() { + oldLifecycle.release() + t.Fatal("old lifecycle accepted new work") + } + }) + } +} + +// TestResetAsyncWaitsForWrites ensures completion also fences Send/WriteBytes, +// whose in-flight operations are protected by packetLock rather than lifecycle. +func TestResetAsyncWaitsForWrites(t *testing.T) { + conn := newFragmentGateNetConn() + ss := newTCPSession(conn, nil).(*session) + writeDone := make(chan struct{}) + go func() { + defer close(writeDone) + _, _ = ss.WriteBytes([]byte("payload")) + }() + select { + case <-conn.entered: + case <-time.After(3 * time.Second): + close(conn.releaseFirst) + t.Fatal("write did not enter transport") + } + done := ss.ResetAsync() + select { + case <-done: + close(conn.releaseFirst) + waitForReset(t, writeDone) + t.Fatal("reset completed while a write was in progress") + case <-time.After(30 * time.Millisecond): + } + close(conn.releaseFirst) + waitForReset(t, done) + waitForReset(t, writeDone) +} + +// resetCleanupConn holds connection cleanup open to test reset's completion fence. +type resetCleanupConn struct { + *gettyTCPConn + entered chan struct{} + release chan struct{} +} + +// CloseConn blocks until the test allows resource cleanup to complete. +func (c *resetCleanupConn) CloseConn(int) { close(c.entered); <-c.release } + +// TestResetAsyncWaitsForGC ensures notification includes asynchronous cleanup, +// even when gc detached the connection before the reset request arrived. +func TestResetAsyncWaitsForGC(t *testing.T) { + conn := &resetCleanupConn{gettyTCPConn: newGettyTCPConn(&eofDataNetConn{}), entered: make(chan struct{}), release: make(chan struct{})} + ss := newSession(nil, conn) + ss.gc() + waitForReset(t, conn.entered) + done := ss.ResetAsync() + select { + case <-done: + close(conn.release) + t.Fatal("reset completed before connection cleanup") + case <-time.After(30 * time.Millisecond): + } + close(conn.release) + waitForReset(t, done) +} + +// TestExternalResetDefersForActiveCallback checks that external Reset returns +// early while cleanup still waits for OnCron. +func TestExternalResetDefersForActiveCallback(t *testing.T) { + listener := &blockingCronEventListener{ + entered: make(chan struct{}), + release: make(chan struct{}), + } + ss := newTCPSession(&eofDataNetConn{}, newServer(TCP_SERVER)).(*session) + ss.SetEventListener(listener) + ctx := &heartbeatContext{session: ss, lifecycle: ss.lifecycle} + heartbeatDone := make(chan error, 1) + go func() { + heartbeatDone <- heartbeat(0, time.Time{}, ctx) + }() + select { + case <-listener.entered: + case <-time.After(time.Second): + t.Fatal("heartbeat did not enter OnCron") + } + + resetReturned := make(chan struct{}) + go func() { + ss.Reset() + close(resetReturned) + }() + select { + case <-resetReturned: + case <-time.After(3 * time.Second): + t.Fatal("external Reset blocked while a callback was active") + } + + if err := ss.TryReset(); !errors.Is(err, ErrResetDuringCallback) { + t.Fatalf("external TryReset = %v, want ErrResetDuringCallback", err) + } + result := ss.ResetAsync() + select { + case <-result: + t.Fatal("reset completed before callback returned") + default: + } + close(listener.release) + select { + case err := <-heartbeatDone: + if err != nil { + t.Fatalf("heartbeat returned error: %v", err) + } + case <-time.After(time.Second): + t.Fatal("heartbeat did not return") + } + waitForReset(t, result) + if ss.Connection != nil || ss.listener != nil { + t.Fatal("reset result arrived before session state was cleared") + } +} + +// panicCloseNetConn makes the reset worker panic before session state is reset. +type panicCloseNetConn struct { + eofDataNetConn +} + +// Close panics so the reset worker has to report a failed reset. +func (*panicCloseNetConn) Close() error { panic("test close panic") } + +// TestResetReportsWorkerPanic ensures a panicking reset is observable and is not +// reported as a successful reset. +func TestResetReportsWorkerPanic(t *testing.T) { + ss := newTCPSession(&panicCloseNetConn{}, nil).(*session) + err := ss.TryReset() + if err == nil { + t.Fatal("TryReset returned nil after reset worker panic") + } + if !strings.Contains(err.Error(), "reset panic") { + t.Fatalf("TryReset error = %v, want reset panic", err) + } + if ss.Connection == nil { + t.Fatal("failed reset unexpectedly cleared the session connection") + } +} + +// TestResetSynchronousCompletion checks the non-callback API and repeat resets. +func TestResetSynchronousCompletion(t *testing.T) { + ss := newTCPSession(&eofDataNetConn{}, nil).(*session) + for range 2 { + ss.SetMaxWriteMsgLen(123) + if err := ss.TryReset(); err != nil { + t.Fatal(err) + } + if ss.Connection != nil || ss.maxWriteMsgLen != 0 { + t.Fatal("Reset did not clear session state") + } + } + // Reset keeps the source-compatible signature and blocks outside callbacks. + ss.SetMaxWriteMsgLen(123) + ss.Reset() + if ss.Connection != nil || ss.maxWriteMsgLen != 0 { + t.Fatal("Reset did not clear session state") + } +} + +// TestResetAcrossSessions checks that only the target session's callbacks +// determine whether Reset waits, even when invoked from another session's callback. +func TestResetAcrossSessions(t *testing.T) { + for _, active := range []bool{false, true} { + name := "target idle" + if active { + name = "target callback active" + } + t.Run(name, func(t *testing.T) { + conn := &resetCleanupConn{gettyTCPConn: newGettyTCPConn(&eofDataNetConn{}), entered: make(chan struct{}), release: make(chan struct{})} + b := newSession(newServer(TCP_SERVER), conn) + var releaseOnce sync.Once + releaseCleanup := func() { releaseOnce.Do(func() { close(conn.release) }) } + defer releaseCleanup() + listener := &blockingCronEventListener{entered: make(chan struct{}), release: make(chan struct{})} + var callbackOnce sync.Once + releaseCallback := func() { callbackOnce.Do(func() { close(listener.release) }) } + defer releaseCallback() + bDone := make(chan struct{}) + if active { + b.SetEventListener(listener) + ctx := &heartbeatContext{session: b, lifecycle: b.lifecycle} + go func() { + defer close(bDone) + _ = heartbeat(0, time.Time{}, ctx) + }() + waitForReset(t, listener.entered) + } + a := newTCPSession(&eofDataNetConn{}, newServer(TCP_SERVER)).(*session) + returned := make(chan struct{}) + a.SetEventListener(&resetContractListener{target: "cron", hook: func(Session) { b.Reset() }}) + ctx := &heartbeatContext{session: a, lifecycle: a.lifecycle} + go func() { + defer close(returned) + _ = heartbeat(0, time.Time{}, ctx) + }() + waitForReset(t, conn.entered) + result := b.ResetAsync() // Join the cleanup held open by conn.release. + if active { + waitForReset(t, returned) + } else { + select { + case <-returned: + t.Fatal("A's callback made idle B.Reset return before cleanup") + case <-time.After(30 * time.Millisecond): + } + } + releaseCleanup() + if active { + select { + case <-result: + t.Fatal("B reset completed before B's callback returned") + case <-time.After(30 * time.Millisecond): + } + releaseCallback() + waitForReset(t, bDone) + } + waitForReset(t, result) + waitForReset(t, returned) + if b.Connection != nil || b.listener != nil { + t.Fatal("B reset completed before clearing state") + } + waitForReset(t, a.ResetAsync()) + }) + } +} + +// callResetDeep keeps real stack frames between a callback and Reset. +// +//go:noinline +func callResetDeep(s *session, depth int) int { + if depth == 0 { + s.Reset() + return 0 + } + return 1 + callResetDeep(s, depth-1) +} + +// TestResetDeepCallback checks that reset does not depend on stack depth. +func TestResetDeepCallback(t *testing.T) { + ss := newTCPSession(&eofDataNetConn{}, newServer(TCP_SERVER)).(*session) + result := make(chan (<-chan error), 1) + ss.SetEventListener(&resetContractListener{target: "cron", hook: func(Session) { + callResetDeep(ss, 128) + result <- ss.ResetAsync() + }}) + ctx := &heartbeatContext{session: ss, lifecycle: ss.lifecycle} + go func() { _ = heartbeat(0, time.Time{}, ctx) }() + select { + case done := <-result: + waitForReset(t, done) + case <-time.After(3 * time.Second): + t.Fatal("Reset blocked inside a deep callback") + } +} + +// TestCallbackDepthUnwinds verifies nested callbacks release the count on panic. +func TestCallbackDepthUnwinds(t *testing.T) { + ss := newTCPSession(&eofDataNetConn{}, nil).(*session) + func() { + defer func() { + if recover() == nil { + t.Error("expected callback panic") + } + }() + ss.invokeCallback(func() { + ss.invokeCallback(func() { + if ss.callbackDepth.Load() != 2 { + t.Error("nested callbacks not counted") + } + panic("callback panic") + }) + }) + }() + if ss.callbackDepth.Load() != 0 { + t.Fatal("panic leaked callback count") + } + waitForReset(t, ss.ResetAsync()) +} + +// TestResetResultSurvivesRetry keeps a failed result unread until a later reset +// succeeds, proving results belong to requests rather than mutable session state. +func TestResetResultSurvivesRetry(t *testing.T) { + ss := newTCPSession(&panicCloseNetConn{}, nil).(*session) + first := ss.ResetAsync() + deadline := time.After(3 * time.Second) + for len(first) == 0 { + select { + case <-deadline: + t.Fatal("first reset did not publish its result") + case <-time.After(time.Millisecond): + } + } + // The failed reset stopped the lifecycle but did not clear the connection. + ss.lifecycle.wait() + ss.lock.Lock() + ss.Connection = newGettyTCPConn(&eofDataNetConn{}) + ss.lock.Unlock() + waitForReset(t, ss.ResetAsync()) + if err := <-first; err == nil || !strings.Contains(err.Error(), "reset panic") { + t.Fatalf("first reset result overwritten: %v", err) + } +} diff --git a/transport/session_test.go b/transport/session_test.go index c610c3ba..f23549a9 100644 --- a/transport/session_test.go +++ b/transport/session_test.go @@ -505,6 +505,7 @@ func TestConcurrentWritePkgTimeoutRestoration(t *testing.T) { } } +// TestResetWaitsForPackageLoop keeps reset pending until the read loop exits. func TestResetWaitsForPackageLoop(t *testing.T) { netConn := &resetBarrierNetConn{ entered: make(chan struct{}), @@ -524,11 +525,7 @@ func TestResetWaitsForPackageLoop(t *testing.T) { t.Fatal("package loop did not enter Read") } - resetDone := make(chan struct{}) - go func() { - ss.Reset() - close(resetDone) - }() + resetDone := ss.ResetAsync() select { case <-resetDone: @@ -551,7 +548,9 @@ func TestResetWaitsForPackageLoop(t *testing.T) { } } -func TestResetWaitsForActiveHeartbeat(t *testing.T) { +// TestResetAsyncWaitsForActiveHeartbeat fences asynchronous completion until an +// already running heartbeat returns. +func TestResetAsyncWaitsForActiveHeartbeat(t *testing.T) { listener := &blockingCronEventListener{ entered: make(chan struct{}), release: make(chan struct{}), @@ -571,11 +570,7 @@ func TestResetWaitsForActiveHeartbeat(t *testing.T) { t.Fatal("heartbeat did not enter OnCron") } - resetDone := make(chan struct{}) - go func() { - ss.Reset() - close(resetDone) - }() + resetDone := ss.ResetAsync() select { case <-resetDone: t.Fatal("Reset returned while the heartbeat callback was still running")