kirillius
/
Cloak
mirror of https://github.com/cbeuw/Cloak
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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

This commit is contained in:
Andy Wang 2020-12-05 12:04:55 +00:00
parent 39c06a6e1d
commit 4baca256f7
No known key found for this signature in database
GPG Key ID: 181B49F9F38F3374
2 changed files with 36 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
internal/multiplex/datagramBufferedPipe.go
View File

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

21
internal/multiplex/streamBufferedPipe.go
View File

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