Repository: pion/interceptor
Pull request #102 (2022-02-21 08:29:58)

DisableCopy option for NackResponder
also decreased default buffer size to a more reasonable 1024

fixes https://github.com/pion/webrtc/issues/2097

codecov[bot]: # [Codecov](https://codecov.io/gh/pion/interceptor/pull/102?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) Report
> Merging [#102](https://codecov.io/gh/pion/interceptor/pull/102?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) (d39e267) into [master](https://codecov.io/gh/pion/interceptor/commit/347f59af87f499363f137f0d653a2d4e1c635b29?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) (347f59a) will **decrease** coverage by `0.14%`.
> The diff coverage is `90.47%`.

[![Impacted file tree graph](https://codecov.io/gh/pion/interceptor/pull/102/graphs/tree.svg?width=650&height=150&src=pr&token=FNH9xUG6AX&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion)](https://codecov.io/gh/pion/interceptor/pull/102?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion)

```diff
@@            Coverage Diff             @@
##           master     #102      +/-   ##
==========================================
- Coverage   79.11%   78.97%   -0.15%     
==========================================
  Files          51       51              
  Lines        2452     2468      +16     
==========================================
+ Hits         1940     1949       +9     
- Misses        422      428       +6     
- Partials       90       91       +1     
```

| Flag | Coverage Δ | |
|---|---|---|
| go | `78.97% <90.47%> (+0.05%)` | :arrow_up: |
| wasm | `?` | |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/pion/interceptor/pull/102?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) | Coverage Δ | |
|---|---|---|
| [pkg/nack/retainable\_packet.go](https://codecov.io/gh/pion/interceptor/pull/102/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL25hY2svcmV0YWluYWJsZV9wYWNrZXQuZ28=) | `86.76% <77.77%> (-1.38%)` | :arrow_down: |
| [pkg/nack/responder\_interceptor.go](https://codecov.io/gh/pion/interceptor/pull/102/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL25hY2svcmVzcG9uZGVyX2ludGVyY2VwdG9yLmdv) | `63.51% <100.00%> (+1.01%)` | :arrow_up: |
| [pkg/nack/responder\_option.go](https://codecov.io/gh/pion/interceptor/pull/102/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL25hY2svcmVzcG9uZGVyX29wdGlvbi5nbw==) | `100.00% <100.00%> (ø)` | |
| [pkg/gcc/arrival\_group\_accumulator.go](https://codecov.io/gh/pion/interceptor/pull/102/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL2djYy9hcnJpdmFsX2dyb3VwX2FjY3VtdWxhdG9yLmdv) | `90.62% <0.00%> (-9.38%)` | :arrow_down: |
| [pkg/gcc/delay\_based\_bwe.go](https://codecov.io/gh/pion/interceptor/pull/102/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL2djYy9kZWxheV9iYXNlZF9id2UuZ28=) | `83.33% <0.00%> (-3.04%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/pion/interceptor/pull/102?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/pion/interceptor/pull/102?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion). Last update [347f59a...d39e267](https://codecov.io/gh/pion/interceptor/pull/102?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion).


Patch:
diff --git a/pkg/nack/responder_interceptor.go b/pkg/nack/responder_interceptor.go
index 9b49f11..7f3123f 100644
--- a/pkg/nack/responder_interceptor.go
+++ b/pkg/nack/responder_interceptor.go
@@ -14,13 +14,16 @@ type ResponderInterceptorFactory struct {
 	opts []ResponderOption
 }
 
+type packetFactory interface {
+	NewPacket(header *rtp.Header, payload []byte) (*retainablePacket, error)
+}
+
 // NewInterceptor constructs a new ResponderInterceptor
 func (r *ResponderInterceptorFactory) NewInterceptor(id string) (interceptor.Interceptor, error) {
 	i := &ResponderInterceptor{
-		size:      8192,
-		log:       logging.NewDefaultLoggerFactory().NewLogger("nack_responder"),
-		streams:   map[uint32]*localStream{},
-		packetMan: newPacketManager(),
+		size:    1024,
+		log:     logging.NewDefaultLoggerFactory().NewLogger("nack_responder"),
+		streams: map[uint32]*localStream{},
 	}
 
 	for _, opt := range r.opts {
@@ -29,6 +32,10 @@ func (r *ResponderInterceptorFactory) NewInterceptor(id string) (interceptor.Int
 		}
 	}
 
+	if i.packetFactory == nil {
+		i.packetFactory = newPacketManager()
+	}
+
 	if _, err := newSendBuffer(i.size); err != nil {
 		return nil, err
 	}
@@ -39,9 +46,9 @@ func (r *ResponderInterceptorFactory) NewInterceptor(id string) (interceptor.Int
 // ResponderInterceptor responds to nack feedback messages
 type ResponderInterceptor struct {
 	interceptor.NoOp
-	size      uint16
-	log       logging.LeveledLogger
-	packetMan *packetManager
+	size          uint16
+	log           logging.LeveledLogger
+	packetFactory packetFactory
 
 	streams   map[uint32]*localStream
 	streamsMu sync.Mutex
@@ -100,7 +107,7 @@ func (n *ResponderInterceptor) BindLocalStream(info *interceptor.StreamInfo, wri
 	n.streamsMu.Unlock()
 
 	return interceptor.RTPWriterFunc(func(header *rtp.Header, payload []byte, attributes interceptor.Attributes) (int, error) {
-		pkt, err := n.packetMan.NewPacket(header, payload)
+		pkt, err := n.packetFactory.NewPacket(header, payload)
 		if err != nil {
 			return 0, err
 		}
diff --git a/pkg/nack/responder_option.go b/pkg/nack/responder_option.go
index 7ad52c8..33c3dfc 100644
--- a/pkg/nack/responder_option.go
+++ b/pkg/nack/responder_option.go
@@ -21,3 +21,12 @@ func ResponderLog(log logging.LeveledLogger) ResponderOption {
 		return nil
 	}
 }
+
+// DisableCopy bypasses copy of underlying packets. It should be used when
+// you are not re-using underlying buffers of packets that have been written
+func DisableCopy() ResponderOption {
+	return func(s *ResponderInterceptor) error {
+		s.packetFactory = &noOpPacketFactory{}
+		return nil
+	}
+}
diff --git a/pkg/nack/retainable_packet.go b/pkg/nack/retainable_packet.go
index 89e9cb6..a27fff2 100644
--- a/pkg/nack/retainable_packet.go
+++ b/pkg/nack/retainable_packet.go
@@ -60,6 +60,22 @@ func (m *packetManager) releasePacket(header *rtp.Header, payload *[]byte) {
 	}
 }
 
+type noOpPacketFactory struct {
+}
+
+func (f *noOpPacketFactory) NewPacket(header *rtp.Header, payload []byte) (*retainablePacket, error) {
+	return &retainablePacket{
+		onRelease: f.releasePacket,
+		count:     1,
+		header:    header,
+		payload:   payload,
+	}, nil
+}
+
+func (f *noOpPacketFactory) releasePacket(header *rtp.Header, payload *[]byte) {
+	// no-op
+}
+
 type retainablePacket struct {
 	onRelease func(*rtp.Header, *[]byte)
 


Test patch:
diff --git a/pkg/nack/responder_interceptor_test.go b/pkg/nack/responder_interceptor_test.go
index 42f125b..fa1d881 100644
--- a/pkg/nack/responder_interceptor_test.go
+++ b/pkg/nack/responder_interceptor_test.go
@@ -9,62 +9,85 @@ import (
 	"github.com/pion/logging"
 	"github.com/pion/rtcp"
 	"github.com/pion/rtp"
-	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestResponderInterceptor(t *testing.T) {
-	f, err := NewResponderInterceptor(
-		ResponderSize(8),
-		ResponderLog(logging.NewDefaultLoggerFactory().NewLogger("test")),
-	)
-	assert.NoError(t, err)
-
-	i, err := f.NewInterceptor("")
-	assert.NoError(t, err)
-
-	stream := test.NewMockStream(&interceptor.StreamInfo{
-		SSRC:         1,
-		RTCPFeedback: []interceptor.RTCPFeedback{{Type: "nack"}},
-	}, i)
-	defer func() {
-		assert.NoError(t, stream.Close())
-	}()
-
-	for _, seqNum := range []uint16{10, 11, 12, 14, 15} {
-		assert.NoError(t, stream.WriteRTP(&rtp.Packet{Header: rtp.Header{SequenceNumber: seqNum}}))
-
-		select {
-		case p := <-stream.WrittenRTP():
-			assert.Equal(t, seqNum, p.SequenceNumber)
-		case <-time.After(10 * time.Millisecond):
-			t.Fatal("written rtp packet not found")
-		}
-	}
-
-	stream.ReceiveRTCP([]rtcp.Packet{
-		&rtcp.TransportLayerNack{
-			MediaSSRC:  1,
-			SenderSSRC: 2,
-			Nacks: []rtcp.NackPair{
-				{PacketID: 11, LostPackets: 0b1011}, // sequence numbers: 11, 12, 13, 15
+	tests := []struct {
+		name string
+		opts []ResponderOption
+	}{
+		{
+			name: "with copy",
+			opts: []ResponderOption{
+				ResponderSize(8),
+				ResponderLog(logging.NewDefaultLoggerFactory().NewLogger("test")),
+			},
+		},
+		{
+			name: "without copy",
+			opts: []ResponderOption{
+				ResponderSize(8),
+				ResponderLog(logging.NewDefaultLoggerFactory().NewLogger("test")),
+				DisableCopy(),
 			},
 		},
-	})
-
-	// seq number 13 was never sent, so it can't be resent
-	for _, seqNum := range []uint16{11, 12, 15} {
-		select {
-		case p := <-stream.WrittenRTP():
-			assert.Equal(t, seqNum, p.SequenceNumber)
-		case <-time.After(10 * time.Millisecond):
-			t.Fatal("written rtp packet not found")
-		}
 	}
 
-	select {
-	case p := <-stream.WrittenRTP():
-		t.Errorf("no more rtp packets expected, found sequence number: %v", p.SequenceNumber)
-	case <-time.After(10 * time.Millisecond):
+	for _, item := range tests {
+		item := item
+		t.Run(item.name, func(t *testing.T) {
+			f, err := NewResponderInterceptor(item.opts...)
+			require.NoError(t, err)
+
+			i, err := f.NewInterceptor("")
+			require.NoError(t, err)
+
+			stream := test.NewMockStream(&interceptor.StreamInfo{
+				SSRC:         1,
+				RTCPFeedback: []interceptor.RTCPFeedback{{Type: "nack"}},
+			}, i)
+			defer func() {
+				require.NoError(t, stream.Close())
+			}()
+
+			for _, seqNum := range []uint16{10, 11, 12, 14, 15} {
+				require.NoError(t, stream.WriteRTP(&rtp.Packet{Header: rtp.Header{SequenceNumber: seqNum}}))
+
+				select {
+				case p := <-stream.WrittenRTP():
+					require.Equal(t, seqNum, p.SequenceNumber)
+				case <-time.After(10 * time.Millisecond):
+					t.Fatal("written rtp packet not found")
+				}
+			}
+
+			stream.ReceiveRTCP([]rtcp.Packet{
+				&rtcp.TransportLayerNack{
+					MediaSSRC:  1,
+					SenderSSRC: 2,
+					Nacks: []rtcp.NackPair{
+						{PacketID: 11, LostPackets: 0b1011}, // sequence numbers: 11, 12, 13, 15
+					},
+				},
+			})
+
+			// seq number 13 was never sent, so it can't be resent
+			for _, seqNum := range []uint16{11, 12, 15} {
+				select {
+				case p := <-stream.WrittenRTP():
+					require.Equal(t, seqNum, p.SequenceNumber)
+				case <-time.After(10 * time.Millisecond):
+					t.Fatal("written rtp packet not found")
+				}
+			}
+
+			select {
+			case p := <-stream.WrittenRTP():
+				t.Errorf("no more rtp packets expected, found sequence number: %v", p.SequenceNumber)
+			case <-time.After(10 * time.Millisecond):
+			}
+		})
 	}
 }
 
@@ -72,7 +95,20 @@ func TestResponderInterceptor_InvalidSize(t *testing.T) {
 	f, _ := NewResponderInterceptor(ResponderSize(5))
 
 	_, err := f.NewInterceptor("")
-	assert.Error(t, err, ErrInvalidSize)
+	require.Error(t, err, ErrInvalidSize)
+}
+
+func TestResponderInterceptor_DisableCopy(t *testing.T) {
+	f, err := NewResponderInterceptor(
+		ResponderSize(8),
+		ResponderLog(logging.NewDefaultLoggerFactory().NewLogger("test")),
+		DisableCopy(),
+	)
+	require.NoError(t, err)
+	i, err := f.NewInterceptor("id")
+	require.NoError(t, err)
+	_, ok := i.(*ResponderInterceptor).packetFactory.(*noOpPacketFactory)
+	require.True(t, ok)
 }
 
 // this test is only useful when being run with the race detector, it won't fail otherwise:
@@ -83,10 +119,10 @@ func TestResponderInterceptor_Race(t *testing.T) {
 		ResponderSize(32768),
 		ResponderLog(logging.NewDefaultLoggerFactory().NewLogger("test")),
 	)
-	assert.NoError(t, err)
+	require.NoError(t, err)
 
 	i, err := f.NewInterceptor("")
-	assert.NoError(t, err)
+	require.NoError(t, err)
 
 	stream := test.NewMockStream(&interceptor.StreamInfo{
 		SSRC:         1,
@@ -94,7 +130,7 @@ func TestResponderInterceptor_Race(t *testing.T) {
 	}, i)
 
 	for seqNum := uint16(0); seqNum < 500; seqNum++ {
-		assert.NoError(t, stream.WriteRTP(&rtp.Packet{Header: rtp.Header{SequenceNumber: seqNum}}))
+		require.NoError(t, stream.WriteRTP(&rtp.Packet{Header: rtp.Header{SequenceNumber: seqNum}}))
 
 		// 25% packet loss
 		if seqNum%4 == 0 {