diff --git a/internal/multiplex/datagramBufferedPipe.go b/internal/multiplex/datagramBufferedPipe.go index eb7be41..863992b 100644 --- a/internal/multiplex/datagramBufferedPipe.go +++ b/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() - }() -} diff --git a/internal/multiplex/streamBufferedPipe.go b/internal/multiplex/streamBufferedPipe.go index abfb27f..47ee101 100644 --- a/internal/multiplex/streamBufferedPipe.go +++ b/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() - }() -}