association.go

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 // MTU for inbound packet (from DTLS)
	initialMTU          uint32 = 1228 // initial MTU for outgoing packets (to DTLS)
	initialRecvBufSize  uint32 = 1024 * 1024
	commonHeaderSize    uint32 = 12
	dataChunkHeaderSize uint32 = 16
)

// association state enums
const (
	closed uint32 = iota
	cookieWait
	cookieEchoed
	established
	shutdownAckSent
	shutdownPending
	shutdownReceived
	shutdownSent
)

// retransmission timer IDs
const (
	timerT1Init int = iota
	timerT1Cookie
	timerT3RTX
	timerReconfig
)

// ack mode (for testing)
const (
	ackModeNormal int = iota
	ackModeNoDelay
	ackModeAlwaysDelay
)

// ack transmission state
const (
	ackStateIdle      int = iota // ack timer is off
	ackStateImmediate            // ack timer is on (ack is being delayed)
	ackStateDelay                // will send ack immediately
)

// other constants
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)
	}
}

// Association represents an SCTP association
// 13.2.  Parameters Necessary per Association (i.e., the TCB)
// Peer        : Tag value to be sent in every packet and is received
// Verification: in the INIT or INIT ACK chunk.
// Tag         :
//
// My          : Tag expected in every inbound packet and sent in the
// Verification: INIT or INIT ACK chunk.
//
// Tag         :
// State       : A state variable indicating what state the association
//             : is in, i.e., COOKIE-WAIT, COOKIE-ECHOED, ESTABLISHED,
//             : SHUTDOWN-PENDING, SHUTDOWN-SENT, SHUTDOWN-RECEIVED,
//             : SHUTDOWN-ACK-SENT.
//
//               Note: No "CLOSED" state is illustrated since if a
//               association is "CLOSED" its TCB SHOULD be removed.
type Association struct {
	bytesReceived uint64
	bytesSent     uint64

	lock sync.RWMutex

	netConn net.Conn

	peerVerificationTag    uint32
	myVerificationTag      uint32
	state                  uint32
	myNextTSN              uint32 // nextTSN
	peerLastTSN            uint32 // lastRcvdTSN
	minTSN2MeasureRTT      uint32 // for RTT measurement
	willSendForwardTSN     bool
	willRetransmitFast     bool
	willRetransmitReconfig bool

	// Reconfig
	myNextRSN        uint32
	reconfigs        map[uint32]*chunkReconfig
	reconfigRequests map[uint32]*paramOutgoingResetRequest

	// Non-RFC internal data
	sourcePort              uint16
	destinationPort         uint16
	myMaxNumInboundStreams  uint16
	myMaxNumOutboundStreams uint16
	myCookie                *paramStateCookie
	payloadQueue            *payloadQueue
	inflightQueue           *payloadQueue
	pendingQueue            *pendingQueue
	controlQueue            *controlQueue
	mtu                     uint32
	maxPayloadSize          uint32 // max DATA chunk payload size
	cumulativeTSNAckPoint   uint32
	advancedPeerTSNAckPoint uint32
	useForwardTSN           bool

	// Congestion control parameters
	maxReceiveBufferSize uint32
	cwnd                 uint32 // my congestion window size
	rwnd                 uint32 // calculated peer's receiver windows size
	ssthresh             uint32 // slow start threshold
	partialBytesAcked    uint32
	inFastRecovery       bool
	fastRecoverExitPoint uint32

	// RTX & Ack timer
	rtoMgr    *rtoManager
	t1Init    *rtxTimer
	t1Cookie  *rtxTimer
	t3RTX     *rtxTimer
	tReconfig *rtxTimer
	ackTimer  *ackTimer

	// Chunks stored for retransmission
	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

	// local error
	silentError error

	ackState int
	ackMode  int // for testing

	// stats
	stats *associationStats

	// per inbound packet context
	delayedAckTriggered   bool
	immediateAckTriggered bool

	name string
	log  logging.LeveledLogger
}

// Config collects the arguments to createAssociation construction into
// a single structure
type Config struct {
	NetConn              net.Conn
	MaxReceiveBufferSize uint32
	LoggerFactory        logging.LoggerFactory
}

// Server accepts a SCTP stream over a conn
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")
	}
}

// Client opens a SCTP stream over a conn
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)

	// RFC 4690 Sec 7.2.1
	//  o  The initial cwnd before DATA transmission or after a sufficiently
	//     long idle period MUST be set to min(4*MTU, max (2*MTU, 4380
	//     bytes)).
	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)        // retransmit forever
	a.tReconfig = newRTXTimer(timerReconfig, a, noMaxRetrans) // retransmit forever
	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())
	}
}

// caller must hold a.lock
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      // TODO: Spec??
	a.destinationPort = 5000 // TODO: Spec??
	outbound.sourcePort = a.sourcePort
	outbound.destinationPort = a.destinationPort

	outbound.chunks = []chunk{a.storedInit}

	a.controlQueue.push(outbound)
	a.awakeWriteLoop()

	return nil
}

// caller must hold a.lock
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
}

// Close ends the SCTP Association and cleans up any state
func (a *Association) Close() error {
	a.log.Debugf("[%s] closing association..", a.name)

	// TODO: shutdown sequence
	a.setState(closed)

	err := a.netConn.Close()

	a.closeAllTimers()

	// awake writeLoop to exit
	a.closeWriteLoopOnce.Do(func() { close(a.closeWriteLoopCh) })

	// Wait for readLoop to end
	<-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() {
	// Close all retransmission & ack timers
	a.t1Init.close()
	a.t1Cookie.close()
	a.t3RTX.close()
	a.tReconfig.close()
	a.ackTimer.close()
}

func (a *Association) readLoop() {
	var closeErr error
	defer func() {
		// also stop writeLoop, otherwise writeLoop can be leaked
		// if connection is lost when there is no writing packet.
		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 is recreated because the user data is
		// passed to the reassembly queue without copying
		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:
	}
}

// unregisterStream un-registers a stream from the association
// The caller should hold the association write lock.
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()
}

// handleInbound parses incoming raw packets
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
}

// The caller should hold the lock
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)
	}

	// Pop unsent data chunks from the pending queue to send as much as
	// cwnd and rwnd allow.
	chunks, sisToReset := a.popPendingDataChunksToSend()
	if len(chunks) > 0 {
		// Start timer. (noop if already started)
		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 // store in the map for retransmission
			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
}

// The caller should hold the lock
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 // end of pending data
			}

			if c.acked || c.abandoned {
				continue
			}

			if c.nSent > 1 || c.missIndicator < 3 {
				continue
			}

			// RFC 4960 Sec 7.2.4 Fast Retransmit on Gap Reports
			//  3)  Determine how many of the earliest (i.e., lowest TSN) DATA chunks
			//      marked for retransmission will fit into a single packet, subject
			//      to constraint of the path MTU of the destination transport
			//      address to which the packet is being sent.  Call this value K.
			//      Retransmit those K DATA chunks in a single packet.  When a Fast
			//      Retransmit is being performed, the sender SHOULD ignore the value
			//      of cwnd and SHOULD NOT delay retransmission for this single
			//		packet.

			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
func (a *Association) gatherOutboundSackPackets(rawPackets [][]byte) [][]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)
		} else {
			rawPackets = append(rawPackets, raw)
		}
	}

	return rawPackets
}

// The caller should hold the lock
func (a *Association) gatherOutboundForwardTSNPackets(rawPackets [][]byte) [][]byte {
	if a.willSendForwardTSN {
		a.willSendForwardTSN = false
		if sna32GT(a.advancedPeerTSNAckPoint, a.cumulativeTSNAckPoint) {
			fwdtsn := a.createForwardTSN()
			raw, err := a.createPacket([]chunk{fwdtsn}).marshal()
			if err != nil {
				a.log.Warnf("[%s] failed to serialize a Forward TSN packet", a.name)
			} else {
				rawPackets = append(rawPackets, raw)
			}
		}
	}

	return rawPackets
}

// gatherOutbound gathers outgoing packets
func (a *Association) gatherOutbound() [][]byte {
	a.lock.Lock()
	defer a.lock.Unlock()

	rawPackets := [][]byte{}

	if a.controlQueue.size() > 0 {
		for _, p := range a.controlQueue.popAll() {
			raw, err := p.marshal()
			if err != nil {
				a.log.Warnf("[%s] failed to serialize a control packet", a.name)
				continue
			}
			rawPackets = append(rawPackets, raw)
		}
	}

	state := a.getState()

	if state == established {
		rawPackets = a.gatherOutboundDataAndReconfigPackets(rawPackets)
		rawPackets = a.gatherOutboundFrastRetransmissionPackets(rawPackets)
		rawPackets = a.gatherOutboundSackPackets(rawPackets)
		rawPackets = a.gatherOutboundForwardTSNPackets(rawPackets)
	}

	return rawPackets
}

func checkPacket(p *packet) error {
	// All packets must adhere to these rules

	// This is the SCTP sender's port number.  It can be used by the
	// receiver in combination with the source IP address, the SCTP
	// destination port, and possibly the destination IP address to
	// identify the association to which this packet belongs.  The port
	// number 0 MUST NOT be used.
	if p.sourcePort == 0 {
		return errors.Errorf("sctp packet must not have a source port of 0")
	}

	// This is the SCTP port number to which this packet is destined.
	// The receiving host will use this port number to de-multiplex the
	// SCTP packet to the correct receiving endpoint/application.  The
	// port number 0 MUST NOT be used.
	if p.destinationPort == 0 {
		return errors.Errorf("sctp packet must not have a destination port of 0")
	}

	// Check values on the packet that are specific to a particular chunk type
	for _, c := range p.chunks {
		switch c.(type) {
		case *chunkInit:
			// An INIT or INIT ACK chunk MUST NOT be bundled with any other chunk.
			// They MUST be the only chunks present in the SCTP packets that carry
			// them.
			if len(p.chunks) != 1 {
				return errors.Errorf("init chunk must not be bundled with any other chunk")
			}

			// A packet containing an INIT chunk MUST have a zero Verification
			// Tag.
			if p.verificationTag != 0 {
				return errors.Errorf("init chunk expects a verification tag of 0 on the packet when out-of-the-blue")
			}
		}
	}

	return nil
}

func min16(a, b uint16) uint16 {
	if a < b {
		return a
	}
	return b
}

func max32(a, b uint32) uint32 {
	if a > b {
		return a
	}
	return b
}

func min32(a, b uint32) uint32 {
	if a < b {
		return a
	}
	return b
}

// setState atomically sets the state of the Association.
// The caller should hold the lock.
func (a *Association) setState(newState uint32) {
	oldState := atomic.SwapUint32(&a.state, newState)
	if newState != oldState {
		a.log.Debugf("[%s] state change: '%s' => '%s'",
			a.name,
			getAssociationStateString(oldState),
			getAssociationStateString(newState))
	}
}

// getState atomically returns the state of the Association.
func (a *Association) getState() uint32 {
	return atomic.LoadUint32(&a.state)
}

// BytesSent returns the number of bytes sent
func (a *Association) BytesSent() uint64 {
	return atomic.LoadUint64(&a.bytesSent)
}

// BytesReceived returns the number of bytes received
func (a *Association) BytesReceived() uint64 {
	return atomic.LoadUint64(&a.bytesReceived)
}

func setSupportedExtensions(init *chunkInitCommon) {
	// TODO RFC5061 https://tools.ietf.org/html/rfc6525#section-5.2
	// An implementation supporting this (Supported Extensions Parameter)
	// extension MUST list the ASCONF, the ASCONF-ACK, and the AUTH chunks
	// in its INIT and INIT-ACK parameters.
	init.params = append(init.params, &paramSupportedExtensions{
		ChunkTypes: []chunkType{ctReconfig, ctForwardTSN},
	})
}

// The caller should hold the lock.
func (a *Association) handleInit(p *packet, i *chunkInit) ([]*packet, error) {
	state := a.getState()
	a.log.Debugf("[%s] chunkInit received in state '%s'", a.name, getAssociationStateString(state))

	// https://tools.ietf.org/html/rfc4960#section-5.2.1
	// Upon receipt of an INIT in the COOKIE-WAIT state, an endpoint MUST
	// respond with an INIT ACK using the same parameters it sent in its
	// original INIT chunk (including its Initiate Tag, unchanged).  When
	// responding, the endpoint MUST send the INIT ACK back to the same
	// address that the original INIT (sent by this endpoint) was sent.

	// https://tools.ietf.org/html/rfc4960#section-5.2.1
	// Upon receipt of an INIT in the COOKIE-ECHOED state, an endpoint MUST
	// respond with an INIT ACK using the same parameters it sent in its
	// original INIT chunk (including its Initiate Tag, unchanged)

	if state != closed && state != cookieWait && state != cookieEchoed {
		// 5.2.2.  Unexpected INIT in States Other than CLOSED, COOKIE-ECHOED,
		//        COOKIE-WAIT, and SHUTDOWN-ACK-SENT
		return nil, errors.Errorf("todo: handle Init when in state %s", getAssociationStateString(state))
	}

	// Should we be setting any of these permanently until we've ACKed further?
	a.myMaxNumInboundStreams = min16(i.numInboundStreams, a.myMaxNumInboundStreams)
	a.myMaxNumOutboundStreams = min16(i.numOutboundStreams, a.myMaxNumOutboundStreams)
	a.peerVerificationTag = i.initiateTag
	a.sourcePort = p.destinationPort
	a.destinationPort = p.sourcePort

	// 13.2 This is the last TSN received in sequence.  This value
	// is set initially by taking the peer's initial TSN,
	// received in the INIT or INIT ACK chunk, and
	// subtracting one from it.
	a.peerLastTSN = i.initialTSN - 1

	outbound := &packet{}
	outbound.verificationTag = a.peerVerificationTag
	outbound.sourcePort = a.sourcePort
	outbound.destinationPort = a.destinationPort

	initAck := &chunkInitAck{}

	initAck.initialTSN = a.myNextTSN
	initAck.numOutboundStreams = a.myMaxNumOutboundStreams
	initAck.numInboundStreams = a.myMaxNumInboundStreams
	initAck.initiateTag = a.myVerificationTag
	initAck.advertisedReceiverWindowCredit = a.maxReceiveBufferSize

	if a.myCookie == nil {
		a.myCookie = newRandomStateCookie()
	}

	initAck.params = []param{a.myCookie}

	setSupportedExtensions(&initAck.chunkInitCommon)

	outbound.chunks = []chunk{initAck}

	return pack(outbound), nil
}

// The caller should hold the lock.
func (a *Association) handleInitAck(p *packet, i *chunkInitAck) error {
	state := a.getState()
	a.log.Debugf("[%s] chunkInitAck received in state '%s'", a.name, getAssociationStateString(state))
	if state != cookieWait {
		// RFC 4960
		// 5.2.3.  Unexpected INIT ACK
		//   If an INIT ACK is received by an endpoint in any state other than the
		//   COOKIE-WAIT state, the endpoint should discard the INIT ACK chunk.
		//   An unexpected INIT ACK usually indicates the processing of an old or
		//   duplicated INIT chunk.
		return nil
	}

	a.myMaxNumInboundStreams = min16(i.numInboundStreams, a.myMaxNumInboundStreams)
	a.myMaxNumOutboundStreams = min16(i.numOutboundStreams, a.myMaxNumOutboundStreams)
	a.peerVerificationTag = i.initiateTag
	a.peerLastTSN = i.initialTSN - 1
	if a.sourcePort != p.destinationPort ||
		a.destinationPort != p.sourcePort {
		a.log.Warnf("[%s] handleInitAck: port mismatch", a.name)
		return nil
	}

	a.rwnd = i.advertisedReceiverWindowCredit
	a.log.Debugf("[%s] initial rwnd=%d", a.name, a.rwnd)

	// RFC 4690 Sec 7.2.1
	//  o  The initial value of ssthresh MAY be arbitrarily high (for
	//     example, implementations MAY use the size of the receiver
	//     advertised window).
	a.ssthresh = a.rwnd
	a.log.Tracef("[%s] updated cwnd=%d ssthresh=%d inflight=%d (INI)",
		a.name, a.cwnd, a.ssthresh, a.inflightQueue.getNumBytes())

	a.t1Init.stop()
	a.storedInit = nil

	var cookieParam *paramStateCookie
	for _, param := range i.params {
		switch v := param.(type) {
		case *paramStateCookie:
			cookieParam = v
		case *paramSupportedExtensions:
			for _, t := range v.ChunkTypes {
				if t == ctForwardTSN {
					a.useForwardTSN = true
				}
			}
		}
	}
	if cookieParam == nil {
		return errors.Errorf("no cookie in InitAck")
	}

	a.storedCookieEcho = &chunkCookieEcho{}
	a.storedCookieEcho.cookie = cookieParam.cookie

	err := a.sendCookieEcho()
	if err != nil {
		a.log.Errorf("[%s] failed to send init: %s", a.name, err.Error())
	}

	a.t1Cookie.start(a.rtoMgr.getRTO())
	a.setState(cookieEchoed)
	return nil
}

// The caller should hold the lock.
func (a *Association) handleHeartbeat(c *chunkHeartbeat) []*packet {
	a.log.Tracef("[%s] chunkHeartbeat", a.name)
	hbi, ok := c.params[0].(*paramHeartbeatInfo)
	if !ok {
		a.log.Warnf("[%s] failed to handle Heartbeat, no ParamHeartbeatInfo", a.name)
	}

	return pack(&packet{
		verificationTag: a.peerVerificationTag,
		sourcePort:      a.sourcePort,
		destinationPort: a.destinationPort,
		chunks: []chunk{&chunkHeartbeatAck{
			params: []param{
				&paramHeartbeatInfo{
					heartbeatInformation: hbi.heartbeatInformation,
				},
			},
		}},
	})
}

// The caller should hold the lock.
func (a *Association) handleCookieEcho(c *chunkCookieEcho) []*packet {
	state := a.getState()
	a.log.Debugf("[%s] COOKIE-ECHO received in state '%s'", a.name, getAssociationStateString(state))
	if state != closed && state != cookieWait && state != cookieEchoed {
		return nil
	}

	if !bytes.Equal(a.myCookie.cookie, c.cookie) {
		return nil
	}

	a.t1Init.stop()
	a.storedInit = nil

	a.t1Cookie.stop()
	a.storedCookieEcho = nil

	p := &packet{
		verificationTag: a.peerVerificationTag,
		sourcePort:      a.sourcePort,
		destinationPort: a.destinationPort,
		chunks:          []chunk{&chunkCookieAck{}},
	}

	a.setState(established)
	a.handshakeCompletedCh <- nil
	return pack(p)
}

// The caller should hold the lock.
func (a *Association) handleCookieAck() []*packet {
	state := a.getState()
	a.log.Debugf("[%s] COOKIE-ACK received in state '%s'", a.name, getAssociationStateString(state))
	if state != cookieEchoed {
		// RFC 4960
		// 5.2.5.  Handle Duplicate COOKIE-ACK.
		//   At any state other than COOKIE-ECHOED, an endpoint should silently
		//   discard a received COOKIE ACK chunk.
		return nil
	}

	a.t1Cookie.stop()
	a.storedCookieEcho = nil

	a.setState(established)
	a.handshakeCompletedCh <- nil
	return nil
}

// The caller should hold the lock.
func (a *Association) handleData(d *chunkPayloadData) []*packet {
	a.log.Tracef("[%s] DATA: tsn=%d immediateSack=%v len=%d",
		a.name, d.tsn, d.immediateSack, len(d.userData))
	a.stats.incDATAs()

	canPush := a.payloadQueue.canPush(d, a.peerLastTSN)
	if canPush {
		s := a.getOrCreateStream(d.streamIdentifier)
		if s == nil {
			// silentely discard the data. (sender will retry on T3-rtx timeout)
			// see pion/sctp#30
			a.log.Debugf("discard %d", d.streamSequenceNumber)
			return nil
		}

		if a.getMyReceiverWindowCredit() > 0 {
			// Pass the new chunk to stream level as soon as it arrives
			a.payloadQueue.push(d, a.peerLastTSN)
			s.handleData(d)
		} else {
			// Receive buffer is full
			lastTSN, ok := a.payloadQueue.getLastTSNReceived()
			if ok && sna32LT(d.tsn, lastTSN) {
				a.log.Debugf("[%s] receive buffer full, but accepted as this is a missing chunk with tsn=%d ssn=%d", a.name, d.tsn, d.streamSequenceNumber)
				a.payloadQueue.push(d, a.peerLastTSN)
				s.handleData(d)
			} else {
				a.log.Debugf("[%s] receive buffer full. dropping DATA with tsn=%d ssn=%d", a.name, d.tsn, d.streamSequenceNumber)
			}
		}
	}

	return a.handlePeerLastTSNAndAcknowledgement(d.immediateSack)
}

// A common routine for handleData and handleForwardTSN routines
// The caller should hold the lock.
func (a *Association) handlePeerLastTSNAndAcknowledgement(sackImmediately bool) []*packet {
	var reply []*packet

	// Try to advance peerLastTSN

	// From RFC 3758 Sec 3.6:
	//   .. and then MUST further advance its cumulative TSN point locally
	//   if possible
	// Meaning, if peerLastTSN+1 points to a chunk that is received,
	// advance peerLastTSN until peerLastTSN+1 points to unreceived chunk.
	for {
		if _, popOk := a.payloadQueue.pop(a.peerLastTSN + 1); !popOk {
			break
		}
		a.peerLastTSN++

		for _, rstReq := range a.reconfigRequests {
			resp := a.resetStreamsIfAny(rstReq)
			if resp != nil {
				a.log.Debugf("[%s] RESET RESPONSE: %+v", a.name, resp)
				reply = append(reply, resp)
			}
		}
	}

	hasPacketLoss := (a.payloadQueue.size() > 0)
	if hasPacketLoss {
		a.log.Tracef("[%s] packetloss: %s", a.name, a.payloadQueue.getGapAckBlocksString(a.peerLastTSN))
	}

	if (a.ackState != ackStateImmediate && !sackImmediately && !hasPacketLoss && a.ackMode == ackModeNormal) || a.ackMode == ackModeAlwaysDelay {
		if a.ackState == ackStateIdle {
			a.delayedAckTriggered = true
		} else {
			a.immediateAckTriggered = true
		}
	} else {
		a.immediateAckTriggered = true
	}

	return reply
}

// The caller should hold the lock.
func (a *Association) getMyReceiverWindowCredit() uint32 {
	var bytesQueued uint32
	for _, s := range a.streams {
		bytesQueued += uint32(s.getNumBytesInReassemblyQueue())
	}

	if bytesQueued >= a.maxReceiveBufferSize {
		return 0
	}
	return a.maxReceiveBufferSize - bytesQueued
}

// OpenStream opens a stream
func (a *Association) OpenStream(streamIdentifier uint16, defaultPayloadType PayloadProtocolIdentifier) (*Stream, error) {
	a.lock.Lock()
	defer a.lock.Unlock()

	if _, ok := a.streams[streamIdentifier]; ok {
		return nil, errors.Errorf("there already exists a stream with identifier %d", streamIdentifier)
	}

	s := a.createStream(streamIdentifier, false)
	s.setDefaultPayloadType(defaultPayloadType)

	return s, nil
}

// AcceptStream accepts a stream
func (a *Association) AcceptStream() (*Stream, error) {
	s, ok := <-a.acceptCh
	if !ok {
		return nil, io.EOF // no more incoming streams
	}
	return s, nil
}

// createStream creates a stream. The caller should hold the lock and check no stream exists for this id.
func (a *Association) createStream(streamIdentifier uint16, accept bool) *Stream {
	s := &Stream{
		association:      a,
		streamIdentifier: streamIdentifier,
		reassemblyQueue:  newReassemblyQueue(streamIdentifier),
		log:              a.log,
		name:             fmt.Sprintf("%d:%s", streamIdentifier, a.name),
	}

	s.readNotifier = sync.NewCond(&s.lock)

	if accept {
		select {
		case a.acceptCh <- s:
			a.streams[streamIdentifier] = s
			a.log.Debugf("[%s] accepted a new stream (streamIdentifier: %d)",
				a.name, streamIdentifier)
		default:
			a.log.Debugf("[%s] dropped a new stream (acceptCh size: %d)",
				a.name, len(a.acceptCh))
			return nil
		}
	} else {
		a.streams[streamIdentifier] = s
	}

	return s
}

// getOrCreateStream gets or creates a stream. The caller should hold the lock.
func (a *Association) getOrCreateStream(streamIdentifier uint16) *Stream {
	if s, ok := a.streams[streamIdentifier]; ok {
		return s
	}

	return a.createStream(streamIdentifier, true)
}

// The caller should hold the lock.
func (a *Association) processSelectiveAck(d *chunkSelectiveAck) (int, uint32, error) {
	var totalBytesAcked int

	// New ack point, so pop all ACKed packets from inflightQueue
	// We add 1 because the "currentAckPoint" has already been popped from the inflight queue
	// For the first SACK we take care of this by setting the ackpoint to cumAck - 1
	for i := a.cumulativeTSNAckPoint + 1; sna32LTE(i, d.cumulativeTSNAck); i++ {
		c, ok := a.inflightQueue.pop(i)
		if !ok {
			return 0, 0, errors.Errorf("tsn %v unable to be popped from inflight queue", i)
		}

		if !c.acked {
			// RFC 4096 sec 6.3.2.  Retransmission Timer Rules
			//   R3)  Whenever a SACK is received that acknowledges the DATA chunk
			//        with the earliest outstanding TSN for that address, restart the
			//        T3-rtx timer for that address with its current RTO (if there is
			//        still outstanding data on that address).
			if i == a.cumulativeTSNAckPoint+1 {
				// T3 timer needs to be reset. Stop it for now.
				a.t3RTX.stop()
			}

			nBytesAcked := len(c.userData)

			// Report the number of bytes acknowledged to the stream who sent this DATA
			// chunk.
			if s, ok := a.streams[c.streamIdentifier]; ok {
				a.lock.Unlock()
				s.onBufferReleased(nBytesAcked)
				a.lock.Lock()
			}

			totalBytesAcked += nBytesAcked

			// RFC 4960 sec 6.3.1.  RTO Calculation
			//   C4)  When data is in flight and when allowed by rule C5 below, a new
			//        RTT measurement MUST be made each round trip.  Furthermore, new
			//        RTT measurements SHOULD be made no more than once per round trip
			//        for a given destination transport address.
			//   C5)  Karn's algorithm: RTT measurements MUST NOT be made using
			//        packets that were retransmitted (and thus for which it is
			//        ambiguous whether the reply was for the first instance of the
			//        chunk or for a later instance)
			if c.nSent == 1 && sna32GTE(c.tsn, a.minTSN2MeasureRTT) {
				a.minTSN2MeasureRTT = a.myNextTSN
				rtt := time.Since(c.since).Seconds() * 1000.0
				srtt := a.rtoMgr.setNewRTT(rtt)
				a.log.Tracef("[%s] SACK: measured-rtt=%f srtt=%f new-rto=%f",
					a.name, rtt, srtt, a.rtoMgr.getRTO())
			}
		}

		if a.inFastRecovery && c.tsn == a.fastRecoverExitPoint {
			a.log.Debugf("[%s] exit fast-recovery", a.name)
			a.inFastRecovery = false
		}
	}

	htna := d.cumulativeTSNAck

	// Mark selectively acknowledged chunks as "acked"
	for _, g := range d.gapAckBlocks {
		for i := g.start; i <= g.end; i++ {
			tsn := d.cumulativeTSNAck + uint32(i)
			c, ok := a.inflightQueue.get(tsn)
			if !ok {
				return 0, 0, errors.Errorf("requested non-existent TSN %v", tsn)
			}

			if !c.acked {
				nBytesAcked := a.inflightQueue.markAsAcked(tsn)

				// Report the number of bytes acknowledged to the stream who sent this DATA
				// chunk.
				if s, ok := a.streams[c.streamIdentifier]; ok {
					a.lock.Unlock()
					s.onBufferReleased(nBytesAcked)
					a.lock.Lock()
				}

				totalBytesAcked += nBytesAcked

				a.log.Tracef("[%s] tsn=%d has been sacked", a.name, c.tsn)

				if c.nSent == 1 {
					a.minTSN2MeasureRTT = a.myNextTSN
					rtt := time.Since(c.since).Seconds() * 1000.0
					srtt := a.rtoMgr.setNewRTT(rtt)
					a.log.Tracef("[%s] SACK: measured-rtt=%f srtt=%f new-rto=%f",
						a.name, rtt, srtt, a.rtoMgr.getRTO())
				}

				if sna32LT(htna, tsn) {
					htna = tsn
				}
			}
		}
	}

	return totalBytesAcked, htna, nil
}

// The caller should hold the lock.
func (a *Association) onCumulativeTSNAckPointAdvanced(totalBytesAcked int) {
	// RFC 4096, sec 6.3.2.  Retransmission Timer Rules
	//   R2)  Whenever all outstanding data sent to an address have been
	//        acknowledged, turn off the T3-rtx timer of that address.
	if a.inflightQueue.size() == 0 {
		a.log.Tracef("[%s] SACK: no more packet in-flight (pending=%d)", a.name, a.pendingQueue.size())
		a.t3RTX.stop()
	} else {
		a.log.Tracef("[%s] T3-rtx timer start (pt2)", a.name)
		a.t3RTX.start(a.rtoMgr.getRTO())
	}

	// Update congestion control parameters
	if a.cwnd <= a.ssthresh {
		// RFC 4096, sec 7.2.1.  Slow-Start
		//   o  When cwnd is less than or equal to ssthresh, an SCTP endpoint MUST
		//		use the slow-start algorithm to increase cwnd only if the current
		//      congestion window is being fully utilized, an incoming SACK
		//      advances the Cumulative TSN Ack Point, and the data sender is not
		//      in Fast Recovery.  Only when these three conditions are met can
		//      the cwnd be increased; otherwise, the cwnd MUST not be increased.
		//		If these conditions are met, then cwnd MUST be increased by, at
		//      most, the lesser of 1) the total size of the previously
		//      outstanding DATA chunk(s) acknowledged, and 2) the destination's
		//      path MTU.
		if !a.inFastRecovery &&
			a.pendingQueue.size() > 0 {
			a.cwnd += min32(uint32(totalBytesAcked), a.cwnd) // TCP way
			//a.cwnd += min32(uint32(totalBytesAcked), a.mtu) // SCTP way (slow)
			a.log.Tracef("[%s] updated cwnd=%d ssthresh=%d acked=%d (SS)",
				a.name, a.cwnd, a.ssthresh, totalBytesAcked)
		} else {
			a.log.Tracef("[%s] cwnd did not grow: cwnd=%d ssthresh=%d acked=%d FR=%v pending=%d",
				a.name, a.cwnd, a.ssthresh, totalBytesAcked, a.inFastRecovery, a.pendingQueue.size())
		}
	} else {
		// RFC 4096, sec 7.2.2.  Congestion Avoidance
		//   o  Whenever cwnd is greater than ssthresh, upon each SACK arrival
		//      that advances the Cumulative TSN Ack Point, increase
		//      partial_bytes_acked by the total number of bytes of all new chunks
		//      acknowledged in that SACK including chunks acknowledged by the new
		//      Cumulative TSN Ack and by Gap Ack Blocks.
		a.partialBytesAcked += uint32(totalBytesAcked)

		//   o  When partial_bytes_acked is equal to or greater than cwnd and
		//      before the arrival of the SACK the sender had cwnd or more bytes
		//      of data outstanding (i.e., before arrival of the SACK, flight size
		//      was greater than or equal to cwnd), increase cwnd by MTU, and
		//      reset partial_bytes_acked to (partial_bytes_acked - cwnd).
		if a.partialBytesAcked >= a.cwnd && a.pendingQueue.size() > 0 {
			a.partialBytesAcked -= a.cwnd
			a.cwnd += a.mtu
			a.log.Tracef("[%s] updated cwnd=%d ssthresh=%d acked=%d (CA)",
				a.name, a.cwnd, a.ssthresh, totalBytesAcked)
		}
	}
}

// The caller should hold the lock.
func (a *Association) processFastRetransmission(cumTSNAckPoint, htna uint32, cumTSNAckPointAdvanced bool) error {
	// HTNA algorithm - RFC 4960 Sec 7.2.4
	// Increment missIndicator of each chunks that the SACK reported missing
	// when either of the following is met:
	// a)  Not in fast-recovery
	//     miss indications are incremented only for missing TSNs prior to the
	//     highest TSN newly acknowledged in the SACK.
	// b)  In fast-recovery AND the Cumulative TSN Ack Point advanced
	//     the miss indications are incremented for all TSNs reported missing
	//     in the SACK.
	if !a.inFastRecovery || (a.inFastRecovery && cumTSNAckPointAdvanced) {
		var maxTSN uint32
		if !a.inFastRecovery {
			// a) increment only for missing TSNs prior to the HTNA
			maxTSN = htna
		} else {
			// b) increment for all TSNs reported missing
			maxTSN = cumTSNAckPoint + uint32(a.inflightQueue.size()) + 1
		}

		for tsn := cumTSNAckPoint + 1; sna32LT(tsn, maxTSN); tsn++ {
			c, ok := a.inflightQueue.get(tsn)
			if !ok {
				return errors.Errorf("requested non-existent TSN %v", tsn)
			}
			if !c.acked && !c.abandoned && c.missIndicator < 3 {
				c.missIndicator++
				if c.missIndicator == 3 {
					if !a.inFastRecovery {
						// 2)  If not in Fast Recovery, adjust the ssthresh and cwnd of the
						//     destination address(es) to which the missing DATA chunks were
						//     last sent, according to the formula described in Section 7.2.3.
						a.inFastRecovery = true
						a.fastRecoverExitPoint = htna
						a.ssthresh = max32(a.cwnd/2, 4*a.mtu)
						a.cwnd = a.ssthresh
						a.partialBytesAcked = 0
						a.willRetransmitFast = true

						a.log.Tracef("[%s] updated cwnd=%d ssthresh=%d inflight=%d (FR)",
							a.name, a.cwnd, a.ssthresh, a.inflightQueue.getNumBytes())
					}
				}
			}
		}
	}

	if a.inFastRecovery && cumTSNAckPointAdvanced {
		a.willRetransmitFast = true
	}

	return nil
}

// The caller should hold the lock.
func (a *Association) handleSack(d *chunkSelectiveAck) error {
	a.log.Tracef("[%s] SACK: cumTSN=%d a_rwnd=%d", a.name, d.cumulativeTSNAck, d.advertisedReceiverWindowCredit)
	state := a.getState()
	if state != established {
		return nil
	}

	a.stats.incSACKs()

	if sna32GT(a.cumulativeTSNAckPoint, d.cumulativeTSNAck) {
		// RFC 4960 sec 6.2.1.  Processing a Received SACK
		// D)
		//   i) If Cumulative TSN Ack is less than the Cumulative TSN Ack
		//      Point, then drop the SACK.  Since Cumulative TSN Ack is
		//      monotonically increasing, a SACK whose Cumulative TSN Ack is
		//      less than the Cumulative TSN Ack Point indicates an out-of-
		//      order SACK.

		a.log.Debugf("[%s] SACK Cumulative ACK %v is older than ACK point %v",
			a.name,
			d.cumulativeTSNAck,
			a.cumulativeTSNAckPoint)

		return nil
	}

	// Process selective ack
	totalBytesAcked, htna, err := a.processSelectiveAck(d)
	if err != nil {
		return err
	}

	// New rwnd value
	// RFC 4960 sec 6.2.1.  Processing a Received SACK
	// D)
	//   ii) Set rwnd equal to the newly received a_rwnd minus the number
	//       of bytes still outstanding after processing the Cumulative
	//       TSN Ack and the Gap Ack Blocks.

	// bytes acked were already subtracted by markAsAcked() method
	bytesOutstanding := uint32(a.inflightQueue.getNumBytes())
	if bytesOutstanding >= d.advertisedReceiverWindowCredit {
		a.rwnd = 0
	} else {
		a.rwnd = d.advertisedReceiverWindowCredit - bytesOutstanding
	}

	cumTSNAckPointAdvanced := false
	if sna32LT(a.cumulativeTSNAckPoint, d.cumulativeTSNAck) {
		a.log.Tracef("[%s] SACK: cumTSN advanced: %d -> %d",
			a.name,
			a.cumulativeTSNAckPoint,
			d.cumulativeTSNAck)

		a.cumulativeTSNAckPoint = d.cumulativeTSNAck
		cumTSNAckPointAdvanced = true
		a.onCumulativeTSNAckPointAdvanced(totalBytesAcked)
	}

	err = a.processFastRetransmission(d.cumulativeTSNAck, htna, cumTSNAckPointAdvanced)
	if err != nil {
		return err
	}

	if a.useForwardTSN {
		// RFC 3758 Sec 3.5 C1
		if sna32LT(a.advancedPeerTSNAckPoint, a.cumulativeTSNAckPoint) {
			a.advancedPeerTSNAckPoint = a.cumulativeTSNAckPoint
		}

		// RFC 3758 Sec 3.5 C2
		for i := a.advancedPeerTSNAckPoint + 1; ; i++ {
			c, ok := a.inflightQueue.get(i)
			if !ok {
				break
			}
			if !c.abandoned {
				break
			}
			a.advancedPeerTSNAckPoint = i
		}

		// RFC 3758 Sec 3.5 C3
		a.willSendForwardTSN = true
		a.awakeWriteLoop()
	}

	if a.inflightQueue.size() > 0 {
		// Start timer. (noop if already started)
		a.log.Tracef("[%s] T3-rtx timer start (pt3)", a.name)
		a.t3RTX.start(a.rtoMgr.getRTO())
	}

	if cumTSNAckPointAdvanced {
		a.awakeWriteLoop()
	}

	return nil
}

// createForwardTSN generates ForwardTSN chunk.
// This method will be be called if useForwardTSN is set to false.
// The caller should hold the lock.
func (a *Association) createForwardTSN() *chunkForwardTSN {
	// RFC 3758 Sec 3.5 C4
	streamMap := map[uint16]uint16{} // to report only once per SI
	for i := a.cumulativeTSNAckPoint + 1; sna32LTE(i, a.advancedPeerTSNAckPoint); i++ {
		c, ok := a.inflightQueue.get(i)
		if !ok {
			break
		}
		if c.acked {
			continue
		}

		ssn, ok := streamMap[c.streamIdentifier]
		if !ok {
			streamMap[c.streamIdentifier] = c.streamSequenceNumber
		} else {
			// to report only once with greatest SSN
			if sna16LT(ssn, c.streamSequenceNumber) {
				streamMap[c.streamIdentifier] = c.streamSequenceNumber
			}
		}
	}

	fwdtsn := &chunkForwardTSN{
		newCumulativeTSN: a.advancedPeerTSNAckPoint,
		streams:          []chunkForwardTSNStream{},
	}

	var streamStr string
	for si, ssn := range streamMap {
		streamStr += fmt.Sprintf("(si=%d ssn=%d)", si, ssn)
		fwdtsn.streams = append(fwdtsn.streams, chunkForwardTSNStream{
			identifier: si,
			sequence:   ssn,
		})
	}
	a.log.Tracef("[%s] building fwdtsn: newCumulativeTSN=%d - %s", a.name, fwdtsn.newCumulativeTSN, streamStr)

	return fwdtsn
}

// createPacket wraps chunks in a packet.
// The caller should hold the read lock.
func (a *Association) createPacket(cs []chunk) *packet {
	return &packet{
		verificationTag: a.peerVerificationTag,
		sourcePort:      a.sourcePort,
		destinationPort: a.destinationPort,
		chunks:          cs,
	}
}

// The caller should hold the lock.
func (a *Association) handleReconfig(c *chunkReconfig) ([]*packet, error) {
	a.log.Tracef("[%s] handleReconfig", a.name)

	pp := make([]*packet, 0)

	p, err := a.handleReconfigParam(c.paramA)
	if err != nil {
		return nil, err
	}
	if p != nil {
		pp = append(pp, p)
	}

	if c.paramB != nil {
		p, err = a.handleReconfigParam(c.paramB)
		if err != nil {
			return nil, err
		}
		if p != nil {
			pp = append(pp, p)
		}
	}
	return pp, nil
}

// The caller should hold the lock.
func (a *Association) handleForwardTSN(c *chunkForwardTSN) []*packet {
	a.log.Tracef("[%s] FwdTSN: %s", a.name, c.String())

	if !a.useForwardTSN {
		// Return an error chunk
		cerr := &chunkError{
			errorCauses: []errorCause{&errorCauseUnrecognizedChunkType{}},
		}
		outbound := &packet{}
		outbound.verificationTag = a.peerVerificationTag
		outbound.sourcePort = a.sourcePort
		outbound.destinationPort = a.destinationPort
		outbound.chunks = []chunk{cerr}
		return []*packet{outbound}
	}

	// From RFC 3758 Sec 3.6:
	//   Note, if the "New Cumulative TSN" value carried in the arrived
	//   FORWARD TSN chunk is found to be behind or at the current cumulative
	//   TSN point, the data receiver MUST treat this FORWARD TSN as out-of-
	//   date and MUST NOT update its Cumulative TSN.  The receiver SHOULD
	//   send a SACK to its peer (the sender of the FORWARD TSN) since such a
	//   duplicate may indicate the previous SACK was lost in the network.

	if sna32LTE(c.newCumulativeTSN, a.peerLastTSN) {
		a.log.Tracef("[%s] sending ack on Forward TSN", a.name)
		a.ackState = ackStateImmediate
		a.ackTimer.stop()
		a.awakeWriteLoop()
		return nil
	}

	// From RFC 3758 Sec 3.6:
	//   the receiver MUST perform the same TSN handling, including duplicate
	//   detection, gap detection, SACK generation, cumulative TSN
	//   advancement, etc. as defined in RFC 2960 [2]---with the following
	//   exceptions and additions.

	//   When a FORWARD TSN chunk arrives, the data receiver MUST first update
	//   its cumulative TSN point to the value carried in the FORWARD TSN
	//   chunk,

	// Advance peerLastTSN
	for sna32LT(a.peerLastTSN, c.newCumulativeTSN) {
		a.payloadQueue.pop(a.peerLastTSN + 1) // may not exist
		a.peerLastTSN++
	}

	// Report new peerLastTSN value and abandoned largest SSN value to
	// corresponding streams so that the abandoned chunks can be removed
	// from the reassemblyQueue.
	for _, forwarded := range c.streams {
		if s, ok := a.streams[forwarded.identifier]; ok {
			s.handleForwardTSN(c.newCumulativeTSN, forwarded.sequence)
		}
	}

	return a.handlePeerLastTSNAndAcknowledgement(false)
}

func (a *Association) sendResetRequest(streamIdentifier uint16) error {
	a.lock.Lock()
	defer a.lock.Unlock()

	state := a.getState()
	if state != established {
		return errors.Errorf("sending reset packet in non-established state: state=%s",
			getAssociationStateString(state))
	}

	// Create DATA chunk which only contains valid stream identifier with
	// nil userData and use it as a EOS from the stream.
	c := &chunkPayloadData{
		streamIdentifier:  streamIdentifier,
		beginningFragment: true,
		endingFragment:    true,
		userData:          nil,
	}

	a.pendingQueue.push(c)
	a.awakeWriteLoop()
	return nil
}

// The caller should hold the lock.
func (a *Association) handleReconfigParam(raw param) (*packet, error) {
	// TODO: Check RSN
	switch p := raw.(type) {
	case *paramOutgoingResetRequest:
		a.reconfigRequests[p.reconfigRequestSequenceNumber] = p
		resp := a.resetStreamsIfAny(p)
		if resp != nil {
			return resp, nil
		}
		return nil, nil

	case *paramReconfigResponse:
		delete(a.reconfigs, p.reconfigResponseSequenceNumber)
		if len(a.reconfigs) == 0 {
			a.tReconfig.stop()
		}
		return nil, nil
	default:
		return nil, errors.Errorf("unexpected parameter type %T", p)
	}
}

// The caller should hold the lock.
func (a *Association) resetStreamsIfAny(p *paramOutgoingResetRequest) *packet {
	result := reconfigResultSuccessPerformed
	if sna32LTE(p.senderLastTSN, a.peerLastTSN) {
		a.log.Debugf("[%s] resetStream(): senderLastTSN=%d <= peerLastTSN=%d",
			a.name, p.senderLastTSN, a.peerLastTSN)
		for _, id := range p.streamIdentifiers {
			s, ok := a.streams[id]
			if !ok {
				continue
			}
			a.unregisterStream(s, io.EOF)
		}
		delete(a.reconfigRequests, p.reconfigRequestSequenceNumber)
	} else {
		a.log.Debugf("[%s] resetStream(): senderLastTSN=%d > peerLastTSN=%d",
			a.name, p.senderLastTSN, a.peerLastTSN)
		result = reconfigResultInProgress
	}

	return a.createPacket([]chunk{&chunkReconfig{
		paramA: &paramReconfigResponse{
			reconfigResponseSequenceNumber: p.reconfigRequestSequenceNumber,
			result:                         result,
		},
	}})
}

// Move the chunk peeked with a.pendingQueue.peek() to the inflightQueue.
// The caller should hold the lock.
func (a *Association) movePendingDataChunkToInflightQueue(c *chunkPayloadData) {
	if err := a.pendingQueue.pop(c); err != nil {
		a.log.Errorf("[%s] failed to pop from pending queue: %s", a.name, err.Error())
	}

	// Assign TSN
	c.tsn = a.generateNextTSN()

	c.since = time.Now() // use to calculate RTT and also for maxPacketLifeTime
	c.nSent = 1          // being sent for the first time

	a.checkPartialReliabilityStatus(c)

	a.log.Tracef("[%s] sending tsn=%d ssn=%d sent=%d len=%d",
		a.name, c.tsn, c.streamSequenceNumber, c.nSent, len(c.userData))

	// Push it into the inflightQueue
	a.inflightQueue.pushNoCheck(c)
}

// popPendingDataChunksToSend pops chunks from the pending queues as many as
// the cwnd and rwnd allows to send.
// The caller should hold the lock.
func (a *Association) popPendingDataChunksToSend() ([]*chunkPayloadData, []uint16) {
	chunks := []*chunkPayloadData{}
	var sisToReset []uint16 // stream identifieres to reset

	if a.pendingQueue.size() > 0 {
		// RFC 4960 sec 6.1.  Transmission of DATA Chunks
		//   A) At any given time, the data sender MUST NOT transmit new data to
		//      any destination transport address if its peer's rwnd indicates
		//      that the peer has no buffer space (i.e., rwnd is 0; see Section
		//      6.2.1).  However, regardless of the value of rwnd (including if it
		//      is 0), the data sender can always have one DATA chunk in flight to
		//      the receiver if allowed by cwnd (see rule B, below).

		for {
			c := a.pendingQueue.peek()
			if c == nil {
				break // no more pending data
			}

			dataLen := uint32(len(c.userData))
			if dataLen == 0 {
				sisToReset = append(sisToReset, c.streamIdentifier)
				err := a.pendingQueue.pop(c)
				if err != nil {
					a.log.Errorf("failed to pop from pending queue: %s", err.Error())
				}
				continue
			}

			if uint32(a.inflightQueue.getNumBytes())+dataLen > a.cwnd {
				break // would exceeds cwnd
			}

			if dataLen > a.rwnd {
				break // no more rwnd
			}

			a.rwnd -= dataLen

			a.movePendingDataChunkToInflightQueue(c)
			chunks = append(chunks, c)
		}

		// the data sender can always have one DATA chunk in flight to the receiver
		if len(chunks) == 0 && a.inflightQueue.size() == 0 {
			// Send zero window probe
			c := a.pendingQueue.peek()
			if c != nil {
				a.movePendingDataChunkToInflightQueue(c)
				chunks = append(chunks, c)
			}
		}
	}

	return chunks, sisToReset
}

// bundleDataChunksIntoPackets packs DATA chunks into packets. It tries to bundle
// DATA chunks into a packet so long as the resulting packet size does not exceed
// the path MTU.
// The caller should hold the lock.
func (a *Association) bundleDataChunksIntoPackets(chunks []*chunkPayloadData) []*packet {
	packets := []*packet{}
	chunksToSend := []chunk{}
	bytesInPacket := int(commonHeaderSize)

	for _, c := range chunks {
		// RFC 4960 sec 6.1.  Transmission of DATA Chunks
		//   Multiple DATA chunks committed for transmission MAY be bundled in a
		//   single packet.  Furthermore, DATA chunks being retransmitted MAY be
		//   bundled with new DATA chunks, as long as the resulting packet size
		//   does not exceed the path MTU.
		if bytesInPacket+len(c.userData) > int(a.mtu) {
			packets = append(packets, a.createPacket(chunksToSend))
			chunksToSend = []chunk{}
			bytesInPacket = int(commonHeaderSize)
		}

		chunksToSend = append(chunksToSend, c)
		bytesInPacket += int(dataChunkHeaderSize) + len(c.userData)
	}

	if len(chunksToSend) > 0 {
		packets = append(packets, a.createPacket(chunksToSend))
	}

	return packets
}

// sendPayloadData sends the data chunks.
func (a *Association) sendPayloadData(chunks []*chunkPayloadData) error {
	a.lock.Lock()
	defer a.lock.Unlock()

	state := a.getState()
	if state != established {
		return errors.Errorf("sending payload data in non-established state: state=%s",
			getAssociationStateString(state))
	}

	// Push the chunks into the pending queue first.
	for _, c := range chunks {
		a.pendingQueue.push(c)
	}

	a.awakeWriteLoop()
	return nil
}

// The caller should hold the lock.
func (a *Association) checkPartialReliabilityStatus(c *chunkPayloadData) {
	if !a.useForwardTSN {
		return
	}

	// PR-SCTP
	if s, ok := a.streams[c.streamIdentifier]; ok {
		s.lock.RLock()
		if s.reliabilityType == ReliabilityTypeRexmit {
			if c.nSent >= s.reliabilityValue {
				c.abandoned = true
				a.log.Tracef("[%s] final (abandoned) tsn=%d (remix: %d)", a.name, c.tsn, c.nSent)
			}
		} else if s.reliabilityType == ReliabilityTypeTimed {
			elapsed := int64(time.Since(c.since).Seconds() * 1000)
			if elapsed >= int64(s.reliabilityValue) {
				c.abandoned = true
				a.log.Tracef("[%s] final (abandoned) tsn=%d (timed: %d)", a.name, c.tsn, elapsed)
			}
		}
		s.lock.RUnlock()
	}
}

// getDataPacketsToRetransmit is called when T3-rtx is timed out and retransmit outstanding data chunks
// that are not acked or abandoned yet.
// The caller should hold the lock.
func (a *Association) getDataPacketsToRetransmit() []*packet {
	awnd := min32(a.cwnd, a.rwnd)
	chunks := []*chunkPayloadData{}
	var bytesToSend int
	var done bool

	for i := 0; !done; i++ {
		c, ok := a.inflightQueue.get(a.cumulativeTSNAckPoint + uint32(i) + 1)
		if !ok {
			break // end of pending data
		}

		if !c.retransmit {
			continue
		}

		if i == 0 && int(a.rwnd) < len(c.userData) {
			// Send it as a zero window probe
			done = true
		} else if bytesToSend+len(c.userData) > int(awnd) {
			break
		}

		// reset the retransmit flag not to retransmit again before the next
		// t3-rtx timer fires
		c.retransmit = false
		bytesToSend += len(c.userData)

		c.nSent++

		a.checkPartialReliabilityStatus(c)

		a.log.Tracef("[%s] retransmitting tsn=%d ssn=%d sent=%d", a.name, c.tsn, c.streamSequenceNumber, c.nSent)

		chunks = append(chunks, c)
	}

	return a.bundleDataChunksIntoPackets(chunks)
}

// generateNextTSN returns the myNextTSN and increases it. The caller should hold the lock.
// The caller should hold the lock.
func (a *Association) generateNextTSN() uint32 {
	tsn := a.myNextTSN
	a.myNextTSN++
	return tsn
}

// generateNextRSN returns the myNextRSN and increases it. The caller should hold the lock.
// The caller should hold the lock.
func (a *Association) generateNextRSN() uint32 {
	rsn := a.myNextRSN
	a.myNextRSN++
	return rsn
}

func (a *Association) createSelectiveAckChunk() *chunkSelectiveAck {
	sack := &chunkSelectiveAck{}
	sack.cumulativeTSNAck = a.peerLastTSN
	sack.advertisedReceiverWindowCredit = a.getMyReceiverWindowCredit()
	sack.duplicateTSN = a.payloadQueue.popDuplicates()
	sack.gapAckBlocks = a.payloadQueue.getGapAckBlocks(a.peerLastTSN)
	return sack
}

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()

	var packets []*packet
	var err error

	if _, err = c.check(); err != nil {
		return errors.Wrap(err, "failed validating chunk")
		// TODO: Create ABORT
	}

	switch c := c.(type) {
	case *chunkInit:
		packets, err = a.handleInit(p, c)

	case *chunkInitAck:
		err = a.handleInitAck(p, c)

	case *chunkAbort:
		var errStr string
		for _, e := range c.errorCauses {
			errStr += fmt.Sprintf("(%s)", e)
		}
		a.log.Debugf("[%s] Abort chunk, with following errors: %s", a.name, errStr)

	case *chunkError:
		var errStr string
		for _, e := range c.errorCauses {
			errStr += fmt.Sprintf("(%s)", e)
		}
		a.log.Debugf("[%s] Error chunk, with following errors: %s", a.name, errStr)

	case *chunkHeartbeat:
		packets = a.handleHeartbeat(c)

	case *chunkCookieEcho:
		packets = a.handleCookieEcho(c)

	case *chunkCookieAck:
		packets = a.handleCookieAck()

	case *chunkPayloadData:
		packets = a.handleData(c)

	case *chunkSelectiveAck:
		err = a.handleSack(c)

	case *chunkReconfig:
		packets, err = a.handleReconfig(c)

	case *chunkForwardTSN:
		packets = a.handleForwardTSN(c)

	default:
		err = errors.Errorf("unhandled chunk type")
	}

	if err != nil {
		return err
	}

	if len(packets) > 0 {
		a.controlQueue.pushAll(packets)
		a.awakeWriteLoop()
	}

	return nil
}

func (a *Association) onRetransmissionTimeout(id int, nRtos uint) {
	a.lock.Lock()
	defer a.lock.Unlock()

	if id == timerT1Init {
		err := a.sendInit()
		if err != nil {
			a.log.Debugf("[%s] failed to retransmit init (nRtos=%d): %v", a.name, nRtos, err)
		}
		return
	}

	if id == timerT1Cookie {
		err := a.sendCookieEcho()
		if err != nil {
			a.log.Debugf("[%s] failed to retransmit cookie-echo (nRtos=%d): %v", a.name, nRtos, err)
		}
		return
	}

	if id == timerT3RTX {
		a.stats.incT3Timeouts()

		// RFC 4960 sec 6.3.3
		//  E1)  For the destination address for which the timer expires, adjust
		//       its ssthresh with rules defined in Section 7.2.3 and set the
		//       cwnd <- MTU.
		// RFC 4960 sec 7.2.3
		//   When the T3-rtx timer expires on an address, SCTP should perform slow
		//   start by:
		//      ssthresh = max(cwnd/2, 4*MTU)
		//      cwnd = 1*MTU

		a.ssthresh = max32(a.cwnd/2, 4*a.mtu)
		a.cwnd = a.mtu
		a.log.Tracef("[%s] updated cwnd=%d ssthresh=%d inflight=%d (RTO)",
			a.name, a.cwnd, a.ssthresh, a.inflightQueue.getNumBytes())

		a.log.Debugf("[%s] T3-rtx timed out: nRtos=%d cwnd=%d ssthresh=%d", a.name, nRtos, a.cwnd, a.ssthresh)

		a.inflightQueue.markAllToRetrasmit()
		a.awakeWriteLoop()

		return
	}

	if id == timerReconfig {
		a.willRetransmitReconfig = true
		a.awakeWriteLoop()
	}
}

func (a *Association) onRetransmissionFailure(id int) {
	a.lock.Lock()
	defer a.lock.Unlock()

	if id == timerT1Init {
		a.log.Errorf("[%s] retransmission failure: T1-init", a.name)
		a.handshakeCompletedCh <- errors.Errorf("handshake failed (INIT ACK)")
		return
	}

	if id == timerT1Cookie {
		a.log.Errorf("[%s] retransmission failure: T1-cookie", a.name)
		a.handshakeCompletedCh <- errors.Errorf("handshake failed (COOKIE ECHO)")
		return
	}

	if id == timerT3RTX {
		// T3-rtx timer will not fail by design
		// Justifications:
		//  * ICE would fail if the connectivity is lost
		//  * WebRTC spec is not clear how this incident should be reported to ULP
		a.log.Errorf("[%s] retransmission failure: T3-rtx (DATA)", a.name)
		return
	}
}

func (a *Association) onAckTimeout() {
	a.lock.Lock()
	defer a.lock.Unlock()

	a.log.Tracef("[%s] ack timed out (ackState: %d)", a.name, a.ackState)
	a.stats.incAckTimeouts()

	a.ackState = ackStateImmediate
	a.awakeWriteLoop()
}

// bufferedAmount returns total amount (in bytes) of currently buffered user data.
// This is used only by testing.
func (a *Association) bufferedAmount() int {
	a.lock.RLock()
	defer a.lock.RUnlock()

	return a.pendingQueue.getNumBytes() + a.inflightQueue.getNumBytes()
}