Fix memory leaking by extra timers.

internal/multiplex/datagramBufferedPipe.go

@@ -20,6 +20,8 @@ type datagramBufferedPipe struct {
 	rwCond    *sync.Cond
 	wtTimeout time.Duration
 	rDeadline time.Time
+
+	timeoutTimer *time.Timer
 }
 
 func NewDatagramBufferedPipe() *datagramBufferedPipe {
@@ -45,7 +47,7 @@ func (d *datagramBufferedPipe) Read(target []byte) (int, error) {
 			if delta <= 0 {
 				return 0, ErrTimeout
 			}
-			time.AfterFunc(delta, d.rwCond.Broadcast)
+			d.broadcastAfter(delta)
 		}
 
 		if len(d.pLens) > 0 {
@@ -81,13 +83,9 @@ func (d *datagramBufferedPipe) WriteTo(w io.Writer) (n int64, err error) {
 			}
 			if d.wtTimeout == 0 {
 				// if there hasn't been a scheduled broadcast
-				time.AfterFunc(delta, d.rwCond.Broadcast)
+				d.broadcastAfter(delta)
 			}
 		}
-		if d.wtTimeout != 0 {
-			d.rDeadline = time.Now().Add(d.wtTimeout)
-			time.AfterFunc(d.wtTimeout, d.rwCond.Broadcast)
-		}
 
 		if len(d.pLens) > 0 {
 			var dataLen int
@@ -100,6 +98,11 @@ func (d *datagramBufferedPipe) WriteTo(w io.Writer) (n int64, err error) {
 			}
 			d.rwCond.Broadcast()
 		} else {
+			if d.wtTimeout != 0 {
+				d.rDeadline = time.Now().Add(d.wtTimeout)
+				d.broadcastAfter(d.wtTimeout)
+			}
+
 			d.rwCond.Wait()
 		}
 	}
@@ -160,3 +163,10 @@ func (d *datagramBufferedPipe) SetWriteToTimeout(t time.Duration) {
 	d.wtTimeout = t
 	d.rwCond.Broadcast()
 }
+
+func (d *datagramBufferedPipe) broadcastAfter(t time.Duration) {
+	if d.timeoutTimer != nil {
+		d.timeoutTimer.Stop()
+	}
+	d.timeoutTimer = time.AfterFunc(t, d.rwCond.Broadcast)
+}

internal/multiplex/streamBufferedPipe.go

@@ -18,6 +18,8 @@ type streamBufferedPipe struct {
 	rwCond    *sync.Cond
 	rDeadline time.Time
 	wtTimeout time.Duration
+
+	timeoutTimer *time.Timer
 }
 
 func NewStreamBufferedPipe() *streamBufferedPipe {
@@ -42,7 +44,7 @@ func (p *streamBufferedPipe) Read(target []byte) (int, error) {
 			if d <= 0 {
 				return 0, ErrTimeout
 			}
-			time.AfterFunc(d, p.rwCond.Broadcast)
+			p.broadcastAfter(d)
 		}
 		if p.buf.Len() > 0 {
 			break
@@ -72,13 +74,9 @@ func (p *streamBufferedPipe) WriteTo(w io.Writer) (n int64, err error) {
 			}
 			if p.wtTimeout == 0 {
 				// if there hasn't been a scheduled broadcast
-				time.AfterFunc(d, p.rwCond.Broadcast)
+				p.broadcastAfter(d)
 			}
 		}
-		if p.wtTimeout != 0 {
-			p.rDeadline = time.Now().Add(p.wtTimeout)
-			time.AfterFunc(p.wtTimeout, p.rwCond.Broadcast)
-		}
 		if p.buf.Len() > 0 {
 			written, er := p.buf.WriteTo(w)
 			n += written
@@ -88,6 +86,10 @@ func (p *streamBufferedPipe) WriteTo(w io.Writer) (n int64, err error) {
 			}
 			p.rwCond.Broadcast()
 		} else {
+			if p.wtTimeout != 0 {
+				p.rDeadline = time.Now().Add(p.wtTimeout)
+				p.broadcastAfter(p.wtTimeout)
+			}
 			p.rwCond.Wait()
 		}
 	}
@@ -139,3 +141,10 @@ func (p *streamBufferedPipe) SetWriteToTimeout(d time.Duration) {
 	p.wtTimeout = d
 	p.rwCond.Broadcast()
 }
+
+func (p *streamBufferedPipe) broadcastAfter(d time.Duration) {
+	if p.timeoutTimer != nil {
+		p.timeoutTimer.Stop()
+	}
+	p.timeoutTimer = time.AfterFunc(d, p.rwCond.Broadcast)
+}