Repository: pion/sctp
Pull request #79 (2019-10-21 02:44:56)

Improve handling of SACK
Relates to #62

SCTP implementation was not following the suggestion mentioned in RFC 4960 for generating SACK, causing the poor performance in receiving data from others (Browsers, etc)

By generating SACK at every second packet, as suggested by the RFC, improved the performance to close to the theoretical throughput.

See #62 for some benchmarks. 

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

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

```diff
@@            Coverage Diff             @@
##           master      #79      +/-   ##
==========================================
+ Coverage   76.35%   76.77%   +0.41%     
==========================================
  Files          43       43              
  Lines        2664     2682      +18     
==========================================
+ Hits         2034     2059      +25     
+ Misses        508      501       -7     
  Partials      122      122
```


| [Impacted Files](https://codecov.io/gh/pion/sctp/pull/79?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [ack\_timer.go](https://codecov.io/gh/pion/sctp/pull/79/diff?src=pr&el=tree#diff-YWNrX3RpbWVyLmdv) | `100% <ø> (ø)` | :arrow_up: |
| [chunk\_selective\_ack.go](https://codecov.io/gh/pion/sctp/pull/79/diff?src=pr&el=tree#diff-Y2h1bmtfc2VsZWN0aXZlX2Fjay5nbw==) | `71.15% <100%> (+16.05%)` | :arrow_up: |
| [association.go](https://codecov.io/gh/pion/sctp/pull/79/diff?src=pr&el=tree#diff-YXNzb2NpYXRpb24uZ28=) | `82.82% <95.83%> (+0.44%)` | :arrow_up: |
| [chunk\_payload\_data.go](https://codecov.io/gh/pion/sctp/pull/79/diff?src=pr&el=tree#diff-Y2h1bmtfcGF5bG9hZF9kYXRhLmdv) | `60.78% <0%> (-3.93%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/pion/sctp/pull/79?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/79?src=pr&el=footer). Last update [d1633f9...e121f66](https://codecov.io/gh/pion/sctp/pull/79?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

Sean-Der: @enobufs this is so amazing, I can't wait to hear the results back from @chrisprobst :)
Sean-Der: If you tag it will also create the bump PRs for you! So you don't have to wait for reviews!

Patch:
diff --git a/ack_timer.go b/ack_timer.go
index db932ed..ba23d54 100644
--- a/ack_timer.go
+++ b/ack_timer.go
@@ -43,7 +43,7 @@ func (t *ackTimer) start() bool {
 		return false
 	}
 
-	// this is a noop if the timer is always running
+	// this is a noop if the timer is already running
 	if t.stopFunc != nil {
 		return false
 	}
diff --git a/association.go b/association.go
index 6a29374..6241292 100644
--- a/association.go
+++ b/association.go
@@ -51,9 +51,9 @@ const (
 
 // ack transmission state
 const (
-	ackStateIdle int = iota
-	ackStateImmediate
-	ackStateDelay
+	ackStateIdle      int = iota // ack timer is off
+	ackStateImmediate            // ack timer is on (ack is being delayed)
+	ackStateDelay                // will send ack immediately
 )
 
 // other constants
@@ -177,6 +177,10 @@ type Association struct {
 	// stats
 	stats *associationStats
 
+	// per inbound packet context
+	delayedAckTriggered   bool
+	immediateAckTriggered bool
+
 	name string
 	log  logging.LeveledLogger
 }
@@ -485,13 +489,17 @@ func (a *Association) handleInbound(raw []byte) error {
 		return errors.Wrap(err, "failed validating packet")
 	}
 
+	a.handleChunkStart(p)
+
 	for _, c := range p.chunks {
 		err := a.handleChunk(p, c)
 		if err != nil {
-			return errors.Wrap(err, "failed handling chunk")
+			a.log.Warnf("[%s] %s", a.name, errors.Wrap(err, "failed handling chunk").Error())
 		}
 	}
 
+	a.handleChunkEnd(p)
+
 	return nil
 }
 
@@ -622,6 +630,7 @@ func (a *Association) gatherOutbound() [][]byte {
 		if a.ackState == ackStateImmediate {
 			a.ackState = ackStateIdle
 			sack := a.createSelectiveAckChunk()
+			a.log.Debugf("[%s] sending SACK: %s", a.name, sack.String())
 			raw, err := a.createPacket([]chunk{sack}).marshal()
 			if err != nil {
 				a.log.Warnf("[%s] failed to serialize a SACK packet", a.name)
@@ -1006,14 +1015,13 @@ func (a *Association) handleData(d *chunkPayloadData) []*packet {
 	}
 
 	if (a.ackState != ackStateImmediate && !d.immediateSack && !hasPacketLoss && a.ackMode == ackModeNormal) || a.ackMode == ackModeAlwaysDelay {
-		// Will send delayed ack in the next ack timeout
-		a.ackState = ackStateDelay
-		a.ackTimer.start()
+		if a.ackState == ackStateIdle {
+			a.delayedAckTriggered = true
+		} else {
+			a.immediateAckTriggered = true
+		}
 	} else {
-		// Send SACK now!
-		a.ackState = ackStateImmediate
-		a.ackTimer.stop()
-		a.awakeWriteLoop()
+		a.immediateAckTriggered = true
 	}
 
 	return reply
@@ -1731,24 +1739,6 @@ func (a *Association) popPendingDataChunksToSend() ([]*chunkPayloadData, []uint1
 				chunks = append(chunks, c)
 			}
 		}
-
-		// Controlling DATA chunk's I-bit (SACK-IMMEDIATELY flag)
-		// RFC 7053 sec 4.2.  Triggering at the SCTP Level
-		//    Another case is where the sending of a DATA chunk fills the
-		//    congestion or receiver window.  Setting the I bit in these cases
-		//    improves the throughput of the transfer.
-
-		if len(chunks) > 0 {
-			// IMPLETEMATION NOTE:
-			// It was learned that setting I-bit to the every last chunk that fills
-			// min(cwnd,rnwd) would end up with generating SACK, on the remote, for
-			// every single DATA chunks when sending large amount of data at once.
-			// In order to overcome this situation, here we set I-bit to true for
-			// the last chunk only.
-
-			lastChunk := chunks[len(chunks)-1]
-			lastChunk.immediateSack = true
-		}
 	}
 
 	return chunks, sisToReset
@@ -1903,6 +1893,30 @@ func pack(p *packet) []*packet {
 	return []*packet{p}
 }
 
+func (a *Association) handleChunkStart(p *packet) {
+	a.lock.Lock()
+	defer a.lock.Unlock()
+
+	a.delayedAckTriggered = false
+	a.immediateAckTriggered = false
+}
+
+func (a *Association) handleChunkEnd(p *packet) {
+	a.lock.Lock()
+	defer a.lock.Unlock()
+
+	if a.immediateAckTriggered {
+		// Send SACK now!
+		a.ackState = ackStateImmediate
+		a.ackTimer.stop()
+		a.awakeWriteLoop()
+	} else if a.delayedAckTriggered {
+		// Will send delayed ack in the next ack timeout
+		a.ackState = ackStateDelay
+		a.ackTimer.start()
+	}
+}
+
 func (a *Association) handleChunk(p *packet, c chunk) error {
 	a.lock.Lock()
 	defer a.lock.Unlock()
diff --git a/chunk_selective_ack.go b/chunk_selective_ack.go
index cd93ead..920d562 100644
--- a/chunk_selective_ack.go
+++ b/chunk_selective_ack.go
@@ -129,10 +129,13 @@ func (s *chunkSelectiveAck) check() (abort bool, err error) {
 
 // String makes chunkSelectiveAck printable
 func (s *chunkSelectiveAck) String() string {
-	res := fmt.Sprintf("%s\n%d", s.chunkHeader, s.cumulativeTSNAck)
+	res := fmt.Sprintf("SACK cumTsnAck=%d arwnd=%d dupTsn=%d",
+		s.cumulativeTSNAck,
+		s.advertisedReceiverWindowCredit,
+		s.duplicateTSN)
 
 	for _, gap := range s.gapAckBlocks {
-		res = fmt.Sprintf("\n gap ack: %s", gap)
+		res = fmt.Sprintf("%s\n gap ack: %s", res, gap)
 	}
 
 	return res


Test patch:
diff --git a/association_test.go b/association_test.go
index 0bab318..646c4aa 100644
--- a/association_test.go
+++ b/association_test.go
@@ -1786,7 +1786,7 @@ func TestAssocCongestionControl(t *testing.T) {
 		t.Logf("nT3Timeouts : %d\n", a0.stats.getNumT3Timeouts())
 
 		assert.Equal(t, uint64(nPacketsToSend), a1.stats.getNumDATAs(), "packet count mismatch")
-		assert.True(t, a0.stats.getNumSACKs() < nPacketsToSend/10, "too many sacks")
+		assert.True(t, a0.stats.getNumSACKs() <= nPacketsToSend/2, "too many sacks")
 		assert.Equal(t, uint64(0), a0.stats.getNumT3Timeouts(), "should be no retransmit")
 
 		closeAssociationPair(br, a0, a1)
@@ -1882,7 +1882,86 @@ func TestAssocCongestionControl(t *testing.T) {
 }
 
 func TestAssocDelayedAck(t *testing.T) {
-	t.Run("Ack all DATA chunks with one SACK", func(t *testing.T) {
+	t.Run("First DATA chunk gets acked with delay", func(t *testing.T) {
+		const si uint16 = 6
+		var n int
+		var nPacketsReceived int
+		var ppi PayloadProtocolIdentifier
+		sbuf := make([]byte, 1000) // size should be less than initial cwnd (4380)
+		rbuf := make([]byte, 1500)
+
+		_, err := cryptoRand.Read(sbuf)
+		if !assert.Nil(t, err, "failed to create associations") {
+			return
+		}
+
+		br := test.NewBridge()
+
+		a0, a1, err := createNewAssociationPair(br, ackModeAlwaysDelay, 0)
+		if !assert.Nil(t, err, "failed to create associations") {
+			assert.FailNow(t, "failed due to earlier error")
+		}
+
+		s0, s1, err := establishSessionPair(br, a0, a1, si)
+		assert.Nil(t, err, "failed to establish session pair")
+
+		a0.stats.reset()
+		a1.stats.reset()
+
+		// Writes data (will fragmented)
+		n, err = s0.WriteSCTP(sbuf, PayloadTypeWebRTCBinary)
+		assert.Nil(t, err, "WriteSCTP failed")
+		assert.Equal(t, n, len(sbuf), "unexpected length of received data")
+
+		// Repeat calling br.Tick() until the buffered amount becomes 0
+		since := time.Now()
+		for s0.BufferedAmount() > 0 {
+			for {
+				n = br.Tick()
+				if n == 0 {
+					break
+				}
+			}
+
+			for {
+				s1.lock.RLock()
+				readable := s1.reassemblyQueue.isReadable()
+				s1.lock.RUnlock()
+				if !readable {
+					break
+				}
+				n, ppi, err = s1.ReadSCTP(rbuf)
+				if !assert.Nil(t, err, "ReadSCTP failed") {
+					return
+				}
+				assert.Equal(t, len(sbuf), n, "unexpected length of received data")
+				assert.Equal(t, ppi, PayloadTypeWebRTCBinary, "unexpected ppi")
+
+				nPacketsReceived++
+			}
+		}
+		delay := time.Since(since).Seconds()
+		t.Logf("received in %.03f seconds", delay)
+		assert.True(t, delay >= 0.2, "should be >= 200msec")
+
+		br.Process()
+
+		assert.Equal(t, 1, nPacketsReceived, "should be one packet received")
+		assert.Equal(t, 0, s1.getNumBytesInReassemblyQueue(), "reassembly queue should be empty")
+
+		t.Logf("nDATAs      : %d\n", a1.stats.getNumDATAs())
+		t.Logf("nSACKs      : %d\n", a0.stats.getNumSACKs())
+		t.Logf("nAckTimeouts: %d\n", a1.stats.getNumAckTimeouts())
+
+		assert.Equal(t, uint64(1), a1.stats.getNumDATAs(), "DATA chunk count mismatch")
+		assert.Equal(t, a0.stats.getNumSACKs(), a1.stats.getNumDATAs(), "sack count should be equal to the number of data chunks")
+		assert.Equal(t, uint64(1), a1.stats.getNumAckTimeouts(), "ackTimeout count mismatch")
+		assert.Equal(t, uint64(0), a0.stats.getNumT3Timeouts(), "should be no retransmit")
+
+		closeAssociationPair(br, a0, a1)
+	})
+
+	t.Run("Second DATA chunk to generate SACK immedidately", func(t *testing.T) {
 		const si uint16 = 6
 		var n int
 		var nPacketsReceived int
@@ -1950,8 +2029,8 @@ func TestAssocDelayedAck(t *testing.T) {
 		t.Logf("nAckTimeouts: %d\n", a1.stats.getNumAckTimeouts())
 
 		assert.Equal(t, uint64(4), a1.stats.getNumDATAs(), "DATA chunk count mismatch")
-		assert.Equal(t, uint64(1), a0.stats.getNumSACKs(), "sack count should be one")
-		assert.Equal(t, uint64(1), a1.stats.getNumAckTimeouts(), "ackTimeout count mismatch")
+		assert.True(t, a0.stats.getNumSACKs() < a1.stats.getNumDATAs(), "sack count should less than data")
+		assert.Equal(t, uint64(0), a1.stats.getNumAckTimeouts(), "ackTimeout count mismatch")
 		assert.Equal(t, uint64(0), a0.stats.getNumT3Timeouts(), "should be no retransmit")
 
 		closeAssociationPair(br, a0, a1)