更新

Author: guonaihong

api_epoll.go

@@ -178,6 +178,7 @@ func (e *epollState) apiPoll(tv time.Duration) (retVal int, err error) {
 			ev := &e.events[i]
 			conn := e.parent.getConn(int(ev.Fd))
 			if conn == nil {
+				e.getMultiEventLoop().Logger.Debug("ev.Fd is", "fd", ev.Fd)
 				unix.Close(int(ev.Fd))
 				continue
 			}
@@ -226,6 +227,7 @@ func (e *epollState) process(conn *Conn, isRead, isWrite bool) bool {
 	if isRead {
 		err := conn.processWebsocketFrame()
 		if err != nil {
+			e.getMultiEventLoop().Logger.Info("processWebsocketFrame", "err", err.Error())
 			conn.closeWithLock(err)
 			return true
 		}

conn_unix.go

@@ -121,7 +121,7 @@ func duplicateSocket(socketFD int) (int, error) {
 }
 
 // 没有加锁的版本，有外层已经有锁保护，所以不需要加锁
-func (c *Conn) closeInnerWithOnClose(err error, onClose bool) {
+func (c *Conn) closeWithoutLockOnClose(err error, onClose bool) {
 
 	if c.isClosed() {
 		return
@@ -148,9 +148,9 @@ func (c *Conn) closeInnerWithOnClose(err error, onClose bool) {
 
 }
 
-func (c *Conn) closeInner(err error) {
+func (c *Conn) closeWithoutLock(err error) {
 
-	c.closeInnerWithOnClose(err, true)
+	c.closeWithoutLockOnClose(err, true)
 }
 
 func (c *Conn) closeWithLock(err error) {
@@ -167,7 +167,7 @@ func (c *Conn) closeWithLock(err error) {
 	if err == nil {
 		err = io.EOF
 	}
-	c.closeInnerWithOnClose(err, false)
+	c.closeWithoutLockOnClose(err, false)
 
 	c.mu.Unlock()
 
@@ -262,7 +262,7 @@ func (c *Conn) maybeWriteAll(b []byte) (total int, ws writeState, err error) {
 			}
 
 			c.getLogger().Error("writeOrAddPoll", "err", err.Error(), slog.Int64("fd", c.fd), slog.Int("b.len", len(b)))
-			c.closeInner(err)
+			c.closeWithoutLock(err)
 			return total, writeDefault, err
 		}
 
@@ -354,9 +354,10 @@ func (c *Conn) processWebsocketFrame() (err error) {
 	}
 
 	if c.readTimeout > 0 {
-		if err = c.setReadDeadline(time.Now().Add(c.readTimeout)); err != nil {
-			return err
-		}
+		// if err = c.setReadDeadline(time.Time{}); err != nil {
+		// 	return err
+		// }
+		c.setReadDeadline(time.Now().Add(c.readTimeout))
 	}
 	n := 0
 	var success bool
@@ -492,9 +493,14 @@ func (c *Conn) setDeadlineInner(t **time.Timer, tm time.Time, err error) error {
 		return nil
 	}
 	c.mu.Lock()
-	defer c.mu.Unlock()
+	// c.getLogger().Error("Conn-lock", "addr", uintptr(unsafe.Pointer(c)))
+	defer func() {
+		// c.getLogger().Error("Conn-unlock", "addr", uintptr(unsafe.Pointer(c)))
+		c.mu.Unlock()
+	}()
 	if tm.IsZero() {
 		if *t != nil {
+			// c.getLogger().Error("conn-reset", "addr", uintptr(unsafe.Pointer(c)))
 			(*t).Stop()
 			*t = nil
 		}

task/stream2/stream2_executor.go

@@ -26,26 +26,33 @@ type stream2Executor struct {
 	parent *stream2
 }
 
-func (s *stream2Executor) AddTask(mu *sync.Mutex, f func() bool) error {
-
+func myLock(mu *sync.Mutex) {
 	if mu != nil {
 		mu.Lock()
 	}
+}
+
+func myUnlock(mu *sync.Mutex) {
+	if mu != nil {
+		mu.Unlock()
+	}
+}
+
+func (s *stream2Executor) AddTask(mu *sync.Mutex, f func() bool) error {
 
+	myLock(mu)
 	if atomic.LoadInt32(&s.parent.closed) == 1 {
+		myUnlock(mu)
 		return nil
 	}
-
 	process := len(s.list) == 0
 	s.list = append(s.list, f)
-	if mu != nil {
-		mu.Unlock()
-	}
+	myUnlock(mu)
 
 	if process {
-		mu.Lock()
+		myLock(mu)
 		listSize := len(s.list)
-		mu.Unlock()
+		myUnlock(mu)
 		s.parent.fn <- func() bool {
 			s.run(mu)
 			return false
@@ -67,39 +74,28 @@ func (s *stream2Executor) AddTask(mu *sync.Mutex, f func() bool) error {
 func (s *stream2Executor) run(mu *sync.Mutex) bool {
 	var f func() bool
 	for i := 0; ; i++ {
-		if mu != nil {
-			// 加锁
-			mu.Lock()
-		}
+		myLock(mu)
 
 		if len(s.list) == 0 {
-			if mu != nil {
-				mu.Unlock()
-			}
+			myUnlock(mu)
 			return false
 		}
 
 		if len(s.list) == i {
 			s.list = s.list[0:0]
-			if mu != nil {
-				mu.Unlock()
-			}
+			myUnlock(mu)
 			return false
 		}
 
 		if i >= len(s.list) {
 			s.list = s.list[0:0]
-			if mu != nil {
-				mu.Unlock()
-			}
+			myUnlock(mu)
 			return false
 		}
 
 		f = s.list[i]
 		s.list[i] = nil
-		if mu != nil {
-			mu.Unlock()
-		}
+		myUnlock(mu)
 
 		func() {
 			defer func() {
@@ -117,19 +113,17 @@ func (s *stream2Executor) run(mu *sync.Mutex) bool {
 }
 
 func (s *stream2Executor) Close(mu *sync.Mutex) error {
-	if mu != nil {
-		mu.Lock()
-	}
+	myLock(mu)
 
 	if atomic.LoadInt32(&s.parent.closed) == 1 {
+		myUnlock(mu)
 		return nil
 	}
+
 	s.parent.subOnMessageCount(-len(s.list))
 
 	atomic.StoreInt32(&s.parent.closed, 1)
 	s.list = nil
-	if mu != nil {
-		mu.Unlock()
-	}
+	myUnlock(mu)
 	return nil
 }