fix: detach race

connection_onevent.go

@@ -185,7 +185,7 @@ func (c *connection) onProcess(isProcessable func(c *connection) bool, process f
 				if c.IsActive() {
 					c.Close()
 				} else {
-					c.closeCallback(false)
+					c.closeCallback(false, false)
 				}
 			}
 		}()
@@ -200,7 +200,7 @@ func (c *connection) onProcess(isProcessable func(c *connection) bool, process f
 		}
 		// Handling callback if connection has been closed.
 		if !c.IsActive() {
-			c.closeCallback(false)
+			c.closeCallback(false, false)
 			panicked = false
 			return
 		}
@@ -221,10 +221,16 @@ func (c *connection) onProcess(isProcessable func(c *connection) bool, process f
 // closeCallback .
 // It can be confirmed that closeCallback and onRequest will not be executed concurrently.
 // If onRequest is still running, it will trigger closeCallback on exit.
-func (c *connection) closeCallback(needLock bool) (err error) {
+func (c *connection) closeCallback(needLock bool, needDetach bool) (err error) {
 	if needLock && !c.lock(processing) {
 		return nil
 	}
+	if needDetach && c.operator.poll != nil {
+		// If Close is called during OnPrepare, poll is not registered.
+		if err := c.operator.Control(PollDetach); err != nil {
+			logger.Printf("NETPOLL: onClose detach operator failed: %v", err)
+		}
+	}
 	var latest = c.closeCallbacks.Load()
 	if latest == nil {
 		return nil

connection_reactor.go

@@ -37,23 +37,17 @@ func (c *connection) onHup(p Poll) error {
 	var onConnect, _ = c.onConnectCallback.Load().(OnConnect)
 	var onRequest, _ = c.onRequestCallback.Load().(OnRequest)
 	if onConnect != nil || onRequest != nil {
-		c.closeCallback(true)
+		c.closeCallback(true, false)
 	}
 	return nil
 }
 
 // onClose means close by user.
 func (c *connection) onClose() error {
 	if c.closeBy(user) {
-		// If Close is called during OnPrepare, poll is not registered.
-		if c.operator.poll != nil {
-			if err := c.operator.Control(PollDetach); err != nil {
-				logger.Printf("NETPOLL: onClose detach operator failed: %v", err)
-			}
-		}
 		c.triggerRead(Exception(ErrConnClosed, "self close"))
 		c.triggerWrite(Exception(ErrConnClosed, "self close"))
-		c.closeCallback(true)
+		c.closeCallback(true, true)
 		return nil
 	}
 
@@ -63,7 +57,7 @@ func (c *connection) onClose() error {
 
 	// If OnRequest is nil, relies on the user to actively close the connection to recycle resources.
 	if closedByPoller {
-		c.closeCallback(true)
+		c.closeCallback(true, false)
 	}
 	return nil
 }