func (a *Association) gatherOutboundFrastRetransmissionPackets(rawPackets [][]byte) [][]byte {
	if a.willRetransmitFast {
		a.willRetransmitFast = false
		toFastRetrans := []chunk{}
		fastRetransSize := commonHeaderSize
		for i := 0; ; i++ {
			c, ok := a.inflightQueue.get(a.cumulativeTSNAckPoint + uint32(i) + 1)
			if !ok {
				break  
			}
			if c.acked || c.abandoned {
				continue
			}
			if c.nSent > 1 || c.missIndicator < 3 {
				continue
			}
			dataChunkSize := dataChunkHeaderSize + uint32(len(c.userData))
			if a.mtu < fastRetransSize+dataChunkSize {
				break
			}
			fastRetransSize += dataChunkSize
			a.stats.incFastRetrans()
			c.nSent++
			a.checkPartialReliabilityStatus(c)
			toFastRetrans = append(toFastRetrans, c)
			a.log.Tracef("[%s] fast-retransmit: tsn=%d sent=%d htna=%d",
				a.name, c.tsn, c.nSent, a.fastRecoverExitPoint)
		}
		if len(toFastRetrans) > 0 {
			raw, err := a.createPacket(toFastRetrans).marshal()
			if err != nil {
				a.log.Warnf("[%s] failed to serialize a DATA packet to be fast-retransmitted", a.name)
			} else {
				rawPackets = append(rawPackets, raw)
			}
		}
	}
	return rawPackets
}
The caller should hold the lock
package sctp
import (
	"bytes"
	"fmt"
	"io"
	"math"
	"math/rand"
	"net"
	"sync"
	"sync/atomic"
	"time"
	"github.com/pion/logging"
	"github.com/pkg/errors"
)
const (
	receiveMTU          uint32 = 8192  
	initialMTU          uint32 = 1228  
	initialRecvBufSize  uint32 = 1024 * 1024
	commonHeaderSize    uint32 = 12
	dataChunkHeaderSize uint32 = 16
)
const (
	closed uint32 = iota
	cookieWait
	cookieEchoed
	established
	shutdownAckSent
	shutdownPending
	shutdownReceived
	shutdownSent
)
const (
	timerT1Init int = iota
	timerT1Cookie
	timerT3RTX
	timerReconfig
)
const (
	ackModeNormal int = iota
	ackModeNoDelay
	ackModeAlwaysDelay
)
const (
	ackStateIdle      int = iota  
	ackStateImmediate             
	ackStateDelay                 
)
const (
	acceptChSize = 16
)
func getAssociationStateString(a uint32) string {
	switch a {
	case closed:
		return "Closed"
	case cookieWait:
		return "CookieWait"
	case cookieEchoed:
		return "CookieEchoed"
	case established:
		return "Established"
	case shutdownPending:
		return "ShutdownPending"
	case shutdownSent:
		return "ShutdownSent"
	case shutdownReceived:
		return "ShutdownReceived"
	case shutdownAckSent:
		return "ShutdownAckSent"
	default:
		return fmt.Sprintf("Invalid association state %d", a)
	}
}
type Association struct {
	bytesReceived uint64
	bytesSent     uint64
	lock sync.RWMutex
	netConn net.Conn
	peerVerificationTag    uint32
	myVerificationTag      uint32
	state                  uint32
	myNextTSN              uint32  
	peerLastTSN            uint32  
	minTSN2MeasureRTT      uint32  
	willSendForwardTSN     bool
	willRetransmitFast     bool
	willRetransmitReconfig bool
	myNextRSN        uint32
	reconfigs        map[uint32]*chunkReconfig
	reconfigRequests map[uint32]*paramOutgoingResetRequest
	sourcePort              uint16
	destinationPort         uint16
	myMaxNumInboundStreams  uint16
	myMaxNumOutboundStreams uint16
	myCookie                *paramStateCookie
	payloadQueue            *payloadQueue
	inflightQueue           *payloadQueue
	pendingQueue            *pendingQueue
	controlQueue            *controlQueue
	mtu                     uint32
	maxPayloadSize          uint32  
	cumulativeTSNAckPoint   uint32
	advancedPeerTSNAckPoint uint32
	useForwardTSN           bool
	maxReceiveBufferSize uint32
	cwnd                 uint32  
	rwnd                 uint32  
	ssthresh             uint32  
	partialBytesAcked    uint32
	inFastRecovery       bool
	fastRecoverExitPoint uint32
	rtoMgr    *rtoManager
	t1Init    *rtxTimer
	t1Cookie  *rtxTimer
	t3RTX     *rtxTimer
	tReconfig *rtxTimer
	ackTimer  *ackTimer
	storedInit       *chunkInit
	storedCookieEcho *chunkCookieEcho
	streams              map[uint16]*Stream
	acceptCh             chan *Stream
	readLoopCloseCh      chan struct{}
	awakeWriteLoopCh     chan struct{}
	closeWriteLoopCh     chan struct{}
	handshakeCompletedCh chan error
	closeWriteLoopOnce sync.Once
	silentError error
	ackState int
	ackMode  int  
	stats *associationStats
	delayedAckTriggered   bool
	immediateAckTriggered bool
	name string
	log  logging.LeveledLogger
}
type Config struct {
	NetConn              net.Conn
	MaxReceiveBufferSize uint32
	LoggerFactory        logging.LoggerFactory
}
func Server(config Config) (*Association, error) {
	a := createAssociation(config)
	a.init(false)
	select {
	case err := <-a.handshakeCompletedCh:
		if err != nil {
			return nil, err
		}
		return a, nil
	case <-a.readLoopCloseCh:
		return nil, errors.Errorf("association closed before connecting")
	}
}
func Client(config Config) (*Association, error) {
	a := createAssociation(config)
	a.init(true)
	select {
	case err := <-a.handshakeCompletedCh:
		if err != nil {
			return nil, err
		}
		return a, nil
	case <-a.readLoopCloseCh:
		return nil, errors.Errorf("association closed before connecting")
	}
}
func createAssociation(config Config) *Association {
	rs := rand.NewSource(time.Now().UnixNano())
	r := rand.New(rs)
	var maxReceiveBufferSize uint32
	if config.MaxReceiveBufferSize == 0 {
		maxReceiveBufferSize = initialRecvBufSize
	} else {
		maxReceiveBufferSize = config.MaxReceiveBufferSize
	}
	tsn := r.Uint32()
	a := &Association{
		netConn:                 config.NetConn,
		maxReceiveBufferSize:    maxReceiveBufferSize,
		myMaxNumOutboundStreams: math.MaxUint16,
		myMaxNumInboundStreams:  math.MaxUint16,
		payloadQueue:            newPayloadQueue(),
		inflightQueue:           newPayloadQueue(),
		pendingQueue:            newPendingQueue(),
		controlQueue:            newControlQueue(),
		mtu:                     initialMTU,
		maxPayloadSize:          initialMTU - (commonHeaderSize + dataChunkHeaderSize),
		myVerificationTag:       r.Uint32(),
		myNextTSN:               tsn,
		myNextRSN:               tsn,
		minTSN2MeasureRTT:       tsn,
		state:                   closed,
		rtoMgr:                  newRTOManager(),
		streams:                 map[uint16]*Stream{},
		reconfigs:               map[uint32]*chunkReconfig{},
		reconfigRequests:        map[uint32]*paramOutgoingResetRequest{},
		acceptCh:                make(chan *Stream, acceptChSize),
		readLoopCloseCh:         make(chan struct{}),
		awakeWriteLoopCh:        make(chan struct{}, 1),
		closeWriteLoopCh:        make(chan struct{}),
		handshakeCompletedCh:    make(chan error),
		cumulativeTSNAckPoint:   tsn - 1,
		advancedPeerTSNAckPoint: tsn - 1,
		silentError:             errors.Errorf("silently discard"),
		stats:                   &associationStats{},
		log:                     config.LoggerFactory.NewLogger("sctp"),
	}
	a.name = fmt.Sprintf("%p", a)
	a.cwnd = min32(4*a.mtu, max32(2*a.mtu, 4380))
	a.log.Tracef("[%s] updated cwnd=%d ssthresh=%d inflight=%d (INI)",
		a.name, a.cwnd, a.ssthresh, a.inflightQueue.getNumBytes())
	a.t1Init = newRTXTimer(timerT1Init, a, maxInitRetrans)
	a.t1Cookie = newRTXTimer(timerT1Cookie, a, maxInitRetrans)
	a.t3RTX = newRTXTimer(timerT3RTX, a, noMaxRetrans)         
	a.tReconfig = newRTXTimer(timerReconfig, a, noMaxRetrans)  
	a.ackTimer = newAckTimer(a)
	return a
}
func (a *Association) init(isClient bool) {
	a.lock.Lock()
	defer a.lock.Unlock()
	go a.readLoop()
	go a.writeLoop()
	if isClient {
		a.setState(cookieWait)
		init := &chunkInit{}
		init.initialTSN = a.myNextTSN
		init.numOutboundStreams = a.myMaxNumOutboundStreams
		init.numInboundStreams = a.myMaxNumInboundStreams
		init.initiateTag = a.myVerificationTag
		init.advertisedReceiverWindowCredit = a.maxReceiveBufferSize
		setSupportedExtensions(&init.chunkInitCommon)
		a.storedInit = init
		err := a.sendInit()
		if err != nil {
			a.log.Errorf("[%s] failed to send init: %s", a.name, err.Error())
		}
		a.t1Init.start(a.rtoMgr.getRTO())
	}
}
func (a *Association) sendInit() error {
	a.log.Debugf("[%s] sending INIT", a.name)
	if a.storedInit == nil {
		return errors.Errorf("the init not stored to send")
	}
	outbound := &packet{}
	outbound.verificationTag = a.peerVerificationTag
	a.sourcePort = 5000       
	a.destinationPort = 5000  
	outbound.sourcePort = a.sourcePort
	outbound.destinationPort = a.destinationPort
	outbound.chunks = []chunk{a.storedInit}
	a.controlQueue.push(outbound)
	a.awakeWriteLoop()
	return nil
}
func (a *Association) sendCookieEcho() error {
	if a.storedCookieEcho == nil {
		return errors.Errorf("cookieEcho not stored to send")
	}
	a.log.Debugf("[%s] sending COOKIE-ECHO", a.name)
	outbound := &packet{}
	outbound.verificationTag = a.peerVerificationTag
	outbound.sourcePort = a.sourcePort
	outbound.destinationPort = a.destinationPort
	outbound.chunks = []chunk{a.storedCookieEcho}
	a.controlQueue.push(outbound)
	a.awakeWriteLoop()
	return nil
}
func (a *Association) Close() error {
	a.log.Debugf("[%s] closing association..", a.name)
	a.setState(closed)
	err := a.netConn.Close()
	a.closeAllTimers()
	a.closeWriteLoopOnce.Do(func() { close(a.closeWriteLoopCh) })
	<-a.readLoopCloseCh
	a.log.Debugf("[%s] association closed", a.name)
	a.log.Debugf("[%s] stats nDATAs (in) : %d", a.name, a.stats.getNumDATAs())
	a.log.Debugf("[%s] stats nSACKs (in) : %d", a.name, a.stats.getNumSACKs())
	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 err
}
func (a *Association) closeAllTimers() {
	a.t1Init.close()
	a.t1Cookie.close()
	a.t3RTX.close()
	a.tReconfig.close()
	a.ackTimer.close()
}
func (a *Association) readLoop() {
	var closeErr error
	defer func() {
		a.closeWriteLoopOnce.Do(func() { close(a.closeWriteLoopCh) })
		a.lock.Lock()
		for _, s := range a.streams {
			a.unregisterStream(s, closeErr)
		}
		a.lock.Unlock()
		close(a.acceptCh)
		close(a.readLoopCloseCh)
	}()
	a.log.Debugf("[%s] readLoop entered", a.name)
	for {
		buffer := make([]byte, receiveMTU)
		n, err := a.netConn.Read(buffer)
		if err != nil {
			closeErr = err
			break
		}
		atomic.AddUint64(&a.bytesReceived, uint64(n))
		if err = a.handleInbound(buffer[:n]); err != nil {
			a.log.Warnf("[%s] %s]", a.name, errors.Wrap(err, "failed to push SCTP packet").Error())
		}
	}
	a.log.Debugf("[%s] readLoop exited", a.name)
}
func (a *Association) writeLoop() {
	a.log.Debugf("[%s] writeLoop entered", a.name)
loop:
	for {
		rawPackets := a.gatherOutbound()
		for _, raw := range rawPackets {
			_, err := a.netConn.Write(raw)
			if err != nil {
				if err != io.EOF {
					a.log.Warnf("[%s] failed to write packets on netConn: %v", a.name, err)
				}
				a.log.Debugf("[%s] writeLoop ended", a.name)
				break loop
			}
			atomic.AddUint64(&a.bytesSent, uint64(len(raw)))
		}
		select {
		case <-a.awakeWriteLoopCh:
		case <-a.closeWriteLoopCh:
			break loop
		}
	}
	a.setState(closed)
	a.closeAllTimers()
	a.log.Debugf("[%s] writeLoop exited", a.name)
}
func (a *Association) awakeWriteLoop() {
	select {
	case a.awakeWriteLoopCh <- struct{}{}:
	default:
	}
}
func (a *Association) unregisterStream(s *Stream, err error) {
	s.lock.Lock()
	defer s.lock.Unlock()
	delete(a.streams, s.streamIdentifier)
	s.readErr = err
	s.readNotifier.Broadcast()
}
func (a *Association) handleInbound(raw []byte) error {
	p := &packet{}
	if err := p.unmarshal(raw); err != nil {
		return errors.Wrap(err, "unable to parse SCTP packet")
	}
	if err := checkPacket(p); err != nil {
		return errors.Wrap(err, "failed validating packet")
	}
	a.handleChunkStart(p)
	for _, c := range p.chunks {
		err := a.handleChunk(p, c)
		if err != nil {
			a.log.Warnf("[%s] %s", a.name, errors.Wrap(err, "failed handling chunk").Error())
		}
	}
	a.handleChunkEnd(p)
	return nil
}
func (a *Association) gatherOutboundDataAndReconfigPackets(rawPackets [][]byte) [][]byte {
	for _, p := range a.getDataPacketsToRetransmit() {
		raw, err := p.marshal()
		if err != nil {
			a.log.Warnf("[%s] failed to serialize a DATA packet to be retransmitted", a.name)
			continue
		}
		rawPackets = append(rawPackets, raw)
	}
	chunks, sisToReset := a.popPendingDataChunksToSend()
	if len(chunks) > 0 {
		a.log.Tracef("[%s] T3-rtx timer start (pt1)", a.name)
		a.t3RTX.start(a.rtoMgr.getRTO())
		for _, p := range a.bundleDataChunksIntoPackets(chunks) {
			raw, err := p.marshal()
			if err != nil {
				a.log.Warnf("[%s] failed to serialize a DATA packet", a.name)
				continue
			}
			rawPackets = append(rawPackets, raw)
		}
	}
	if len(sisToReset) > 0 || a.willRetransmitReconfig {
		if a.willRetransmitReconfig {
			a.willRetransmitReconfig = false
			a.log.Debugf("[%s] retransmit %d RECONFIG chunk(s)", a.name, len(a.reconfigs))
			for _, c := range a.reconfigs {
				p := a.createPacket([]chunk{c})
				raw, err := p.marshal()
				if err != nil {
					a.log.Warnf("[%s] failed to serialize a RECONFIG packet to be retransmitted", a.name)
				} else {
					rawPackets = append(rawPackets, raw)
				}
			}
		}
		if len(sisToReset) > 0 {
			rsn := a.generateNextRSN()
			tsn := a.myNextTSN - 1
			c := &chunkReconfig{
				paramA: &paramOutgoingResetRequest{
					reconfigRequestSequenceNumber: rsn,
					senderLastTSN:                 tsn,
					streamIdentifiers:             sisToReset,
				},
			}
			a.reconfigs[rsn] = c  
			a.log.Debugf("[%s] sending RECONFIG: rsn=%d tsn=%d streams=%v",
				a.name, rsn, a.myNextTSN-1, sisToReset)
			p := a.createPacket([]chunk{c})
			raw, err := p.marshal()
			if err != nil {
				a.log.Warnf("[%s] failed to serialize a RECONFIG packet to be transmitted", a.name)
			} else {
				rawPackets = append(rawPackets, raw)
			}
		}
		if len(a.reconfigs) > 0 {
			a.tReconfig.start(a.rtoMgr.getRTO())
		}
	}
	return rawPackets
}