Various cleanup and fixes

association_stats.go

@@ -8,13 +8,32 @@ import (
 )
 
 type associationStats struct {
+	nPackets     uint64
+	nPacketsSent uint64
 	nDATAs       uint64
 	nSACKs       uint64
+	nSACKsSent   uint64
 	nT3Timeouts  uint64
 	nAckTimeouts uint64
 	nFastRetrans uint64
 }
 
+func (s *associationStats) incPackets() {
+	atomic.AddUint64(&s.nPackets, 1)
+}
+
+func (s *associationStats) getNumPackets() uint64 {
+	return atomic.LoadUint64(&s.nPackets)
+}
+
+func (s *associationStats) incPacketsSent() {
+	atomic.AddUint64(&s.nPacketsSent, 1)
+}
+
+func (s *associationStats) getNumPacketsSent() uint64 {
+	return atomic.LoadUint64(&s.nPacketsSent)
+}
+
 func (s *associationStats) incDATAs() {
 	atomic.AddUint64(&s.nDATAs, 1)
 }
@@ -31,6 +50,14 @@ func (s *associationStats) getNumSACKs() uint64 {
 	return atomic.LoadUint64(&s.nSACKs)
 }
 
+func (s *associationStats) incSACKsSent() {
+	atomic.AddUint64(&s.nSACKsSent, 1)
+}
+
+func (s *associationStats) getNumSACKsSent() uint64 {
+	return atomic.LoadUint64(&s.nSACKsSent)
+}
+
 func (s *associationStats) incT3Timeouts() {
 	atomic.AddUint64(&s.nT3Timeouts, 1)
 }
@@ -56,8 +83,11 @@ func (s *associationStats) getNumFastRetrans() uint64 {
 }
 
 func (s *associationStats) reset() {
+	atomic.StoreUint64(&s.nPackets, 0)
+	atomic.StoreUint64(&s.nPacketsSent, 0)
 	atomic.StoreUint64(&s.nDATAs, 0)
 	atomic.StoreUint64(&s.nSACKs, 0)
+	atomic.StoreUint64(&s.nSACKsSent, 0)
 	atomic.StoreUint64(&s.nT3Timeouts, 0)
 	atomic.StoreUint64(&s.nAckTimeouts, 0)
 	atomic.StoreUint64(&s.nFastRetrans, 0)

association_test.go

@@ -257,14 +257,19 @@ func createNewAssociationPair(br *test.Bridge, ackMode int, recvBufSize uint32)
 
 	go func() {
 		a0, err0 = Client(Config{
+			Name:                 "a0",
 			NetConn:              br.GetConn0(),
 			MaxReceiveBufferSize: recvBufSize,
 			LoggerFactory:        loggerFactory,
 		})
 		handshake0Ch <- true
 	}()
 	go func() {
-		a1, err1 = Client(Config{
+		// we could have two "client"s here but it's more
+		// standard to have one peer starting initialization and
+		// another waiting for the initialization to be requested (INIT).
+		a1, err1 = Server(Config{
+			Name:                 "a1",
 			NetConn:              br.GetConn1(),
 			MaxReceiveBufferSize: recvBufSize,
 			LoggerFactory:        loggerFactory,
@@ -1752,7 +1757,7 @@ func TestAssocT3RtxTimer(t *testing.T) {
 }
 
 func TestAssocCongestionControl(t *testing.T) {
-	// sbuf - large enobh not to be bundled
+	// sbuf - large enough not to be bundled
 	sbuf := make([]byte, 1000)
 	for i := 0; i < len(sbuf); i++ {
 		sbuf[i] = byte(i & 0xcc)

chunk_payload_data.go

@@ -206,3 +206,7 @@ func (p *chunkPayloadData) setAllInflight() {
 		}
 	}
 }
+
+func (p *chunkPayloadData) isFragmented() bool {
+	return !(p.head == nil && p.beginningFragment && p.endingFragment)
+}

control_queue.go

@@ -3,9 +3,14 @@
 
 package sctp
 
+import (
+	"sync"
+)
+
 // control queue
 
 type controlQueue struct {
+	mu    sync.RWMutex
 	queue []*packet
 }
 
@@ -14,19 +19,28 @@ func newControlQueue() *controlQueue {
 }
 
 func (q *controlQueue) push(c *packet) {
+	q.mu.Lock()
 	q.queue = append(q.queue, c)
+	q.mu.Unlock()
 }
 
 func (q *controlQueue) pushAll(packets []*packet) {
+	q.mu.Lock()
 	q.queue = append(q.queue, packets...)
+	q.mu.Unlock()
 }
 
 func (q *controlQueue) popAll() []*packet {
+	q.mu.Lock()
 	packets := q.queue
 	q.queue = []*packet{}
+	q.mu.Unlock()
 	return packets
 }
 
 func (q *controlQueue) size() int {
-	return len(q.queue)
+	q.mu.RLock()
+	size := len(q.queue)
+	q.mu.RUnlock()
+	return size
 }

packet_test.go

@@ -20,10 +20,10 @@ func TestPacketUnmarshal(t *testing.T) {
 	switch {
 	case err != nil:
 		t.Errorf("Unmarshal failed for SCTP packet with no chunks: %v", err)
-	case pkt.sourcePort != 5000:
-		t.Errorf("Unmarshal passed for SCTP packet, but got incorrect source port exp: %d act: %d", 5000, pkt.sourcePort)
-	case pkt.destinationPort != 5000:
-		t.Errorf("Unmarshal passed for SCTP packet, but got incorrect destination port exp: %d act: %d", 5000, pkt.destinationPort)
+	case pkt.sourcePort != defaultSCTPSrcDstPort:
+		t.Errorf("Unmarshal passed for SCTP packet, but got incorrect source port exp: %d act: %d", defaultSCTPSrcDstPort, pkt.sourcePort)
+	case pkt.destinationPort != defaultSCTPSrcDstPort:
+		t.Errorf("Unmarshal passed for SCTP packet, but got incorrect destination port exp: %d act: %d", defaultSCTPSrcDstPort, pkt.destinationPort)
 	case pkt.verificationTag != 0:
 		t.Errorf("Unmarshal passed for SCTP packet, but got incorrect verification tag exp: %d act: %d", 0, pkt.verificationTag)
 	}

payload_queue.go

@@ -6,9 +6,11 @@ package sctp
 import (
 	"fmt"
 	"sort"
+	"sync"
 )
 
 type payloadQueue struct {
+	mu       sync.RWMutex
 	chunkMap map[uint32]*chunkPayloadData
 	sorted   []uint32
 	dupTSN   []uint32
@@ -19,7 +21,15 @@ func newPayloadQueue() *payloadQueue {
 	return &payloadQueue{chunkMap: map[uint32]*chunkPayloadData{}}
 }
 
+//nolint:unused
 func (q *payloadQueue) updateSortedKeys() {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	q.updateSortedKeysWithLock()
+}
+
+func (q *payloadQueue) updateSortedKeysWithLock() {
 	if q.sorted != nil {
 		return
 	}
@@ -37,6 +47,9 @@ func (q *payloadQueue) updateSortedKeys() {
 }
 
 func (q *payloadQueue) canPush(p *chunkPayloadData, cumulativeTSN uint32) bool {
+	q.mu.RLock()
+	defer q.mu.RUnlock()
+
 	_, ok := q.chunkMap[p.tsn]
 	if ok || sna32LTE(p.tsn, cumulativeTSN) {
 		return false
@@ -45,6 +58,9 @@ func (q *payloadQueue) canPush(p *chunkPayloadData, cumulativeTSN uint32) bool {
 }
 
 func (q *payloadQueue) pushNoCheck(p *chunkPayloadData) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
 	q.chunkMap[p.tsn] = p
 	q.nBytes += len(p.userData)
 	q.sorted = nil
@@ -54,6 +70,9 @@ func (q *payloadQueue) pushNoCheck(p *chunkPayloadData) {
 // older than our cumulativeTSN marker, it will be recored as duplications,
 // which can later be retrieved using popDuplicates.
 func (q *payloadQueue) push(p *chunkPayloadData, cumulativeTSN uint32) bool {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
 	_, ok := q.chunkMap[p.tsn]
 	if ok || sna32LTE(p.tsn, cumulativeTSN) {
 		// Found the packet, log in dups
@@ -69,7 +88,10 @@ func (q *payloadQueue) push(p *chunkPayloadData, cumulativeTSN uint32) bool {
 
 // pop pops only if the oldest chunk's TSN matches the given TSN.
 func (q *payloadQueue) pop(tsn uint32) (*chunkPayloadData, bool) {
-	q.updateSortedKeys()
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	q.updateSortedKeysWithLock()
 
 	if len(q.chunkMap) > 0 && tsn == q.sorted[0] {
 		q.sorted = q.sorted[1:]
@@ -85,25 +107,34 @@ func (q *payloadQueue) pop(tsn uint32) (*chunkPayloadData, bool) {
 
 // get returns reference to chunkPayloadData with the given TSN value.
 func (q *payloadQueue) get(tsn uint32) (*chunkPayloadData, bool) {
+	q.mu.RLock()
+	defer q.mu.RUnlock()
+
 	c, ok := q.chunkMap[tsn]
 	return c, ok
 }
 
 // popDuplicates returns an array of TSN values that were found duplicate.
 func (q *payloadQueue) popDuplicates() []uint32 {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
 	dups := q.dupTSN
 	q.dupTSN = []uint32{}
 	return dups
 }
 
 func (q *payloadQueue) getGapAckBlocks(cumulativeTSN uint32) (gapAckBlocks []gapAckBlock) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
 	var b gapAckBlock
 
 	if len(q.chunkMap) == 0 {
 		return []gapAckBlock{}
 	}
 
-	q.updateSortedKeys()
+	q.updateSortedKeysWithLock()
 
 	for i, tsn := range q.sorted {
 		if i == 0 {
@@ -155,7 +186,10 @@ func (q *payloadQueue) markAsAcked(tsn uint32) int {
 }
 
 func (q *payloadQueue) getLastTSNReceived() (uint32, bool) {
-	q.updateSortedKeys()
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	q.updateSortedKeysWithLock()
 
 	qlen := len(q.sorted)
 	if qlen == 0 {
@@ -165,6 +199,9 @@ func (q *payloadQueue) getLastTSNReceived() (uint32, bool) {
 }
 
 func (q *payloadQueue) markAllToRetrasmit() {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
 	for _, c := range q.chunkMap {
 		if c.acked || c.abandoned() {
 			continue
@@ -174,9 +211,15 @@ func (q *payloadQueue) markAllToRetrasmit() {
 }
 
 func (q *payloadQueue) getNumBytes() int {
+	q.mu.RLock()
+	defer q.mu.RUnlock()
+
 	return q.nBytes
 }
 
 func (q *payloadQueue) size() int {
+	q.mu.RLock()
+	defer q.mu.RUnlock()
+
 	return len(q.chunkMap)
 }

pending_queue.go

@@ -5,11 +5,14 @@ package sctp
 
 import (
 	"errors"
+	"sync"
+	"sync/atomic"
 )
 
 // pendingBaseQueue
 
 type pendingBaseQueue struct {
+	mu    sync.RWMutex
 	queue []*chunkPayloadData
 }
 
@@ -18,10 +21,14 @@ func newPendingBaseQueue() *pendingBaseQueue {
 }
 
 func (q *pendingBaseQueue) push(c *chunkPayloadData) {
+	q.mu.Lock()
 	q.queue = append(q.queue, c)
+	q.mu.Unlock()
 }
 
 func (q *pendingBaseQueue) pop() *chunkPayloadData {
+	q.mu.Lock()
+	defer q.mu.Unlock()
 	if len(q.queue) == 0 {
 		return nil
 	}
@@ -31,13 +38,17 @@ func (q *pendingBaseQueue) pop() *chunkPayloadData {
 }
 
 func (q *pendingBaseQueue) get(i int) *chunkPayloadData {
+	q.mu.RLock()
+	defer q.mu.RUnlock()
 	if len(q.queue) == 0 || i < 0 || i >= len(q.queue) {
 		return nil
 	}
 	return q.queue[i]
 }
 
 func (q *pendingBaseQueue) size() int {
+	q.mu.RLock()
+	defer q.mu.RUnlock()
 	return len(q.queue)
 }
 
@@ -46,7 +57,7 @@ func (q *pendingBaseQueue) size() int {
 type pendingQueue struct {
 	unorderedQueue      *pendingBaseQueue
 	orderedQueue        *pendingBaseQueue
-	nBytes              int
+	nBytes              uint64
 	selected            bool
 	unorderedIsSelected bool
 }
@@ -71,7 +82,7 @@ func (q *pendingQueue) push(c *chunkPayloadData) {
 	} else {
 		q.orderedQueue.push(c)
 	}
-	q.nBytes += len(c.userData)
+	atomic.AddUint64(&q.nBytes, uint64(len(c.userData)))
 }
 
 func (q *pendingQueue) peek() *chunkPayloadData {
@@ -129,12 +140,12 @@ func (q *pendingQueue) pop(c *chunkPayloadData) error {
 			}
 		}
 	}
-	q.nBytes -= len(c.userData)
+	atomic.AddUint64(&q.nBytes, -uint64(len(c.userData)))
 	return nil
 }
 
 func (q *pendingQueue) getNumBytes() int {
-	return q.nBytes
+	return int(atomic.LoadUint64(&q.nBytes))
 }
 
 func (q *pendingQueue) size() int {