From 208a7f249f5f788237908108946e147cad1dd29c Mon Sep 17 00:00:00 2001 From: notsure2 Date: Sun, 6 Dec 2020 01:47:52 +0200 Subject: [PATCH 1/2] Fix memory leaking by extra timers. --- internal/multiplex/datagramBufferedPipe.go | 22 ++++++++++++++++------ internal/multiplex/streamBufferedPipe.go | 21 +++++++++++++++------ 2 files changed, 31 insertions(+), 12 deletions(-) diff --git a/internal/multiplex/datagramBufferedPipe.go b/internal/multiplex/datagramBufferedPipe.go index 863992b..9a99b80 100644 --- a/internal/multiplex/datagramBufferedPipe.go +++ b/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) +} diff --git a/internal/multiplex/streamBufferedPipe.go b/internal/multiplex/streamBufferedPipe.go index 47ee101..52ae512 100644 --- a/internal/multiplex/streamBufferedPipe.go +++ b/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) +} From 0327d0ffb391d0abd9cf67d3e0253cd5795f4099 Mon Sep 17 00:00:00 2001 From: Andy Wang Date: Sun, 6 Dec 2020 00:16:29 +0000 Subject: [PATCH 2/2] Further reduce the amount of calls to AfterFunc --- internal/multiplex/datagramBufferedPipe.go | 28 +++++++++++--------- internal/multiplex/streamBufferedPipe.go | 30 +++++++++++++--------- 2 files changed, 34 insertions(+), 24 deletions(-) diff --git a/internal/multiplex/datagramBufferedPipe.go b/internal/multiplex/datagramBufferedPipe.go index 9a99b80..e1a0462 100644 --- a/internal/multiplex/datagramBufferedPipe.go +++ b/internal/multiplex/datagramBufferedPipe.go @@ -42,17 +42,20 @@ func (d *datagramBufferedPipe) Read(target []byte) (int, error) { return 0, io.EOF } - if !d.rDeadline.IsZero() { - delta := time.Until(d.rDeadline) - if delta <= 0 { + hasRDeadline := !d.rDeadline.IsZero() + if hasRDeadline { + if time.Until(d.rDeadline) <= 0 { return 0, ErrTimeout } - d.broadcastAfter(delta) } if len(d.pLens) > 0 { break } + + if hasRDeadline { + d.broadcastAfter(time.Until(d.rDeadline)) + } d.rwCond.Wait() } dataLen := d.pLens[0] @@ -76,15 +79,12 @@ func (d *datagramBufferedPipe) WriteTo(w io.Writer) (n int64, err error) { if d.closed && len(d.pLens) == 0 { return 0, io.EOF } - if !d.rDeadline.IsZero() { - delta := time.Until(d.rDeadline) - if delta <= 0 { + + hasRDeadline := !d.rDeadline.IsZero() + if hasRDeadline { + if time.Until(d.rDeadline) <= 0 { return 0, ErrTimeout } - if d.wtTimeout == 0 { - // if there hasn't been a scheduled broadcast - d.broadcastAfter(delta) - } } if len(d.pLens) > 0 { @@ -98,7 +98,11 @@ func (d *datagramBufferedPipe) WriteTo(w io.Writer) (n int64, err error) { } d.rwCond.Broadcast() } else { - if d.wtTimeout != 0 { + if d.wtTimeout == 0 { + if hasRDeadline { + d.broadcastAfter(time.Until(d.rDeadline)) + } + } else { d.rDeadline = time.Now().Add(d.wtTimeout) d.broadcastAfter(d.wtTimeout) } diff --git a/internal/multiplex/streamBufferedPipe.go b/internal/multiplex/streamBufferedPipe.go index 52ae512..66dacec 100644 --- a/internal/multiplex/streamBufferedPipe.go +++ b/internal/multiplex/streamBufferedPipe.go @@ -39,16 +39,20 @@ func (p *streamBufferedPipe) Read(target []byte) (int, error) { if p.closed && p.buf.Len() == 0 { return 0, io.EOF } - if !p.rDeadline.IsZero() { - d := time.Until(p.rDeadline) - if d <= 0 { + + hasRDeadline := !p.rDeadline.IsZero() + if hasRDeadline { + if time.Until(p.rDeadline) <= 0 { return 0, ErrTimeout } - p.broadcastAfter(d) } if p.buf.Len() > 0 { break } + + if hasRDeadline { + p.broadcastAfter(time.Until(p.rDeadline)) + } p.rwCond.Wait() } n, err := p.buf.Read(target) @@ -67,15 +71,12 @@ func (p *streamBufferedPipe) WriteTo(w io.Writer) (n int64, err error) { if p.closed && p.buf.Len() == 0 { return 0, io.EOF } - if !p.rDeadline.IsZero() { - d := time.Until(p.rDeadline) - if d <= 0 { + + hasRDeadline := !p.rDeadline.IsZero() + if hasRDeadline { + if time.Until(p.rDeadline) <= 0 { return 0, ErrTimeout } - if p.wtTimeout == 0 { - // if there hasn't been a scheduled broadcast - p.broadcastAfter(d) - } } if p.buf.Len() > 0 { written, er := p.buf.WriteTo(w) @@ -86,10 +87,15 @@ func (p *streamBufferedPipe) WriteTo(w io.Writer) (n int64, err error) { } p.rwCond.Broadcast() } else { - if p.wtTimeout != 0 { + if p.wtTimeout == 0 { + if hasRDeadline { + p.broadcastAfter(time.Until(p.rDeadline)) + } + } else { p.rDeadline = time.Now().Add(p.wtTimeout) p.broadcastAfter(p.wtTimeout) } + p.rwCond.Wait() } }