Reduce memory allocated in DataChannel.readLoop

See pion#1516

This patch preserves the semantics of the OnMessage handler and is
more safe but less efficient than the patch first described in pion#1516.

As expected, when run with GOMAXPROCS=1 there is opportunity for
re-use and syn.Pool is pure overhead.  As allocs/op increases due to
sync.Pool overhead but B/op and ns/op improve drastically.

    $ git checkout datachannel.go && \
      got test -bench=DataChannelSend -run=XXX \
      -benchmem -cpu=1,2,4,8 -count=10 > original.txt
    $ git checkout datachannel.go && git apply pool.patch && \
      go test -bench=DataChannelSend -run=XXX \
    	-benchmem -cpu=1,2,4,8 -count=10 > patched.txt

    $ benchstat original.txt patched.txt
    name               old time/op    new time/op    delta
    DCSend      1.45µs ±35%    1.24µs ±40%      ~     (p=0.211 n=9+10)
    DCSend-2     885ns ±14%    2221ns ±44%  +150.87%  (p=0.000 n=10+9)
    DCSend-4    5.19µs ±15%    3.27µs ±16%   -37.13%  (p=0.000 n=9+10)
    DCSend-8    8.07µs ±22%    3.97µs ± 4%   -50.78%  (p=0.000 n=10+8)

    name               old alloc/op   new alloc/op   delta
    DCSend      1.67kB ±72%    0.31kB ±12%   -81.55%  (p=0.000 n=9+10)
    DCSend-2    1.80kB ±21%    0.34kB ±14%   -81.14%  (p=0.000 n=10+8)
    DCSend-4    1.47kB ±14%    1.40kB ± 0%    -4.39%  (p=0.031 n=10+8)
    DCSend-8    2.18kB ±14%    1.45kB ± 0%   -33.57%  (p=0.000 n=9+8)

    name               old allocs/op  new allocs/op  delta
    DCSend        6.00 ± 0%      6.80 ±18%   +13.33%  (p=0.008 n=8+10)
    DCSend-2      6.00 ± 0%      8.11 ±36%   +35.19%  (p=0.000 n=10+9)
    DCSend-4      7.00 ± 0%     37.00 ± 0%  +428.57%  (p=0.000 n=10+10)
    DCSend-8      7.70 ± 9%     39.00 ± 0%  +406.49%  (p=0.000 n=10+7)

datachannel.go

@@ -311,11 +311,17 @@ func (d *DataChannel) onError(err error) {
 	}
 }
 
+// See https://github.com/pion/webrtc/issues/1516
+var rlBufPool = sync.Pool{New: func() interface{} {
+	return make([]byte, dataChannelBufferSize)
+}}
+
 func (d *DataChannel) readLoop() {
 	for {
-		buffer := make([]byte, dataChannelBufferSize)
+		buffer := rlBufPool.Get().([]byte)
 		n, isString, err := d.dataChannel.ReadDataChannel(buffer)
 		if err != nil {
+			rlBufPool.Put(buffer)
 			d.setReadyState(DataChannelStateClosed)
 			if err != io.EOF {
 				d.onError(err)
@@ -324,7 +330,10 @@ func (d *DataChannel) readLoop() {
 			return
 		}
 
-		d.onMessage(DataChannelMessage{Data: buffer[:n], IsString: isString})
+		m := DataChannelMessage{Data: make([]byte, n), IsString: isString}
+		copy(m.Data, buffer[:n])
+		d.onMessage(m)
+		rlBufPool.Put(buffer)
 	}
 }
 

datachannel_test.go

@@ -15,7 +15,7 @@ import (
 // bindings this is a requirement).
 const expectedLabel = "data"
 
-func closePairNow(t *testing.T, pc1, pc2 io.Closer) {
+func closePairNow(t testing.TB, pc1, pc2 io.Closer) {
 	var fail bool
 	if err := pc1.Close(); err != nil {
 		t.Errorf("Failed to close PeerConnection: %v", err)
@@ -63,6 +63,48 @@ func closeReliabilityParamTest(t *testing.T, pc1, pc2 *PeerConnection, done chan
 	closePair(t, pc1, pc2, done)
 }
 
+// See https://github.com/pion/webrtc/issues/1516
+func BenchmarkDataChannelSend(b *testing.B) {
+	offerPC, answerPC, err := newPair()
+	if err != nil {
+		b.Fatalf("Failed to create a PC pair for testing")
+	}
+
+	var (
+		//received   int
+		answerOpen = make(chan bool)
+		done       = make(chan bool)
+	)
+	answerPC.OnDataChannel(func(d *DataChannel) {
+		if d.Label() != expectedLabel {
+			return
+		}
+		d.OnOpen(func() {
+			answerOpen <- true
+		})
+		//d.OnMessage(func(_ DataChannelMessage) {
+		//	received += 1
+		//})
+	})
+
+	dc, err := offerPC.CreateDataChannel(expectedLabel, nil)
+	assert.NoError(b, err)
+
+	dc.OnOpen(func() {
+		<-answerOpen
+		for n := 0; n < b.N; n++ {
+			if err := dc.SendText("Ping"); err != nil {
+				b.Fatalf("Unexpected error sending data: %v", err)
+			}
+		}
+		done <- true
+	})
+
+	assert.NoError(b, signalPair(offerPC, answerPC))
+	<-done
+	closePairNow(b, offerPC, answerPC)
+}
+
 func TestDataChannel_Open(t *testing.T) {
 	t.Run("handler should be called once", func(t *testing.T) {
 		report := test.CheckRoutines(t)