更新

Author: guonaihong

api_epoll.go

@@ -178,7 +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)
+				e.getMultiEventLoop().Logger.Debug("ev.Fd get conn is nil", "fd", ev.Fd)
 				unix.Close(int(ev.Fd))
 				continue
 			}
@@ -225,6 +225,8 @@ func (e *epollState) apiPoll(tv time.Duration) (retVal int, err error) {
 
 func (e *epollState) process(conn *Conn, isRead, isWrite bool) bool {
 	if isRead {
+
+		e.getMultiEventLoop().addReadEvNum()
 		err := conn.processWebsocketFrame()
 		if err != nil {
 			e.getMultiEventLoop().Logger.Info("processWebsocketFrame", "err", err.Error())
@@ -234,6 +236,7 @@ func (e *epollState) process(conn *Conn, isRead, isWrite bool) bool {
 	}
 
 	if isWrite {
+		e.getMultiEventLoop().addWriteEvNum()
 		// 刷新下直接写入失败的数据
 		conn.flushOrClose()
 	}

task/stream2/stream2.go

@@ -44,7 +44,6 @@ type stream2 struct {
 	conf           *driver.Conf     // 初始化传递过来的参数
 	process        func()           // 处理单个任务的循环
 	onMessageCount int64            //需要处理的OnMessage个数
-	closed         int32
 }
 
 func (s *stream2) addOnMessageCount(n int) {

task/stream2/stream2_executor.go

@@ -17,7 +17,6 @@ package stream2
 import (
 	"runtime"
 	"sync"
-	"sync/atomic"
 	"unsafe"
 )
 
@@ -41,10 +40,7 @@ func myUnlock(mu *sync.Mutex) {
 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)
 	myUnlock(mu)
@@ -115,14 +111,8 @@ func (s *stream2Executor) run(mu *sync.Mutex) bool {
 func (s *stream2Executor) Close(mu *sync.Mutex) error {
 	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
 	myUnlock(mu)
 	return nil