Skip to content

abolish an extra goroutine #501

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 5 commits into
base: master
Choose a base branch
from
Open
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Next Next commit
abolish an extra goroutine
  • Loading branch information
andriibeee committed Nov 13, 2024
commit ad3f0c2cbd6c11bc8adfc16a38e28f6f258546f0
6 changes: 0 additions & 6 deletions close.go
Original file line number Diff line number Diff line change
@@ -232,12 +232,6 @@ func (c *Conn) waitGoroutines() error {
t := time.NewTimer(time.Second * 15)
defer t.Stop()

select {
case <-c.timeoutLoopDone:
case <-t.C:
return errors.New("failed to wait for timeoutLoop goroutine to exit")
}

c.closeReadMu.Lock()
closeRead := c.closeReadCtx != nil
c.closeReadMu.Unlock()
63 changes: 36 additions & 27 deletions conn.go
Original file line number Diff line number Diff line change
@@ -52,9 +52,8 @@
br *bufio.Reader
bw *bufio.Writer

readTimeout chan context.Context
writeTimeout chan context.Context
timeoutLoopDone chan struct{}
readTimeoutCloser atomic.Value
writeTimeoutCloser atomic.Value
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think atomic.Pointer would be cleaner. atomic.Value still stores the value as a pointer, it's just not explicit. The upside of atomic.Pointer is that it requires no type casting.


// Read state.
readMu *mu
@@ -104,10 +103,6 @@
br: cfg.br,
bw: cfg.bw,

readTimeout: make(chan context.Context),
writeTimeout: make(chan context.Context),
timeoutLoopDone: make(chan struct{}),

closed: make(chan struct{}),
activePings: make(map[string]chan<- struct{}),
}
@@ -133,8 +128,6 @@
c.close()
})

go c.timeoutLoop()

return c
}

@@ -164,26 +157,42 @@
return err
}

func (c *Conn) timeoutLoop() {
defer close(c.timeoutLoopDone)
func (c *Conn) setupWriteTimeout(ctx context.Context) {
hammerTime := context.AfterFunc(ctx, func() {

Check failure on line 161 in conn.go

GitHub Actions / bench

undefined: context.AfterFunc
c.close()
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I wonder if it'd be possible for this function to also remove itself from c.writeTimeoutCloser? Perhaps it can put a nil in it / or swap it with a nil and run what was in it if it's not nil. That way we'll let GC collect the function once its ran. Same below.

Well, it just needs to call clearWriteTimeout().

})

readCtx := context.Background()
writeCtx := context.Background()
if closer := c.writeTimeoutCloser.Swap(hammerTime); closer != nil {
if fn, ok := closer.(func() bool); ok {
fn()
}
}
}

for {
select {
case <-c.closed:
return

case writeCtx = <-c.writeTimeout:
case readCtx = <-c.readTimeout:

case <-readCtx.Done():
c.close()
return
case <-writeCtx.Done():
c.close()
return
func (c *Conn) clearWriteTimeout() {
if closer := c.writeTimeoutCloser.Load(); closer != nil {
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Shouldn't this be a swap too to remove the reference to the function to let GC take it? Same below.

if fn, ok := closer.(func() bool); ok {
fn()
}
}
}

func (c *Conn) setupReadTimeout(ctx context.Context) {
hammerTime := context.AfterFunc(ctx, func() {

Check failure on line 181 in conn.go

GitHub Actions / bench

undefined: context.AfterFunc
defer c.close()
})

if closer := c.readTimeoutCloser.Swap(hammerTime); closer != nil {
if fn, ok := closer.(func() bool); ok {
fn()
}
}
}

func (c *Conn) clearReadTimeout() {
if closer := c.readTimeoutCloser.Load(); closer != nil {
if fn, ok := closer.(func() bool); ok {
fn()
}
}
}
12 changes: 8 additions & 4 deletions read.go
Original file line number Diff line number Diff line change
@@ -221,7 +221,8 @@ func (c *Conn) readFrameHeader(ctx context.Context) (header, error) {
select {
case <-c.closed:
return header{}, net.ErrClosed
case c.readTimeout <- ctx:
default:
c.setupReadTimeout(ctx)
}

h, err := readFrameHeader(c.br, c.readHeaderBuf[:])
@@ -239,7 +240,8 @@ func (c *Conn) readFrameHeader(ctx context.Context) (header, error) {
select {
case <-c.closed:
return header{}, net.ErrClosed
case c.readTimeout <- context.Background():
default:
c.clearReadTimeout()
}

return h, nil
@@ -249,7 +251,8 @@ func (c *Conn) readFramePayload(ctx context.Context, p []byte) (int, error) {
select {
case <-c.closed:
return 0, net.ErrClosed
case c.readTimeout <- ctx:
default:
c.setupReadTimeout(ctx)
}

n, err := io.ReadFull(c.br, p)
@@ -267,7 +270,8 @@ func (c *Conn) readFramePayload(ctx context.Context, p []byte) (int, error) {
select {
case <-c.closed:
return n, net.ErrClosed
case c.readTimeout <- context.Background():
default:
c.clearReadTimeout()
}

return n, err
6 changes: 4 additions & 2 deletions write.go
Original file line number Diff line number Diff line change
@@ -252,7 +252,8 @@ func (c *Conn) writeFrame(ctx context.Context, fin bool, flate bool, opcode opco
select {
case <-c.closed:
return 0, net.ErrClosed
case c.writeTimeout <- ctx:
default:
c.setupWriteTimeout(ctx)
}

defer func() {
@@ -309,7 +310,8 @@ func (c *Conn) writeFrame(ctx context.Context, fin bool, flate bool, opcode opco
return n, nil
}
return n, net.ErrClosed
case c.writeTimeout <- context.Background():
default:
c.clearWriteTimeout()
}

return n, nil
Loading
Oops, something went wrong.