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

Merge pull request #138 from notsure2/explicit-stop-timer 

Optimise calls to time.AfterFunc in buffered pipe implemenations to reduce memory usage
This commit is contained in:
Andy Wang 2020-12-06 00:53:30 +00:00 committed by GitHub
parent c0040f20c3 0327d0ffb3
commit 0482d28f86
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 59 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

44
internal/multiplex/datagramBufferedPipe.go
View File

@ -20,6 +20,8 @@ type datagramBufferedPipe struct {
rwCond *sync.Cond
wtTimeout time.Duration
rDeadline time.Time
timeoutTimer *time.Timer
}
func NewDatagramBufferedPipe() *datagramBufferedPipe {
@ -40,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
}
time.AfterFunc(delta, d.rwCond.Broadcast)
}
if len(d.pLens) > 0 {
break
}
if hasRDeadline {
d.broadcastAfter(time.Until(d.rDeadline))
}
d.rwCond.Wait()
}
dataLen := d.pLens[0]
@ -74,19 +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
time.AfterFunc(delta, d.rwCond.Broadcast)
}
}
if d.wtTimeout != 0 {
d.rDeadline = time.Now().Add(d.wtTimeout)
time.AfterFunc(d.wtTimeout, d.rwCond.Broadcast)
}
if len(d.pLens) > 0 {
@ -100,6 +98,15 @@ func (d *datagramBufferedPipe) WriteTo(w io.Writer) (n int64, err error) {
}
d.rwCond.Broadcast()
} else {
if d.wtTimeout == 0 {
if hasRDeadline {
d.broadcastAfter(time.Until(d.rDeadline))
}
} else {
d.rDeadline = time.Now().Add(d.wtTimeout)
d.broadcastAfter(d.wtTimeout)
}
d.rwCond.Wait()
}
}
@ -160,3 +167,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)
}

45
internal/multiplex/streamBufferedPipe.go
View File

@ -18,6 +18,8 @@ type streamBufferedPipe struct {
rwCond *sync.Cond
rDeadline time.Time
wtTimeout time.Duration
timeoutTimer *time.Timer
}
func NewStreamBufferedPipe() *streamBufferedPipe {
@ -37,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
}
time.AfterFunc(d, p.rwCond.Broadcast)
}
if p.buf.Len() > 0 {
break
}
if hasRDeadline {
p.broadcastAfter(time.Until(p.rDeadline))
}
p.rwCond.Wait()
}
n, err := p.buf.Read(target)
@ -65,19 +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
time.AfterFunc(d, p.rwCond.Broadcast)
}
}
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)
@ -88,6 +87,15 @@ func (p *streamBufferedPipe) WriteTo(w io.Writer) (n int64, err error) {
}
p.rwCond.Broadcast()
} else {
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()
}
}
@ -139,3 +147,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)
}