Repository: pion/sctp
Pull request #56 (2019-08-17 04:04:10)

Fix routine leak
#### Description

writeLoop routine can be leaked if the connection is lost when there is no writing packet.
So, writeLoop should also be stopped if readLoop is stopped.
And, closeWriteLoopCh should be closed, even if netConn.Close() returned error.

#### Additional resources

After running the system using webrtc.PeerConnection for a while on the real environment, pprof could find routine leaks.
```
goroutine profile: total 92
79 @ 0x42dcbb 0x43da76 0x78de46 0x45b161
#	0x78de45	github.com/pion/sctp.(*Association).writeLoop+0x1b5	/go/pkg/mod/github.com/pion/sctp@v1.6.4/association.go:426

2 @ 0x42dcbb 0x42dd63 0x404e8d 0x404c65 0x79b63b 0x7ab3f6 0x45b161
#	0x79b63a	github.com/pion/sctp.(*Association).onRetransmissionFailure+0x41a	/go/pkg/mod/github.com/pion/sctp@v1.6.4/association.go:2003
#	0x7ab3f5	github.com/pion/sctp.(*rtxTimer).start.func1+0x175			/go/pkg/mod/github.com/pion/sctp@v1.6.4/rtx_timer.go:158
```
The number of the writeLoop routine seems increased after `retransmission failure: T1-init` on my environment.

Heap memory was also leaked like:
![image](https://user-images.githubusercontent.com/8390204/63206469-ea2c4f80-c0ef-11e9-8919-ff1c65e17bba.png)
and these memory leaks disappeared after applying this PR.

codecov-io: # [Codecov](https://codecov.io/gh/pion/sctp/pull/56?src=pr&el=h1) Report
> Merging [#56](https://codecov.io/gh/pion/sctp/pull/56?src=pr&el=desc) into [master](https://codecov.io/gh/pion/sctp/commit/99012aee4a5b5830ceb4942096cf4c4ba281f960?src=pr&el=desc) will **increase** coverage by `0.1%`.
> The diff coverage is `100%`.

[![Impacted file tree graph](https://codecov.io/gh/pion/sctp/pull/56/graphs/tree.svg?width=650&token=AB8S3U6xR9&height=150&src=pr)](https://codecov.io/gh/pion/sctp/pull/56?src=pr&el=tree)

```diff
@@            Coverage Diff            @@
##           master      #56     +/-   ##
=========================================
+ Coverage   75.68%   75.79%   +0.1%     
=========================================
  Files          43       43             
  Lines        2612     2611      -1     
=========================================
+ Hits         1977     1979      +2     
+ Misses        512      509      -3     
  Partials      123      123
```


| [Impacted Files](https://codecov.io/gh/pion/sctp/pull/56?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [association.go](https://codecov.io/gh/pion/sctp/pull/56/diff?src=pr&el=tree#diff-YXNzb2NpYXRpb24uZ28=) | `81.56% <100%> (+0.27%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/pion/sctp/pull/56?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/pion/sctp/pull/56?src=pr&el=footer). Last update [99012ae...c3b4b74](https://codecov.io/gh/pion/sctp/pull/56?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

at-wat: @enobufs I have addressed your comment. PTAL

Patch:
diff --git a/README.md b/README.md
index 1ee542f..4191671 100644
--- a/README.md
+++ b/README.md
@@ -44,6 +44,7 @@ Check out the **[contributing wiki](https://github.com/pion/webrtc/wiki/Contribu
 * [Antoine Baché](https://github.com/Antonito) - *SCTP Profiling*
 * [Cecylia Bocovich](https://github.com/cohosh) - *Fix SCTP reads*
 * [Hugo Arregui](https://github.com/hugoArregui)
+* [Atsushi Watanabe](https://github.com/at-wat)
 
 ### License
 MIT License - see [LICENSE](LICENSE) for full text
diff --git a/association.go b/association.go
index bff491a..9bbd20f 100644
--- a/association.go
+++ b/association.go
@@ -163,6 +163,8 @@ type Association struct {
 	closeWriteLoopCh     chan struct{}
 	handshakeCompletedCh chan error
 
+	closeWriteLoopOnce sync.Once
+
 	// local error
 	silentError error
 
@@ -346,14 +348,11 @@ func (a *Association) Close() error {
 	a.setState(closed)
 
 	err := a.netConn.Close()
-	if err != nil {
-		return err
-	}
 
 	a.closeAllTimers()
 
 	// awake writeLoop to exit
-	close(a.closeWriteLoopCh)
+	a.closeWriteLoopOnce.Do(func() { close(a.closeWriteLoopCh) })
 
 	// Wait for readLoop to end
 	<-a.readLoopCloseCh
@@ -364,7 +363,7 @@ func (a *Association) Close() error {
 	a.log.Debugf("[%s] stats nT3Timeouts : %d", a.name, a.stats.getNumT3Timeouts())
 	a.log.Debugf("[%s] stats nAckTimeouts: %d", a.name, a.stats.getNumAckTimeouts())
 	a.log.Debugf("[%s] stats nFastRetrans: %d", a.name, a.stats.getNumFastRetrans())
-	return nil
+	return err
 }
 
 func (a *Association) closeAllTimers() {
@@ -378,6 +377,10 @@ func (a *Association) closeAllTimers() {
 func (a *Association) readLoop() {
 	var closeErr error
 	defer func() {
+		// also stop writeLoop, otherwise writeLoop can be leaked
+		// if connection is lost when there is no writing packet.
+		a.closeWriteLoopOnce.Do(func() { close(a.closeWriteLoopCh) })
+
 		a.lock.Lock()
 		for _, s := range a.streams {
 			a.unregisterStream(s, closeErr)


Test patch:
diff --git a/association_test.go b/association_test.go
index 8974e6f..4802829 100644
--- a/association_test.go
+++ b/association_test.go
@@ -8,6 +8,7 @@ import (
 	"math"
 	"math/rand"
 	"net"
+	"runtime"
 	"sync"
 	"testing"
 	"time"
@@ -2082,3 +2083,108 @@ func TestAssocReset(t *testing.T) {
 		closeAssociationPair(br, a0, a1)
 	})
 }
+
+type fakeEchoConn struct {
+	echo     chan []byte
+	done     chan struct{}
+	closed   chan struct{}
+	once     sync.Once
+	errClose error
+	mu       sync.Mutex
+}
+
+func newFakeEchoConn(errClose error) *fakeEchoConn {
+	return &fakeEchoConn{
+		echo:     make(chan []byte, 1),
+		done:     make(chan struct{}),
+		closed:   make(chan struct{}),
+		errClose: errClose,
+	}
+}
+func (c *fakeEchoConn) Read(b []byte) (int, error) {
+	r, ok := <-c.echo
+	if ok {
+		copy(b, r)
+		c.once.Do(func() { close(c.done) })
+		return len(r), nil
+	}
+	return 0, io.EOF
+}
+func (c *fakeEchoConn) Write(b []byte) (int, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	select {
+	case <-c.closed:
+		return 0, io.EOF
+	default:
+	}
+	c.echo <- b
+	return len(b), nil
+}
+func (c *fakeEchoConn) Close() error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	close(c.echo)
+	close(c.closed)
+	return c.errClose
+}
+func (c *fakeEchoConn) LocalAddr() net.Addr                { return nil }
+func (c *fakeEchoConn) RemoteAddr() net.Addr               { return nil }
+func (c *fakeEchoConn) SetDeadline(t time.Time) error      { return nil }
+func (c *fakeEchoConn) SetReadDeadline(t time.Time) error  { return nil }
+func (c *fakeEchoConn) SetWriteDeadline(t time.Time) error { return nil }
+
+func TestRoutineLeak(t *testing.T) {
+	loggerFactory := logging.NewDefaultLoggerFactory()
+	t.Run("Close failed", func(t *testing.T) {
+		runtime.GC()
+		n0 := runtime.NumGoroutine()
+
+		conn := newFakeEchoConn(io.EOF)
+		a, err := Client(Config{NetConn: conn, LoggerFactory: loggerFactory})
+		assert.Equal(t, nil, err, "errored to initialize Client")
+
+		<-conn.done
+
+		err = a.Close()
+		assert.Equal(t, io.EOF, err, "Close() should fail with EOF")
+
+		select {
+		case _, ok := <-a.closeWriteLoopCh:
+			if ok {
+				t.Errorf("closeWriteLoopCh is expected to be closed, but received signal")
+			}
+		default:
+			t.Errorf("closeWriteLoopCh is expected to be closed, but not")
+		}
+		_ = a
+		runtime.GC()
+		assert.Equal(t, n0, runtime.NumGoroutine(), "goroutine is leaked")
+	})
+	t.Run("Connection closed by remote host", func(t *testing.T) {
+		runtime.GC()
+		n0 := runtime.NumGoroutine()
+
+		conn := newFakeEchoConn(nil)
+		a, err := Client(Config{NetConn: conn, LoggerFactory: loggerFactory})
+		assert.Equal(t, nil, err, "errored to initialize Client")
+
+		<-conn.done
+
+		err = conn.Close() // close connection
+		assert.Equal(t, nil, err, "fake connection returned unexpected error")
+		<-conn.closed
+		<-time.After(10 * time.Millisecond) // switch context to make read/write loops finished
+
+		select {
+		case _, ok := <-a.closeWriteLoopCh:
+			if ok {
+				t.Errorf("closeWriteLoopCh is expected to be closed, but received signal")
+			}
+		default:
+			t.Errorf("closeWriteLoopCh is expected to be closed, but not")
+		}
+		runtime.GC()
+		assert.Equal(t, n0, runtime.NumGoroutine(), "goroutine is leaked")
+	})
+}