fix: race condition

Author: joway

connection_impl.go

@@ -503,17 +503,10 @@ func (c *connection) flush() error {
 	if c.outputBuffer.IsEmpty() {
 		return nil
 	}
-	if c.operator.getMode() == ophup {
-		// triggered read throttled, so here shouldn't trigger read event again
-		err = c.operator.Control(PollHup2W)
-	} else {
-		err = c.operator.Control(PollR2RW)
-	}
-	c.operator.done()
-	if err != nil {
-		return Exception(err, "when flush")
-	}
 
+	// no need to check if resume write successfully
+	// if resume failed, the connection will be triggered triggerWrite(err), and waitFlush will return err
+	c.resumeWrite()
 	return c.waitFlush()
 }
 
@@ -546,8 +539,8 @@ func (c *connection) waitFlush() (err error) {
 		default:
 		}
 		// if timeout, remove write event from poller
-		// we cannot flush it again, since we don't if the poller is still process outputBuffer
-		c.operator.Control(PollRW2R)
+		// we cannot flush it again, since we don't know if the poller is still processing outputBuffer
+		c.pauseWrite()
 		return Exception(ErrWriteTimeout, c.remoteAddr.String())
 	}
 }

connection_reactor.go

@@ -80,6 +80,12 @@ func (c *connection) closeBuffer() {
 
 // inputs implements FDOperator.
 func (c *connection) inputs(vs [][]byte) (rs [][]byte) {
+	// trigger throttle
+	if c.readBufferThreshold > 0 && int64(c.inputBuffer.Len()) >= c.readBufferThreshold {
+		c.pauseRead()
+		return
+	}
+
 	vs[0] = c.inputBuffer.book(c.bookSize, c.maxSize)
 	return vs[:1]
 }
@@ -123,6 +129,7 @@ func (c *connection) inputAck(n int) (err error) {
 func (c *connection) outputs(vs [][]byte) (rs [][]byte, supportZeroCopy bool) {
 	if c.outputBuffer.IsEmpty() {
 		c.pauseWrite()
+		c.triggerWrite(nil)
 		return rs, c.supportZeroCopy
 	}
 	rs = c.outputBuffer.GetBytes(vs)
@@ -137,50 +144,47 @@ func (c *connection) outputAck(n int) (err error) {
 	}
 	if c.outputBuffer.IsEmpty() {
 		c.pauseWrite()
+		c.triggerWrite(nil)
 	}
 	return nil
 }
 
+/* The race description of operator event monitoring
+- Pause operation will remove old event monitor of operator
+- Resume operation will add new event monitor of operator
+- Only poller could use Pause to remove event monitor, and poller already hold the op.do() locker
+- Only user could use Resume, and user's operation maybe compete with poller's operation
+- If competition happen, because of all resume operation will monitor all events, it's safe to do that with a race condition.
+    * If resume first and pause latter, poller will monitor the accurate events it needs.
+    * If pause first and resume latter,  poller will monitor the duplicate events which will be removed after next poller triggered.
+      And poller will ensure to remove the duplicate events.
+- If there is no readBufferThreshold option, the code path will be more simple and efficient.
+*/
+
 // pauseWrite removed the monitoring of write events.
 // pauseWrite used in poller
 func (c *connection) pauseWrite() {
-	switch c.operator.getMode() {
-	case opreadwrite:
-		c.operator.Control(PollRW2R)
-	case opwrite:
-		c.operator.Control(PollW2Hup)
-	}
-	c.triggerWrite(nil)
+	c.operator.Control(PollRW2R)
+}
+
+// resumeWrite add the monitoring of write events.
+// resumeWrite used by users
+func (c *connection) resumeWrite() {
+	c.operator.Control(PollR2RW)
 }
 
 // pauseRead removed the monitoring of read events.
 // pauseRead used in poller
 func (c *connection) pauseRead() {
-	// Note that the poller ensure that every fd should read all left data in socket buffer before detach it.
-	// So the operator mode should never be ophup.
-	var changeTo PollEvent
-	switch c.operator.getMode() {
-	case opread:
-		changeTo = PollR2Hup
-	case opreadwrite:
-		changeTo = PollRW2W
-	}
-	if changeTo > 0 && atomic.CompareAndSwapInt32(&c.operator.throttled, 0, 1) {
-		c.operator.Control(changeTo)
+	if atomic.CompareAndSwapInt32(&c.operator.throttled, 0, 1) {
+		c.operator.Control(PollRW2W)
 	}
 }
 
 // resumeRead add the monitoring of read events.
 // resumeRead used by users
 func (c *connection) resumeRead() {
-	var changeTo PollEvent
-	switch c.operator.getMode() {
-	case ophup:
-		changeTo = PollHup2R
-	case opwrite:
-		changeTo = PollW2RW
-	}
-	if changeTo > 0 && atomic.CompareAndSwapInt32(&c.operator.throttled, 1, 0) {
-		c.operator.Control(changeTo)
+	if atomic.CompareAndSwapInt32(&c.operator.throttled, 1, 0) {
+		c.operator.Control(PollW2RW)
 	}
 }

poll.go

@@ -48,31 +48,31 @@ type PollEvent int
 const (
 	// PollReadable is used to monitor whether the FDOperator registered by
 	// listener and connection is readable or closed.
-	PollReadable PollEvent = 0x1
+	PollReadable PollEvent = iota + 1
 
 	// PollWritable is used to monitor whether the FDOperator created by the dialer is writable or closed.
 	// ET mode must be used (still need to poll hup after being writable)
-	PollWritable PollEvent = 0x2
+	PollWritable
 
 	// PollDetach is used to remove the FDOperator from poll.
-	PollDetach PollEvent = 0x3
+	PollDetach
 
 	// PollR2RW is used to monitor writable for FDOperator,
 	// which is only called when the socket write buffer is full.
-	PollR2RW PollEvent = 0x4
+	PollR2RW
 	// PollRW2R is used to remove the writable monitor of FDOperator, generally used with PollR2RW.
-	PollRW2R PollEvent = 0x5
+	PollRW2R
 
 	// PollRW2W is used to remove the readable monitor of FDOperator.
-	PollRW2W PollEvent = 0x6
+	PollRW2W
 	// PollW2RW is used to add the readable monitor of FDOperator, generally used with PollRW2W.
-	PollW2RW  PollEvent = 0x7
-	PollW2Hup PollEvent = 0x8
+	PollW2RW
+	PollW2Hup
 
 	// PollR2Hup is used to remove the readable monitor of FDOperator.
-	PollR2Hup PollEvent = 0x9
+	PollR2Hup
 	// PollHup2R is used to add the readable monitor of FDOperator, generally used with PollR2Hup.
-	PollHup2R PollEvent = 0xA
+	PollHup2R
 	// PollHup2W is used to add the writeable monitor of FDOperator.
-	PollHup2W PollEvent = 0xB
+	PollHup2W
 )