Revert "Prevent time.Timer memory leak by using a singleton timer in bufferedPipes. Fix #137"

This reverts commit 4baca256

internal/multiplex/datagramBufferedPipe.go

@@ -20,14 +20,11 @@ type datagramBufferedPipe struct {
 	rwCond    *sync.Cond
 	wtTimeout time.Duration
 	rDeadline time.Time
-
-	timer *time.Timer
 }
 
 func NewDatagramBufferedPipe() *datagramBufferedPipe {
 	d := &datagramBufferedPipe{
 		rwCond: sync.NewCond(&sync.Mutex{}),
-		timer:  time.NewTimer(0),
 	}
 	return d
 }
@@ -48,7 +45,7 @@ func (d *datagramBufferedPipe) Read(target []byte) (int, error) {
 			if delta <= 0 {
 				return 0, ErrTimeout
 			}
-			d.broadcastAfter(delta)
+			time.AfterFunc(delta, d.rwCond.Broadcast)
 		}
 
 		if len(d.pLens) > 0 {
@@ -84,12 +81,12 @@ func (d *datagramBufferedPipe) WriteTo(w io.Writer) (n int64, err error) {
 			}
 			if d.wtTimeout == 0 {
 				// if there hasn't been a scheduled broadcast
-				d.broadcastAfter(delta)
+				time.AfterFunc(delta, d.rwCond.Broadcast)
 			}
 		}
 		if d.wtTimeout != 0 {
 			d.rDeadline = time.Now().Add(d.wtTimeout)
-			d.broadcastAfter(d.wtTimeout)
+			time.AfterFunc(d.wtTimeout, d.rwCond.Broadcast)
 		}
 
 		if len(d.pLens) > 0 {
@@ -163,15 +160,3 @@ func (d *datagramBufferedPipe) SetWriteToTimeout(t time.Duration) {
 	d.wtTimeout = t
 	d.rwCond.Broadcast()
 }
-
-func (d *datagramBufferedPipe) broadcastAfter(delta time.Duration) {
-	// d.rwCond.L must be held, otherwise the following timer operations will race
-	if !d.timer.Stop() {
-		<-d.timer.C
-	}
-	d.timer.Reset(delta)
-	go func() {
-		<-d.timer.C
-		d.rwCond.Broadcast()
-	}()
-}

internal/multiplex/streamBufferedPipe.go

@@ -18,14 +18,11 @@ type streamBufferedPipe struct {
 	rwCond    *sync.Cond
 	rDeadline time.Time
 	wtTimeout time.Duration
-
-	timer *time.Timer
 }
 
 func NewStreamBufferedPipe() *streamBufferedPipe {
 	p := &streamBufferedPipe{
 		rwCond: sync.NewCond(&sync.Mutex{}),
-		timer:  time.NewTimer(0),
 	}
 	return p
 }
@@ -45,7 +42,7 @@ func (p *streamBufferedPipe) Read(target []byte) (int, error) {
 			if d <= 0 {
 				return 0, ErrTimeout
 			}
-			p.broadcastAfter(d)
+			time.AfterFunc(d, p.rwCond.Broadcast)
 		}
 		if p.buf.Len() > 0 {
 			break
@@ -75,12 +72,12 @@ func (p *streamBufferedPipe) WriteTo(w io.Writer) (n int64, err error) {
 			}
 			if p.wtTimeout == 0 {
 				// if there hasn't been a scheduled broadcast
-				p.broadcastAfter(d)
+				time.AfterFunc(d, p.rwCond.Broadcast)
 			}
 		}
 		if p.wtTimeout != 0 {
 			p.rDeadline = time.Now().Add(p.wtTimeout)
-			p.broadcastAfter(p.wtTimeout)
+			time.AfterFunc(p.wtTimeout, p.rwCond.Broadcast)
 		}
 		if p.buf.Len() > 0 {
 			written, er := p.buf.WriteTo(w)
@@ -142,15 +139,3 @@ func (p *streamBufferedPipe) SetWriteToTimeout(d time.Duration) {
 	p.wtTimeout = d
 	p.rwCond.Broadcast()
 }
-
-func (p *streamBufferedPipe) broadcastAfter(d time.Duration) {
-	// p.rwCond.L must be held, otherwise the following timer operations will race
-	if !p.timer.Stop() {
-		<-p.timer.C
-	}
-	p.timer.Reset(d)
-	go func() {
-		<-p.timer.C
-		p.rwCond.Broadcast()
-	}()
-}