Initial UDT implementation for file transfer.

New function RequestFileTransferUDT to download files from other peers.
This commit is contained in:
Kleissner
2021-10-24 04:26:30 +02:00
parent 6c2d526251
commit 75e3d9a42e
50 changed files with 4679 additions and 2 deletions

View File

@@ -12,6 +12,7 @@ import (
"github.com/PeernetOfficial/core/dht"
"github.com/PeernetOfficial/core/protocol"
"github.com/PeernetOfficial/core/warehouse"
)
// respondClosesContactsCount is the number of closest contact to respond.
@@ -234,3 +235,47 @@ func SendChatAll(text string) {
peer.Chat(text)
}
}
// cmdTransfer handles an incoming transfer message
func (peer *PeerInfo) cmdTransfer(msg *protocol.MessageTransfer, connection *Connection) {
// Only UDT protocol is currently supported for file transfer.
if msg.TransferProtocol != 0 {
return
}
switch msg.Control {
case protocol.TransferControlRequestStart:
// First check if the file available in the warehouse.
if _, fileInfo, status, _ := UserWarehouse.FileExists(msg.Hash); status != warehouse.StatusOK {
// File not available.
peer.sendTransfer(nil, protocol.TransferControlNotAvailable, msg.TransferProtocol, msg.Hash, 0, 0, msg.Sequence)
return
} else if msg.Limit > 0 && fileInfo.Size() < int64(msg.Offset)+int64(msg.Limit) {
// If the read limit is out of bounds, this request is considered invalid and silently discarded.
return
}
// Create a local UDT client to connect to the remote UDT server and serve the file!
go peer.startFileTransferUDT(msg.Hash, msg.Offset, msg.Limit, msg.Sequence)
case protocol.TransferControlActive:
if v, ok := msg.SequenceInfo.Data.(*virtualPacketConn); ok {
v.receiveData(msg.Data)
return
}
case protocol.TransferControlNotAvailable:
if v, ok := msg.SequenceInfo.Data.(*virtualPacketConn); ok {
v.Terminate(false, 404)
return
}
case protocol.TransferControlTerminate:
if v, ok := msg.SequenceInfo.Data.(*virtualPacketConn); ok {
// Since an incoming terminate notice means the remote peer already terminated the connection, set sendNotice to false.
v.Terminate(false, 2)
return
}
}
}

View File

@@ -186,7 +186,8 @@ func (nets *Networks) packetWorker() {
case protocol.CommandResponse: // Response
if response, _ := protocol.DecodeResponse(raw); response != nil {
// Validate sequence number which prevents unsolicited responses.
sequenceInfo, valid, rtt := nets.Sequences.ValidateSequence(raw.SenderPublicKey, raw.Sequence, response.Actions&(1<<protocol.ActionSequenceLast) > 0, true)
isLast := response.IsLast()
sequenceInfo, valid, rtt := nets.Sequences.ValidateSequence(raw.SenderPublicKey, raw.Sequence, isLast, !isLast)
if !valid {
//Filters.LogError("packetWorker", "message with invalid sequence %d command %d from %s\n", raw.Sequence, raw.Command, raw.connection.Address.String()) // Only log for debug purposes.
continue
@@ -257,6 +258,22 @@ func (nets *Networks) packetWorker() {
}
}
case protocol.CommandTransfer:
if msg, _ := protocol.DecodeTransfer(raw); msg != nil {
// Validate sequence number which prevents unsolicited responses.
isLast := msg.IsLast()
sequenceInfo, valid, rtt := nets.Sequences.ValidateSequenceBi(raw.SenderPublicKey, raw.Sequence, isLast)
if msg.Control != protocol.TransferControlRequestStart && !valid {
//Filters.LogError("packetWorker", "message with invalid sequence %d command %d from %s\n", raw.Sequence, raw.Command, raw.connection.Address.String()) // Only log for debug purposes.
continue
} else if rtt > 0 {
connection.RoundTripTime = rtt
}
raw.SequenceInfo = sequenceInfo
peer.cmdTransfer(msg, connection)
}
default: // Unknown command
Filters.MessageIn(peer, raw, nil)

74
Transfer UDT.go Normal file
View File

@@ -0,0 +1,74 @@
/*
File Name: Transfer UDT.go
Copyright: 2021 Peernet s.r.o.
Author: Peter Kleissner
The strategy is to create a virtual net.PacketConn which can be used by the UDT package for input/output.
TODO: Add timeouts for listening and sending.
*/
package core
import (
"errors"
"net"
"time"
"github.com/PeernetOfficial/core/protocol"
"github.com/PeernetOfficial/core/udt"
)
// transferSequenceTimeout is the timeout for a follow-up message to appear, otherwise the transfer will be terminated.
var transferSequenceTimeout = time.Minute * 10
// startFileTransferUDT starts a file transfer to a remote peer.
// It creates a virtual UDT client to transfer data to a remote peer. Counterintuitively, this will be the "file server" peer.
func (peer *PeerInfo) startFileTransferUDT(hash []byte, offset, limit uint64, sequenceNumber uint32) (err error) {
virtualConnection := newVirtualPacketConn(peer, 0, hash, offset, limit, false)
// register the sequence since packets are sent bi-directional
virtualConnection.sequenceNumber = sequenceNumber
networks.Sequences.RegisterSequenceBi(peer.PublicKey, sequenceNumber, virtualConnection, transferSequenceTimeout, virtualConnection.sequenceTerminate)
// start UDT sender
udtConn, err := udt.DialUDT(udt.DefaultConfig(), virtualConnection, true)
if err != nil {
return err
}
_, err = UserWarehouse.ReadFile(hash, int64(offset), int64(limit), udtConn)
// close the UDT client and virtual connection in any case
udtConn.Close() // warning: This is currently blocking.
//virtualConnection.Terminate(false, 1)
return err
}
// RequestFileTransferUDT creates a UDT server listening for incoming data transfer and requests a file transfer from a remote peer.
func (peer *PeerInfo) RequestFileTransferUDT(hash []byte, offset, limit uint64) (udtConn net.Conn, udtListener net.Listener, err error) {
virtualConnection := newVirtualPacketConn(peer, 0, hash, offset, limit, true)
// new sequence
sequence := networks.Sequences.NewSequenceBi(peer.PublicKey, &peer.messageSequence, virtualConnection, transferSequenceTimeout, virtualConnection.sequenceTerminate)
if sequence == nil {
return nil, nil, errors.New("cannot acquire sequence")
}
virtualConnection.sequenceNumber = sequence.SequenceNumber
// start UDT receiver
udtListener = udt.ListenUDT(udt.DefaultConfig(), virtualConnection)
// request file transfer
peer.sendTransfer(nil, protocol.TransferControlRequestStart, virtualConnection.transferProtocol, hash, offset, limit, virtualConnection.sequenceNumber)
// accept the connection
udtConn, err = udtListener.Accept()
if err != nil {
udtListener.Close()
return nil, nil, err
}
return udtConn, udtListener, nil
}

View File

@@ -143,7 +143,7 @@ func (v *virtualPacketConn) sequenceTerminate() {
// Close closes the connection.
func (v *virtualPacketConn) Close() (err error) {
return v.Terminate(true, 1)
return v.Terminate(false, 1)
}
// WriteTo writes a packet with payload p to addr. WriteTo can be made to time out and return an Error after a fixed time limit.

123
udt/FloydRivest.go Normal file
View File

@@ -0,0 +1,123 @@
package udt
import (
"math"
"sort"
)
// fork from github.com/furstenheim/nth_element/FloydRivest
// FloydRivestBuckets. Sort a slice into buckets of given size. All elements from one bucket are smaller than any element from the next one.
// elements at position i * bucketSize are guaranteed to be the (i * bucketSize) th smallest elements
// s := // some slice
// FloydRivest.Buckets(sort.Interface(s), 5)
// s is now sorted into buckets of size 5
// max(s[0:5]) < min(s[5:10])
// max(s[10: 15]) < min(s[15:20])
// ...
func FloydRivestBuckets(slice sort.Interface, bucketSize int) {
left := 0
right := slice.Len() - 1
s := floydRivestStack([]int{left, right})
var mid int
for len(s) > 0 {
s, right = s.Pop()
s, left = s.Pop()
if right-left <= bucketSize {
continue
}
// + bucketSize - 1 is to do math ceil
mid = left + ((right-left+bucketSize-1)/bucketSize/2)*bucketSize
FloydRivestSelect(slice, mid, left, right)
s = s.Push(left)
s = s.Push(mid)
s = s.Push(mid)
s = s.Push(right)
}
}
// left is the left index for the interval
// right is the right index for the interval
// k is the desired index value, where array[k] is the k+1 smallest element
// when left = 0
func FloydRivestSelect(array sort.Interface, k, left, right int) {
length := array.Len()
for right > left {
if right-left > 600 {
var n = float64(right - left + 1)
var kf = float64(k)
var m = float64(k - left + 1)
var z = math.Log(n)
var s = 0.5 * math.Exp(2*z/3)
sign := float64(1)
if m-n/2 < 0 {
sign = -1
}
var sd = 0.5 * math.Sqrt(z*s*(n-s)/n) * sign
var newLeft = intMax(left, int(math.Floor(kf-m*s/n+sd)))
var newRight = intMin(right, int(math.Floor(kf+(n-m)*s/n+sd)))
FloydRivestSelect(array, k, newLeft, newRight)
}
var i = left
var j = right
array.Swap(left, k)
// in the original algorithm array[k] is stored to a value. To use golangs sort interface we need to keep track of the changes for the index
// we define it as right because in the first iteration of for i<j it will be changed
pointIndex := right
if array.Less(left, right) {
array.Swap(left, right)
pointIndex = left
}
for i < j {
// pointIndex is swapped only once in the first iteration. Later it will either be bigger (if left) or smaller (if right)
array.Swap(i, j)
i++
j--
for i < length && array.Less(i, pointIndex) {
i++
}
for j >= 0 && array.Less(pointIndex, j) {
j--
}
}
// All equal points
if !array.Less(left, pointIndex) && !array.Less(pointIndex, left) {
array.Swap(left, j)
} else {
j++
array.Swap(j, right)
}
if j <= k {
left = j + 1
}
if k <= j {
right = j - 1
}
}
}
func intMin(a, b int) int {
if a < b {
return a
}
return b
}
func intMax(a, b int) int {
if a > b {
return a
}
return b
}
type floydRivestStack []int
func (s floydRivestStack) Push(v int) floydRivestStack {
return append(s, v)
}
func (s floydRivestStack) Pop() (floydRivestStack, int) {
l := len(s)
return s[:l-1], s[l-1]
}

97
udt/acceptsock_heap.go Normal file
View File

@@ -0,0 +1,97 @@
package udt
import (
"container/heap"
"time"
"github.com/PeernetOfficial/core/udt/packet"
)
type acceptSockInfo struct {
sockID uint32
initSeqNo packet.PacketID
lastTouch time.Time
sock *udtSocket
}
// acceptSockHeap defines a list of acceptSockInfo records sorted by their peer socketID and initial sequence number
type acceptSockHeap []acceptSockInfo
func (h acceptSockHeap) Len() int {
return len(h)
}
func (h acceptSockHeap) Less(i, j int) bool {
if h[i].sockID != h[j].sockID {
return h[i].sockID < h[j].sockID
}
return h[i].initSeqNo.Seq < h[j].initSeqNo.Seq
}
func (h acceptSockHeap) Swap(i, j int) {
h[i], h[j] = h[j], h[i]
}
func (h *acceptSockHeap) Push(x interface{}) { // Push and Pop use pointer receivers because they modify the slice's length, not just its contents.
*h = append(*h, x.(acceptSockInfo))
}
func (h *acceptSockHeap) Pop() interface{} {
old := *h
n := len(old)
x := old[n-1]
*h = old[0 : n-1]
return x
}
func (h acceptSockHeap) compare(sockID uint32, initSeqNo packet.PacketID, idx int) int {
if sockID < h[idx].sockID {
return -1
}
if sockID > h[idx].sockID {
return +1
}
if initSeqNo.Seq < h[idx].initSeqNo.Seq {
return -1
}
if initSeqNo.Seq > h[idx].initSeqNo.Seq {
return +1
}
return 0
}
// Find does a binary search of the heap for the specified packetID which is returned
func (h acceptSockHeap) Find(sockID uint32, initSeqNo packet.PacketID) (*udtSocket, int) {
len := len(h)
idx := 0
for idx < len {
cmp := h.compare(sockID, initSeqNo, idx)
if cmp == 0 {
return h[idx].sock, idx
} else if cmp > 0 {
idx = idx * 2
} else {
idx = idx*2 + 1
}
}
return nil, -1
}
// Prune removes any entries that have a lastTouched before the specified time
func (h *acceptSockHeap) Prune(pruneBefore time.Time) {
for {
l := len(*h)
foundOne := false
for idx := 0; idx < l; idx++ {
if (*h)[idx].lastTouch.Before(pruneBefore) {
foundOne = true
heap.Remove(h, idx)
break
}
}
if !foundOne {
// nothing left to prune
return
}
}
}

57
udt/ack_history_heap.go Normal file
View File

@@ -0,0 +1,57 @@
package udt
import (
"time"
"github.com/PeernetOfficial/core/udt/packet"
)
type ackHistoryEntry struct {
ackID uint32
lastPacket packet.PacketID
sendTime time.Time
}
// receiveLossList defines a list of ACK records sorted by their ACK id
type ackHistoryHeap []*ackHistoryEntry
func (h ackHistoryHeap) Len() int {
return len(h)
}
func (h ackHistoryHeap) Less(i, j int) bool {
return h[i].ackID < h[j].ackID
}
func (h ackHistoryHeap) Swap(i, j int) {
h[i], h[j] = h[j], h[i]
}
func (h *ackHistoryHeap) Push(x interface{}) { // Push and Pop use pointer receivers because they modify the slice's length, not just its contents.
*h = append(*h, x.(*ackHistoryEntry))
}
func (h *ackHistoryHeap) Pop() interface{} {
old := *h
n := len(old)
x := old[n-1]
*h = old[0 : n-1]
return x
}
// Find does a binary search of the heap for the specified ackID which is returned
func (h ackHistoryHeap) Find(ackID uint32) (*ackHistoryEntry, int) {
len := len(h)
idx := 0
for idx < len {
here := h[idx].ackID
if here == ackID {
return h[idx], idx
} else if here > ackID {
idx = idx * 2
} else {
idx = idx*2 + 1
}
}
return nil, -1
}

18
udt/atomic_duration.go Normal file
View File

@@ -0,0 +1,18 @@
package udt
import (
"sync/atomic"
"time"
)
type atomicDuration struct {
val int64
}
func (s *atomicDuration) get() time.Duration {
return time.Duration(atomic.LoadInt64(&s.val))
}
func (s *atomicDuration) set(v time.Duration) {
atomic.StoreInt64(&s.val, int64(v))
}

17
udt/atomic_uint32.go Normal file
View File

@@ -0,0 +1,17 @@
package udt
import (
"sync/atomic"
)
type atomicUint32 struct {
val uint32
}
func (s *atomicUint32) get() uint32 {
return atomic.LoadUint32(&s.val)
}
func (s *atomicUint32) set(v uint32) {
atomic.StoreUint32(&s.val, v)
}

39
udt/config.go Normal file
View File

@@ -0,0 +1,39 @@
package udt
import (
"time"
"github.com/PeernetOfficial/core/udt/packet"
)
// Config controls behavior of sockets created with it
type Config struct {
CanAcceptDgram bool // can this listener accept datagrams?
CanAcceptStream bool // can this listener accept streams?
ListenReplayWindow time.Duration // length of time to wait for repeated incoming connections
MaxPacketSize uint // Upper limit on maximum packet size (0 = unlimited)
MaxBandwidth uint64 // Maximum bandwidth to take with this connection (in bytes/sec, 0 = unlimited)
LingerTime time.Duration // time to wait for retransmit requests after connection shutdown
MaxFlowWinSize uint // maximum number of unacknowledged packets to permit (minimum 32)
MTU uint // MTU is the maximum UDT packet size
SynTime time.Duration // SynTime
CanAccept func(hsPacket *packet.HandshakePacket) error // can this listener accept this connection?
CongestionForSocket func(sock *udtSocket) CongestionControl // create or otherwise return the CongestionControl for this socket
}
// DefaultConfig constructs a Config with default values
func DefaultConfig() *Config {
return &Config{
CanAcceptDgram: true,
CanAcceptStream: true,
ListenReplayWindow: 5 * time.Minute,
LingerTime: 10 * time.Second,
MaxFlowWinSize: 64,
MTU: 65535,
SynTime: 10000 * time.Microsecond,
CongestionForSocket: func(sock *udtSocket) CongestionControl {
return &NativeCongestionControl{}
},
}
}

73
udt/congestion.go Normal file
View File

@@ -0,0 +1,73 @@
package udt
import (
"time"
"github.com/PeernetOfficial/core/udt/packet"
)
// CongestionControlParms permits a CongestionControl implementation to interface with the UDT socket
type CongestionControlParms interface {
// GetSndCurrSeqNo is the most recently sent packet ID
GetSndCurrSeqNo() packet.PacketID
// SetCongestionWindowSize sets the size of the congestion window (in packets)
SetCongestionWindowSize(uint)
// GetCongestionWindowSize gets the size of the congestion window (in packets)
GetCongestionWindowSize() uint
// GetPacketSendPeriod gets the current delay between sending packets
GetPacketSendPeriod() time.Duration
// SetPacketSendPeriod sets the current delay between sending packets
SetPacketSendPeriod(time.Duration)
// GetMaxFlowWindow is the largest number of unacknowledged packets we can receive (in packets)
GetMaxFlowWindow() uint
// GetReceiveRates is the current calculated receive rate and bandwidth (in packets/sec)
GetReceiveRates() (recvSpeed, bandwidth uint)
// GetRTT is the current calculated roundtrip time between peers
GetRTT() time.Duration
// GetMSS is the largest packet size we can currently send (in bytes)
GetMSS() uint
// SetACKPerid sets the time between ACKs sent to the peer
SetACKPeriod(time.Duration)
// SetACKInterval sets the number of packets sent to the peer before sending an ACK
SetACKInterval(uint)
// SetRTOPeriod overrides the default EXP timeout calculations waiting for data from the peer
SetRTOPeriod(time.Duration)
}
// CongestionControl controls how timing is handled and UDT connections tuned
type CongestionControl interface {
// Init to be called (only) at the start of a UDT connection.
Init(CongestionControlParms, time.Duration)
// Close to be called when a UDT connection is closed.
Close(CongestionControlParms)
// OnACK to be called when an ACK packet is received
OnACK(CongestionControlParms, packet.PacketID)
// OnNAK to be called when a loss report is received
OnNAK(CongestionControlParms, []packet.PacketID)
// OnTimeout to be called when a timeout event occurs
OnTimeout(CongestionControlParms)
// OnPktSent to be called when data is sent
OnPktSent(CongestionControlParms, packet.Packet)
// OnPktRecv to be called when data is received
OnPktRecv(CongestionControlParms, packet.DataPacket)
// OnCustomMsg to process a user-defined packet
OnCustomMsg(CongestionControlParms, packet.UserDefControlPacket)
}

228
udt/congestion_native.go Normal file
View File

@@ -0,0 +1,228 @@
package udt
import (
"math"
"math/rand"
"time"
"github.com/PeernetOfficial/core/udt/packet"
)
// NativeCongestionControl implements the default congestion control logic for UDP
type NativeCongestionControl struct {
rcInterval time.Duration // UDT Rate control interval
lastRCTime time.Time // last rate increase time
slowStart bool // if in slow start phase
lastAck packet.PacketID // last ACKed seq no
loss bool // if loss happened since last rate increase
lastDecSeq packet.PacketID // biggest sequence number when last time the packet sending rate is decreased
lastDecPeriod time.Duration // value of PacketSendPeriod when last decrease happened
nakCount int // current number of NAKs in the current period
decRandom int // random threshold on decrease by number of loss events
avgNAKNum int // average number of NAKs in a congestion period
decCount int // number of decreases in a congestion epoch
}
// Init to be called (only) at the start of a UDT connection.
func (ncc NativeCongestionControl) Init(parms CongestionControlParms, synTime time.Duration) {
ncc.rcInterval = synTime
ncc.lastRCTime = time.Now()
parms.SetACKPeriod(ncc.rcInterval)
ncc.slowStart = true
ncc.lastAck = parms.GetSndCurrSeqNo()
ncc.loss = false
ncc.lastDecSeq = ncc.lastAck.Add(-1)
ncc.lastDecPeriod = 1 * time.Microsecond
ncc.avgNAKNum = 0
ncc.nakCount = 0
ncc.decRandom = 1
parms.SetCongestionWindowSize(16)
parms.SetPacketSendPeriod(1 * time.Microsecond)
}
// Close to be called when a UDT connection is closed.
func (ncc NativeCongestionControl) Close(parms CongestionControlParms) {
// nothing done for this event
}
// OnACK to be called when an ACK packet is received
func (ncc NativeCongestionControl) OnACK(parms CongestionControlParms, ack packet.PacketID) {
currTime := time.Now()
if currTime.Sub(ncc.lastRCTime) < ncc.rcInterval {
return
}
ncc.lastRCTime = currTime
cWndSize := parms.GetCongestionWindowSize()
pktSendPeriod := parms.GetPacketSendPeriod()
recvRate, bandwidth := parms.GetReceiveRates()
rtt := parms.GetRTT()
// If the current status is in the slow start phase, set the congestion window
// size to the product of packet arrival rate and (RTT + SYN). Slow Start ends. Stop.
if ncc.slowStart {
cWndSize = uint(int(cWndSize) + int(ack.BlindDiff(ncc.lastAck)))
ncc.lastAck = ack
if cWndSize > parms.GetMaxFlowWindow() {
ncc.slowStart = false
if recvRate > 0 {
parms.SetPacketSendPeriod(time.Second / time.Duration(recvRate))
} else {
parms.SetPacketSendPeriod((rtt + ncc.rcInterval) / time.Duration(cWndSize))
}
} else {
// During Slow Start, no rate increase
parms.SetCongestionWindowSize(cWndSize)
return
}
} else {
// Set the congestion window size (CWND) to: CWND = A * (RTT + SYN) + 16.
cWndSize = uint((float64(recvRate)/float64(time.Second))*float64(rtt+ncc.rcInterval) + 16)
}
if ncc.loss {
ncc.loss = false
parms.SetCongestionWindowSize(cWndSize)
return
}
/*
The number of sent packets to be increased in the next SYN period
(inc) is calculated as:
if (B <= C)
inc = 1/PS;
else
inc = max(10^(ceil(log10((B-C)*PS*8))) * Beta/PS, 1/PS);
where B is the estimated link capacity and C is the current
sending speed. All are counted as packets per second. PS is the
fixed size of UDT packet counted in bytes. Beta is a constant
value of 0.0000015.
*/
// Note: 1/24/2012
// The minimum increase parameter is increased from "1.0 / m_iMSS" to 0.01
// because the original was too small and caused sending rate to stay at low level
// for long time.
var inc float64
const minInc float64 = 0.01
B := time.Duration(bandwidth) - time.Second/time.Duration(pktSendPeriod)
bandwidth9 := time.Duration(bandwidth / 9)
if (pktSendPeriod > ncc.lastDecPeriod) && (bandwidth9 < B) {
B = bandwidth9
}
if B <= 0 {
inc = minInc
} else {
// inc = max(10 ^ ceil(log10( B * MSS * 8 ) * Beta / MSS, 1/MSS)
// Beta = 1.5 * 10^(-6)
mss := parms.GetMSS()
inc = math.Pow10(int(math.Ceil(math.Log10(float64(B)*float64(mss)*8.0)))) * 0.0000015 / float64(mss)
if inc < minInc {
inc = minInc
}
}
// The SND period is updated as: SND = (SND * SYN) / (SND * inc + SYN).
parms.SetPacketSendPeriod(time.Duration(float64(pktSendPeriod*ncc.rcInterval) / (float64(pktSendPeriod)*inc + float64(ncc.rcInterval))))
}
// OnNAK to be called when a loss report is received
func (ncc NativeCongestionControl) OnNAK(parms CongestionControlParms, losslist []packet.PacketID) {
// If it is in slow start phase, set inter-packet interval to 1/recvrate. Slow start ends. Stop.
if ncc.slowStart {
ncc.slowStart = false
recvRate, _ := parms.GetReceiveRates()
if recvRate > 0 {
// Set the sending rate to the receiving rate.
parms.SetPacketSendPeriod(time.Second / time.Duration(recvRate))
return
}
// If no receiving rate is observed, we have to compute the sending
// rate according to the current window size, and decrease it
// using the method below.
parms.SetPacketSendPeriod(time.Duration(float64(time.Microsecond) * float64(parms.GetCongestionWindowSize()) / float64(parms.GetRTT()+ncc.rcInterval)))
}
ncc.loss = true
/*
2) If this NAK starts a new congestion period, increase inter-packet
interval (snd) to snd = snd * 1.125; Update AvgNAKNum, reset
NAKCount to 1, and compute DecRandom to a random (average
distribution) number between 1 and AvgNAKNum. Update LastDecSeq.
Stop.
3) If DecCount <= 5, and NAKCount == DecCount * DecRandom:
a. Update SND period: SND = SND * 1.125;
b. Increase DecCount by 1;
c. Record the current largest sent sequence number (LastDecSeq).
*/
pktSendPeriod := parms.GetPacketSendPeriod()
if ncc.lastDecSeq.BlindDiff(losslist[0]) > 0 {
ncc.lastDecPeriod = pktSendPeriod
parms.SetPacketSendPeriod(pktSendPeriod * 1125 / 1000)
ncc.avgNAKNum = int(math.Ceil(float64(ncc.avgNAKNum)*0.875 + float64(ncc.nakCount)*0.125))
ncc.nakCount = 1
ncc.decCount = 1
ncc.lastDecSeq = parms.GetSndCurrSeqNo()
// remove global synchronization using randomization
rand := float64(rand.Uint32()) / math.MaxUint32
ncc.decRandom = int(math.Ceil(float64(ncc.avgNAKNum) * rand))
if ncc.decRandom < 1 {
ncc.decRandom = 1
}
} else {
if ncc.decCount < 5 {
ncc.nakCount++
if ncc.nakCount%ncc.decRandom != 0 {
ncc.decCount++
return
}
}
ncc.decCount++
// 0.875^5 = 0.51, rate should not be decreased by more than half within a congestion period
parms.SetPacketSendPeriod(pktSendPeriod * 1125 / 1000)
ncc.lastDecSeq = parms.GetSndCurrSeqNo()
}
}
// OnTimeout to be called when a timeout event occurs
func (ncc NativeCongestionControl) OnTimeout(parms CongestionControlParms) {
if ncc.slowStart {
ncc.slowStart = false
recvRate, _ := parms.GetReceiveRates()
if recvRate > 0 {
parms.SetPacketSendPeriod(time.Second / time.Duration(recvRate))
} else {
parms.SetPacketSendPeriod(time.Duration(float64(time.Microsecond) * float64(parms.GetCongestionWindowSize()) / float64(parms.GetRTT()+ncc.rcInterval)))
}
} else {
/*
pktSendPeriod := parms.GetPacketSendPeriod()
ncc.lastDecPeriod = pktSendPeriod
parms.SetPacketSendPeriod(math.Ceil(pktSendPeriod * 2))
ncc.lastDecSeq = ncc.lastAck
*/
}
}
// OnPktSent to be called when data is sent
func (ncc NativeCongestionControl) OnPktSent(parms CongestionControlParms, pkt packet.Packet) {
// nothing done for this event
}
// OnPktRecv to be called when a data is received
func (ncc NativeCongestionControl) OnPktRecv(parms CongestionControlParms, pkt packet.DataPacket) {
// nothing done for this event
}
// OnCustomMsg to process a user-defined packet
func (ncc NativeCongestionControl) OnCustomMsg(parms CongestionControlParms, pkt packet.UserDefControlPacket) {
// nothing done for this event
}

104
udt/datapacket_heap.go Normal file
View File

@@ -0,0 +1,104 @@
package udt
import (
"container/heap"
"github.com/PeernetOfficial/core/udt/packet"
)
// receiveLossList defines a list of recvLossEntry records sorted by their packet ID
type dataPacketHeap []*packet.DataPacket
func (h dataPacketHeap) Len() int {
return len(h)
}
func (h dataPacketHeap) Less(i, j int) bool {
return h[i].Seq.Seq < h[j].Seq.Seq
}
func (h dataPacketHeap) Swap(i, j int) {
h[i], h[j] = h[j], h[i]
}
func (h *dataPacketHeap) Push(x interface{}) { // Push and Pop use pointer receivers because they modify the slice's length, not just its contents.
*h = append(*h, x.(*packet.DataPacket))
}
func (h *dataPacketHeap) Pop() interface{} {
old := *h
n := len(old)
x := old[n-1]
*h = old[0 : n-1]
return x
}
// Find does a binary search of the heap for the specified packetID which is returned
func (h dataPacketHeap) Find(packetID packet.PacketID) (*packet.DataPacket, int) {
len := len(h)
idx := 0
for idx < len {
pid := h[idx].Seq
if pid == packetID {
return h[idx], idx
} else if pid.Seq > packetID.Seq {
idx = idx * 2
} else {
idx = idx*2 + 1
}
}
return nil, -1
}
// Min does a binary search of the heap for the entry with the lowest packetID greater than or equal to the specified value
func (h dataPacketHeap) Min(greaterEqual packet.PacketID, lessEqual packet.PacketID) (*packet.DataPacket, int) {
len := len(h)
idx := 0
wrapped := greaterEqual.Seq > lessEqual.Seq
for idx < len {
pid := h[idx].Seq
var next int
if pid.Seq == greaterEqual.Seq {
return h[idx], idx
} else if pid.Seq >= greaterEqual.Seq {
next = idx * 2
} else {
next = idx*2 + 1
}
if next >= len && h[idx].Seq.Seq > greaterEqual.Seq && (wrapped || h[idx].Seq.Seq <= lessEqual.Seq) {
return h[idx], idx
}
idx = next
}
// can't find any packets with greater value, wrap around
if wrapped {
idx = 0
for {
next := idx * 2
if next >= len && h[idx].Seq.Seq <= lessEqual.Seq {
return h[idx], idx
}
idx = next
}
}
return nil, -1
}
// Remove does a binary search of the heap for the specified packetID, which is removed
func (h *dataPacketHeap) Remove(packetID packet.PacketID) bool {
len := len(*h)
idx := 0
for idx < len {
pid := (*h)[idx].Seq
if pid.Seq == packetID.Seq {
heap.Remove(h, idx)
return true
} else if pid.Seq > packetID.Seq {
idx = idx * 2
} else {
idx = idx*2 + 1
}
}
return false
}

17
udt/duration_array.go Normal file
View File

@@ -0,0 +1,17 @@
package udt
import "time"
type sortableDurnArray []time.Duration
func (a sortableDurnArray) Len() int {
return len(a)
}
func (a sortableDurnArray) Less(i, j int) bool {
return a[i] < a[j]
}
func (a sortableDurnArray) Swap(i, j int) {
a[i], a[j] = a[j], a[i]
}

178
udt/listener.go Normal file
View File

@@ -0,0 +1,178 @@
package udt
import (
"container/heap"
"encoding/binary"
"errors"
"fmt"
"net"
"sync"
"time"
"github.com/PeernetOfficial/core/udt/packet"
)
var (
endianness = binary.BigEndian
)
/*
Listener implements the io.Listener interface for UDT.
*/
type listener struct {
m *multiplexer
accept chan *udtSocket
closed chan struct{}
acceptHist acceptSockHeap
acceptHistProt sync.Mutex
config *Config
}
func (l *listener) Accept() (net.Conn, error) {
socket, ok := <-l.accept
if ok {
return socket, nil
}
return nil, errors.New("Listener closed")
}
func (l *listener) Close() (err error) {
a := l.accept
c := l.closed
l.accept = nil
l.closed = nil
if a == nil || c == nil {
return errors.New("Listener closed")
}
close(a)
close(c)
l.m.unlistenUDT(l)
return nil
}
func (l *listener) Addr() net.Addr {
//return l.m.laddr
return nil
}
// checkValidHandshake checks to see if we want to accept a new connection with this handshake.
func (l *listener) checkValidHandshake(m *multiplexer, p *packet.HandshakePacket) bool {
return true
}
func (l *listener) rejectHandshake(m *multiplexer, hsPacket *packet.HandshakePacket) {
fmt.Printf("(listener) sending handshake(reject) (id=%d)\n", hsPacket.SockID)
m.sendPacket(hsPacket.SockID, 0, &packet.HandshakePacket{
UdtVer: hsPacket.UdtVer,
SockType: hsPacket.SockType,
ReqType: packet.HsRefused,
})
}
func (l *listener) readHandshake(m *multiplexer, hsPacket *packet.HandshakePacket) bool {
if hsPacket.ReqType == packet.HsRequest {
fmt.Printf("(listener) sending handshake(request) (id=%d)\n", hsPacket.SockID)
m.sendPacket(hsPacket.SockID, 0, &packet.HandshakePacket{
UdtVer: hsPacket.UdtVer,
SockType: hsPacket.SockType,
InitPktSeq: hsPacket.InitPktSeq,
//MaxPktSize uint32 // maximum packet size (including UDP/IP headers)
//MaxFlowWinSize uint32 // maximum flow window size
ReqType: packet.HsRequest,
// SockID = 0
})
return true
}
// Here used to be a SYNC cookie check. Not needed.
if !l.checkValidHandshake(m, hsPacket) {
l.rejectHandshake(m, hsPacket)
return false
}
now := time.Now()
l.acceptHistProt.Lock()
if l.acceptHist != nil {
replayWindow := l.config.ListenReplayWindow
if replayWindow <= 0 {
replayWindow = DefaultConfig().ListenReplayWindow
}
l.acceptHist.Prune(time.Now().Add(-replayWindow))
s, idx := l.acceptHist.Find(hsPacket.SockID, hsPacket.InitPktSeq)
if s != nil {
l.acceptHist[idx].lastTouch = now
l.acceptHistProt.Unlock()
return s.readHandshake(m, hsPacket)
}
}
l.acceptHistProt.Unlock()
if !l.config.CanAcceptDgram && hsPacket.SockType == packet.TypeDGRAM {
fmt.Printf("Refusing new socket creation from listener requesting DGRAM\n")
l.rejectHandshake(m, hsPacket)
return false
}
if !l.config.CanAcceptStream && hsPacket.SockType == packet.TypeSTREAM {
fmt.Printf("Refusing new socket creation from listener requesting STREAM\n")
l.rejectHandshake(m, hsPacket)
return false
}
if l.config.CanAccept != nil {
err := l.config.CanAccept(hsPacket)
if err != nil {
fmt.Printf("New socket creation from listener rejected by config: %s\n", err.Error())
l.rejectHandshake(m, hsPacket)
return false
}
}
s := l.m.newSocket(l.config, true, hsPacket.SockType == packet.TypeDGRAM)
l.acceptHistProt.Lock()
if l.acceptHist == nil {
l.acceptHist = []acceptSockInfo{{
sockID: hsPacket.SockID,
initSeqNo: hsPacket.InitPktSeq,
lastTouch: now,
sock: s,
}}
heap.Init(&l.acceptHist)
} else {
heap.Push(&l.acceptHist, acceptSockInfo{
sockID: hsPacket.SockID,
initSeqNo: hsPacket.InitPktSeq,
lastTouch: now,
sock: s,
})
}
l.acceptHistProt.Unlock()
if !s.checkValidHandshake(m, hsPacket) {
l.rejectHandshake(m, hsPacket)
return false
}
if !s.readHandshake(m, hsPacket) {
l.rejectHandshake(m, hsPacket)
return false
}
l.accept <- s
return true
}
// ListenUDT listens for incoming UDT connections using the existing provided packet connection. It creates a UDT server.
func ListenUDT(config *Config, packetConn net.PacketConn) net.Listener {
m := newMultiplexer(packetConn, config.MTU)
l := &listener{
m: m,
accept: make(chan *udtSocket, 100),
closed: make(chan struct{}, 1),
config: config,
}
m.listenSock = l
return l
}

138
udt/multiplexer.go Normal file
View File

@@ -0,0 +1,138 @@
// Note: The multiplexer is no longer a multiplexer. Before, it tried send out future UDT traffic over an old (invalidated) PacketConn.
package udt
import (
"fmt"
"math/rand"
"net"
"sync"
"github.com/PeernetOfficial/core/udt/packet"
)
// A multiplexer is a single UDT socket over a single PacketConn.
type multiplexer struct {
conn net.PacketConn // the UDPConn from which we read/write
socket *udtSocket // Socket
socketID uint32 // Socket ID
listenSock *listener // the server socket listening to incoming connections, if there is one. Set by caller.
mtu uint // the Maximum Transmission Unit of packets sent from this address
pktOut chan packet.Packet // packets queued for immediate sending
sync.Mutex // Synchronized access to socket/listenSock
}
func newMultiplexer(conn net.PacketConn, mtu uint) (m *multiplexer) {
m = &multiplexer{
conn: conn,
mtu: mtu, // to be verified?!
pktOut: make(chan packet.Packet, 100), // todo: figure out how to size this
}
go m.goRead()
go m.goWrite()
return
}
// unlistenUDT is the closeListen equivalent
func (m *multiplexer) unlistenUDT(l *listener) {
m.Lock()
defer m.Unlock()
if m.listenSock == nil {
return
}
m.listenSock = nil
m.conn.Close()
close(m.pktOut)
}
func (m *multiplexer) newSocket(config *Config, isServer bool, isDatagram bool) (s *udtSocket) {
m.socketID = rand.Uint32()
m.socket = newSocket(m, config, m.socketID, isServer, isDatagram)
return m.socket
}
func (m *multiplexer) closeSocket(sockID uint32) {
m.Lock()
defer m.Unlock()
if m.socket == nil {
return
}
m.socket = nil
m.conn.Close()
close(m.pktOut)
}
// read runs in a goroutine and reads packets from conn using a buffer from the readBufferPool, or a new buffer.
func (m *multiplexer) goRead() {
buf := make([]byte, m.mtu)
for {
numBytes, _, err := m.conn.ReadFrom(buf)
if err != nil {
return
}
p, err := packet.DecodePacket(buf[0:numBytes])
if err != nil {
fmt.Printf("Unable to decode packet: %s\n", err)
return
}
// attempt to route the packet
sockID := p.SocketID()
if sockID == 0 {
var hsPacket *packet.HandshakePacket
var ok bool
if hsPacket, ok = p.(*packet.HandshakePacket); !ok {
fmt.Printf("Received non-handshake packet with destination socket = 0\n")
return
}
m.Lock()
if m.listenSock != nil {
m.listenSock.readHandshake(m, hsPacket)
}
m.Unlock()
}
if m.socketID == sockID && m.socket != nil {
m.socket.readPacket(m, p)
}
}
}
// write runs in a goroutine and writes packets to conn using a buffer from the writeBufferPool, or a new buffer.
func (m *multiplexer) goWrite() {
buf := make([]byte, m.mtu)
for pkt := range m.pktOut {
plen, err := pkt.WriteTo(buf)
if err != nil {
// TODO: handle write error
fmt.Printf("Unable to buffer out: %s\n", err.Error())
return
}
if _, err = m.conn.WriteTo(buf[0:plen], nil); err != nil {
// TODO: handle write error
fmt.Printf("Unable to write out: %s\n", err.Error())
return
}
}
}
func (m *multiplexer) sendPacket(destSockID uint32, ts uint32, p packet.Packet) {
p.SetHeader(destSockID, ts)
if destSockID == 0 {
if _, ok := p.(*packet.HandshakePacket); !ok {
fmt.Printf("Sending non-handshake packet with destination socket = 0\n")
return
}
}
m.pktOut <- p
}

210
udt/packet/packet.go Normal file
View File

@@ -0,0 +1,210 @@
package packet
// Structure of packets and functions for writing/reading them
import (
"encoding/binary"
"errors"
"fmt"
)
const (
// Leading bit for distinguishing control from data packets
flagBit32 = 1 << 31 // 32 bit
flagBit16 = 1 << 15 // 16 bit
)
// SocketType describes the kind of socket this is (i.e. streaming vs message)
type SocketType uint16
const (
// TypeSTREAM describes a reliable streaming protocol (e.g. TCP)
TypeSTREAM SocketType = 1
// TypeDGRAM describes a partially-reliable messaging protocol
TypeDGRAM SocketType = 2
)
// PacketType describes the type of UDP packet we're dealing with
type PacketType uint16
const (
// Control packet types
ptHandshake PacketType = 0x0
ptKeepalive PacketType = 0x1
ptAck PacketType = 0x2
ptNak PacketType = 0x3
ptCongestion PacketType = 0x4 // unused in ver4
ptShutdown PacketType = 0x5
ptAck2 PacketType = 0x6
ptMsgDropReq PacketType = 0x7
ptSpecialErr PacketType = 0x8 // undocumented but reference implementation seems to use it
ptUserDefPkt PacketType = 0x7FFF
ptData PacketType = 0x8000 // not found in any control packet, but used to identify data packets
)
// PacketTypeName returns a name describing the specified packet type
func PacketTypeName(pt PacketType) string {
switch pt {
case ptHandshake:
return "handshake"
case ptKeepalive:
return "keep-alive"
case ptAck:
return "ack"
case ptNak:
return "nak"
case ptCongestion:
return "congestion"
case ptShutdown:
return "shutdown"
case ptAck2:
return "ack2"
case ptMsgDropReq:
return "msg-drop"
case ptSpecialErr:
return "error"
case ptUserDefPkt:
return "user-defined"
case ptData:
return "data"
default:
return fmt.Sprintf("packet-type-%d", int(pt))
}
}
var (
endianness = binary.BigEndian
)
// Packet represents a UDT packet
type Packet interface {
// socketId retrieves the socket id of a packet
SocketID() (sockID uint32)
// SendTime retrieves the timesamp of the packet
SendTime() (ts uint32)
// WriteTo writes this packet to the provided buffer, returning the length of the packet
WriteTo(buf []byte) (uint, error)
// readFrom reads the packet from a Reader
readFrom(data []byte) (err error)
SetHeader(destSockID uint32, ts uint32)
PacketType() PacketType
}
// ControlPacket represents a UDT control packet.
type ControlPacket interface {
// socketId retrieves the socket id of a packet
SocketID() (sockID uint32)
// SendTime retrieves the timesamp of the packet
SendTime() (ts uint32)
WriteTo(buf []byte) (uint, error)
// readFrom reads the packet from a Reader
readFrom(data []byte) (err error)
SetHeader(destSockID uint32, ts uint32)
PacketType() PacketType
}
type ctrlHeader struct {
ts uint32
DstSockID uint32
}
func (h *ctrlHeader) SocketID() (sockID uint32) {
return h.DstSockID
}
func (h *ctrlHeader) SendTime() (ts uint32) {
return h.ts
}
func (h *ctrlHeader) SetHeader(destSockID uint32, ts uint32) {
h.DstSockID = destSockID
h.ts = ts
}
func (h *ctrlHeader) writeHdrTo(buf []byte, msgType PacketType, info uint32) (uint, error) {
l := len(buf)
if l < 16 {
return 0, errors.New("packet too small")
}
// Sets the flag bit to indicate this is a control packet
endianness.PutUint16(buf[0:2], uint16(msgType)|flagBit16)
endianness.PutUint16(buf[2:4], uint16(0)) // Write 16 bit reserved data
endianness.PutUint32(buf[4:8], info)
endianness.PutUint32(buf[8:12], h.ts)
endianness.PutUint32(buf[12:16], h.DstSockID)
return 16, nil
}
func (h *ctrlHeader) readHdrFrom(data []byte) (addtlInfo uint32, err error) {
l := len(data)
if l < 16 {
return 0, errors.New("packet too small")
}
addtlInfo = endianness.Uint32(data[4:8])
h.ts = endianness.Uint32(data[8:12])
h.DstSockID = endianness.Uint32(data[12:16])
return
}
// DecodePacket takes the contents of a UDP packet and decodes it into a UDT packet
func DecodePacket(data []byte) (p Packet, err error) {
h := endianness.Uint32(data[0:4])
if h&flagBit32 == flagBit32 {
// this is a control packet
// Remove flag bit
h = h &^ flagBit32
// Message type is leading 16 bits
msgType := PacketType(h >> 16)
switch msgType {
case ptHandshake:
p = &HandshakePacket{}
case ptKeepalive:
p = &KeepAlivePacket{}
case ptAck:
if len(data) == 20 {
p = &LightAckPacket{}
} else {
p = &AckPacket{}
}
case ptNak:
p = &NakPacket{}
case ptCongestion:
p = &CongestionPacket{}
case ptShutdown:
p = &ShutdownPacket{}
case ptAck2:
p = &Ack2Packet{}
case ptMsgDropReq:
p = &MsgDropReqPacket{}
case ptSpecialErr:
p = &ErrPacket{}
case ptUserDefPkt:
p = &UserDefControlPacket{msgType: uint16(h & 0xffff)}
default:
return nil, fmt.Errorf("Unknown control packet type: %X", msgType)
}
err = p.readFrom(data)
return
}
// this is a data packet
p = &DataPacket{
Seq: PacketID{h},
}
err = p.readFrom(data)
return
}

77
udt/packet/packet_ack.go Normal file
View File

@@ -0,0 +1,77 @@
package packet
// Structure of packets and functions for writing/reading them
import (
"errors"
)
// AckPacket is a UDT packet acknowledging previously-received data packets and describing the state of the link
type AckPacket struct {
ctrlHeader
AckSeqNo uint32 // ACK sequence number
PktSeqHi PacketID // The packet sequence number to which all the previous packets have been received (excluding)
Rtt uint32 // RTT (in microseconds)
RttVar uint32 // RTT variance
BuffAvail uint32 // Available buffer size (in bytes)
// the following data is optional (not sent more than SYN)
IncludeLink bool
PktRecvRate uint32 // Packets receiving rate (in number of packets per second)
EstLinkCap uint32 // Estimated link capacity (in number of packets per second)
}
// WriteTo writes this packet to the provided buffer, returning the length of the packet
func (p *AckPacket) WriteTo(buf []byte) (uint, error) {
l := len(buf)
if l < 32 {
return 0, errors.New("packet too small")
}
if _, err := p.writeHdrTo(buf, ptAck, p.AckSeqNo); err != nil {
return 0, err
}
endianness.PutUint32(buf[16:20], p.PktSeqHi.Seq)
endianness.PutUint32(buf[20:24], p.Rtt)
endianness.PutUint32(buf[24:28], p.RttVar)
endianness.PutUint32(buf[28:32], p.BuffAvail)
if p.IncludeLink {
if l < 40 {
return 0, errors.New("packet too small")
}
endianness.PutUint32(buf[32:36], p.PktRecvRate)
endianness.PutUint32(buf[36:40], p.EstLinkCap)
return 40, nil
}
return 32, nil
}
func (p *AckPacket) readFrom(data []byte) (err error) {
l := len(data)
if l < 32 {
return errors.New("packet too small")
}
if p.AckSeqNo, err = p.readHdrFrom(data); err != nil {
return err
}
p.PktSeqHi = PacketID{endianness.Uint32(data[16:20])}
p.Rtt = endianness.Uint32(data[20:24])
p.RttVar = endianness.Uint32(data[24:28])
p.BuffAvail = endianness.Uint32(data[28:32])
if l >= 36 {
p.IncludeLink = true
p.PktRecvRate = endianness.Uint32(data[32:36])
if l >= 40 {
p.EstLinkCap = endianness.Uint32(data[36:40])
}
}
return nil
}
// PacketType returns the packetType associated with this packet
func (p *AckPacket) PacketType() PacketType {
return ptAck
}

24
udt/packet/packet_ack2.go Normal file
View File

@@ -0,0 +1,24 @@
package packet
// Structure of packets and functions for writing/reading them
// Ack2Packet is a UDT packet acknowledging receipt of an ACK packet
type Ack2Packet struct {
ctrlHeader
AckSeqNo uint32 // ACK sequence number
}
// WriteTo writes this packet to the provided buffer, returning the length of the packet
func (p *Ack2Packet) WriteTo(buf []byte) (uint, error) {
return p.writeHdrTo(buf, ptAck2, p.AckSeqNo)
}
func (p *Ack2Packet) readFrom(data []byte) (err error) {
p.AckSeqNo, err = p.readHdrFrom(data)
return
}
// PacketType returns the packetType associated with this packet
func (p *Ack2Packet) PacketType() PacketType {
return ptAck2
}

View File

@@ -0,0 +1,13 @@
package packet
import (
"testing"
)
func TestACK2Packet(t *testing.T) {
pkt1 := &Ack2Packet{
AckSeqNo: 90,
}
pkt1.SetHeader(59, 100)
testPacket(pkt1, t)
}

View File

@@ -0,0 +1,31 @@
package packet
import (
"testing"
)
func TestACKPacket(t *testing.T) {
pkt1 := &AckPacket{
AckSeqNo: 90,
PktSeqHi: PacketID{Seq: 91},
Rtt: 92,
RttVar: 93,
BuffAvail: 94,
IncludeLink: true,
PktRecvRate: 95,
EstLinkCap: 96,
}
pkt1.SetHeader(59, 100)
testPacket(pkt1, t)
pkt2 := &AckPacket{
AckSeqNo: 90,
PktSeqHi: PacketID{Seq: 91},
Rtt: 92,
RttVar: 93,
BuffAvail: 94,
IncludeLink: false,
}
pkt2.SetHeader(59, 100)
testPacket(pkt2, t)
}

View File

@@ -0,0 +1,23 @@
package packet
// Structure of packets and functions for writing/reading them
// CongestionPacket is a (deprecated) UDT packet notifying the peer of increased congestion
type CongestionPacket struct {
ctrlHeader
}
// WriteTo writes this packet to the provided buffer, returning the length of the packet
func (p *CongestionPacket) WriteTo(buf []byte) (uint, error) {
return p.writeHdrTo(buf, ptCongestion, 0)
}
func (p *CongestionPacket) readFrom(data []byte) (err error) {
_, err = p.readHdrFrom(data)
return
}
// PacketType returns the packetType associated with this packet
func (p *CongestionPacket) PacketType() PacketType {
return ptCongestion
}

View File

@@ -0,0 +1,11 @@
package packet
import (
"testing"
)
func TestCongestionPacket(t *testing.T) {
pkt1 := &CongestionPacket{}
pkt1.SetHeader(59, 100)
testPacket(pkt1, t)
}

94
udt/packet/packet_data.go Normal file
View File

@@ -0,0 +1,94 @@
package packet
import "errors"
// MessageBoundary flags for where this packet falls within a message
type MessageBoundary uint8
const (
// MbFirst is the first packet in a multi-packet message
MbFirst MessageBoundary = 2
// MbLast is the last packet in a multi-packet message
MbLast MessageBoundary = 1
// MbOnly is the only packet in this message
MbOnly MessageBoundary = 3
// MbMiddle is neither the first nor last packet in a multi-packet message
MbMiddle MessageBoundary = 0
)
// DataPacket is a UDT packet containing message data
type DataPacket struct {
Seq PacketID // packet sequence number (top bit = 0)
msg uint32 // message sequence number (top three bits = message control)
ts uint32 // timestamp when message is sent
DstSockID uint32 // destination socket
Data []byte // payload
}
// PacketType returns the packetType associated with this packet
func (dp *DataPacket) PacketType() PacketType {
return ptData
}
// SetHeader sets the fields common to UDT data packets
func (dp *DataPacket) SetHeader(destSockID uint32, ts uint32) {
dp.DstSockID = destSockID
dp.ts = ts
}
// SocketID sets the Socket ID for this data packet
func (dp *DataPacket) SocketID() (sockID uint32) {
return dp.DstSockID
}
// SendTime sets the timestamp field for this data packet
func (dp *DataPacket) SendTime() (ts uint32) {
return dp.ts
}
// SetMessageData sets the message field for this data packet
func (dp *DataPacket) SetMessageData(boundary MessageBoundary, order bool, msg uint32) {
var iOrder uint32 = 0
if order {
iOrder = 0x20000000
}
dp.msg = (uint32(boundary) << 30) | iOrder | (msg & 0x1FFFFFFF)
}
// GetMessageData returns the message field for this data packet
func (dp *DataPacket) GetMessageData() (MessageBoundary, bool, uint32) {
return MessageBoundary(dp.msg >> 30), (dp.msg & 0x20000000) != 0, dp.msg & 0x1FFFFFFF
}
// WriteTo writes this packet to the provided buffer, returning the length of the packet
func (dp *DataPacket) WriteTo(buf []byte) (uint, error) {
l := len(buf)
ol := 16 + len(dp.Data)
if l < ol {
return 0, errors.New("packet too small")
}
endianness.PutUint32(buf[0:4], dp.Seq.Seq&0x7FFFFFFF)
endianness.PutUint32(buf[4:8], dp.msg)
endianness.PutUint32(buf[8:12], dp.ts)
endianness.PutUint32(buf[12:16], dp.DstSockID)
copy(buf[16:], dp.Data)
return uint(ol), nil
}
func (dp *DataPacket) readFrom(data []byte) (err error) {
l := len(data)
if l < 16 {
return errors.New("packet too small")
}
//dp.seq = endianness.Uint32(data[0:4])
dp.msg = endianness.Uint32(data[4:8])
dp.ts = endianness.Uint32(data[8:12])
dp.DstSockID = endianness.Uint32(data[12:16])
// The data is whatever is what comes after the 16 bytes of header
dp.Data = make([]byte, l-16)
copy(dp.Data, data[16:])
return
}

View File

@@ -0,0 +1,15 @@
package packet
import (
"testing"
)
func TestDataPacket(t *testing.T) {
testPacket(
&DataPacket{
Seq: PacketID{Seq: 50},
ts: 1409,
DstSockID: 90,
Data: []byte("Hello UDT World!"),
}, t)
}

24
udt/packet/packet_err.go Normal file
View File

@@ -0,0 +1,24 @@
package packet
// Structure of packets and functions for writing/reading them
// ErrPacket is a (undocumented) UDT packet describing an out-of-band error code
type ErrPacket struct {
ctrlHeader
Errno uint32 // error code
}
// WriteTo writes this packet to the provided buffer, returning the length of the packet
func (p *ErrPacket) WriteTo(buf []byte) (uint, error) {
return p.writeHdrTo(buf, ptSpecialErr, p.Errno)
}
func (p *ErrPacket) readFrom(data []byte) (err error) {
p.Errno, err = p.readHdrFrom(data)
return
}
// PacketType returns the packetType associated with this packet
func (p *ErrPacket) PacketType() PacketType {
return ptSpecialErr
}

View File

@@ -0,0 +1,13 @@
package packet
import (
"testing"
)
func TestErrPacket(t *testing.T) {
pkt1 := &ErrPacket{
Errno: 90,
}
pkt1.SetHeader(59, 100)
testPacket(pkt1, t)
}

View File

@@ -0,0 +1,90 @@
package packet
// Structure of packets and functions for writing/reading them
import (
"errors"
)
// HandshakeReqType describes the type of handshake packet
type HandshakeReqType int32
const (
// HsRequest represents an attempt to establish a new connection
HsRequest HandshakeReqType = 1
//HsRendezvous represents an attempt to establish a new connection using mutual rendezvous packets
HsRendezvous HandshakeReqType = 0
//HsResponse is a response to a handshake request
HsResponse HandshakeReqType = -1
//HsResponse2 is an acknowledgement that a HsResponse was received
HsResponse2 HandshakeReqType = -2
//HsRefused notifies the peer of a connection refusal
HsRefused HandshakeReqType = 1002
)
// HandshakePacket is a UDT packet used to negotiate a new connection
type HandshakePacket struct {
ctrlHeader
UdtVer uint32 // UDT version
SockType SocketType // Socket Type (1 = STREAM or 2 = DGRAM)
InitPktSeq PacketID // initial packet sequence number
MaxPktSize uint32 // maximum packet size (including UDP/IP headers)
MaxFlowWinSize uint32 // maximum flow window size
ReqType HandshakeReqType // connection type (regular(1), rendezvous(0), -1/-2 response)
SockID uint32 // socket ID
}
// WriteTo writes this packet to the provided buffer, returning the length of the packet
func (p *HandshakePacket) WriteTo(buf []byte) (uint, error) {
l := len(buf)
if l < 64 {
return 0, errors.New("packet too small")
}
if _, err := p.writeHdrTo(buf, ptHandshake, 0); err != nil {
return 0, err
}
endianness.PutUint32(buf[16:20], p.UdtVer)
endianness.PutUint32(buf[20:24], uint32(p.SockType))
endianness.PutUint32(buf[24:28], p.InitPktSeq.Seq)
endianness.PutUint32(buf[28:32], p.MaxPktSize)
endianness.PutUint32(buf[32:36], p.MaxFlowWinSize)
endianness.PutUint32(buf[36:40], uint32(p.ReqType))
endianness.PutUint32(buf[40:44], p.SockID)
//endianness.PutUint32(buf[44:48], p.SynCookie)
//sockAddr := make([]byte, 16)
//copy(sockAddr, p.SockAddr)
//copy(buf[48:64], sockAddr)
return 64, nil
}
func (p *HandshakePacket) readFrom(data []byte) error {
l := len(data)
if l < 64 {
return errors.New("packet too small")
}
if _, err := p.readHdrFrom(data); err != nil {
return err
}
p.UdtVer = endianness.Uint32(data[16:20])
p.SockType = SocketType(endianness.Uint32(data[20:24]))
p.InitPktSeq = PacketID{endianness.Uint32(data[24:28])}
p.MaxPktSize = endianness.Uint32(data[28:32])
p.MaxFlowWinSize = endianness.Uint32(data[32:36])
p.ReqType = HandshakeReqType(endianness.Uint32(data[36:40]))
p.SockID = endianness.Uint32(data[40:44])
//p.SynCookie = endianness.Uint32(data[44:48])
//p.SockAddr = make(net.IP, 16)
//copy(p.SockAddr, data[48:64])
return nil
}
// PacketType returns the packetType associated with this packet
func (p *HandshakePacket) PacketType() PacketType {
return ptHandshake
}

View File

@@ -0,0 +1,23 @@
package packet
// Structure of packets and functions for writing/reading them
// KeepAlivePacket is a UDT packet used to keep a connection alive when no data is being sent
type KeepAlivePacket struct {
ctrlHeader
}
// WriteTo writes this packet to the provided buffer, returning the length of the packet
func (p *KeepAlivePacket) WriteTo(buf []byte) (uint, error) {
return p.writeHdrTo(buf, ptKeepalive, 0)
}
func (p *KeepAlivePacket) readFrom(data []byte) (err error) {
_, err = p.readHdrFrom(data)
return
}
// PacketType returns the packetType associated with this packet
func (p *KeepAlivePacket) PacketType() PacketType {
return ptKeepalive
}

View File

@@ -0,0 +1,11 @@
package packet
import (
"testing"
)
func TestKeepAlivePacket(t *testing.T) {
pkt1 := &KeepAlivePacket{}
pkt1.SetHeader(59, 100)
testPacket(pkt1, t)
}

View File

@@ -0,0 +1,47 @@
package packet
// Structure of packets and functions for writing/reading them
import (
"errors"
)
// LightAckPacket is a UDT variant of the ACK packet for acknowledging received data with minimal information
type LightAckPacket struct {
ctrlHeader
PktSeqHi PacketID // The packet sequence number to which all the previous packets have been received (excluding)
}
// WriteTo writes this packet to the provided buffer, returning the length of the packet
func (p *LightAckPacket) WriteTo(buf []byte) (uint, error) {
l := len(buf)
if l < 20 {
return 0, errors.New("packet too small")
}
if _, err := p.writeHdrTo(buf, ptAck, 0); err != nil {
return 0, err
}
endianness.PutUint32(buf[16:20], p.PktSeqHi.Seq)
return 20, nil
}
func (p *LightAckPacket) readFrom(data []byte) (err error) {
l := len(data)
if l < 20 {
return errors.New("packet too small")
}
if _, err = p.readHdrFrom(data); err != nil {
return err
}
p.PktSeqHi = PacketID{endianness.Uint32(data[16:20])}
return nil
}
// PacketType returns the packetType associated with this packet
func (p *LightAckPacket) PacketType() PacketType {
return ptAck
}

View File

@@ -0,0 +1,13 @@
package packet
import (
"testing"
)
func TestLightAckPacket(t *testing.T) {
pkt1 := &LightAckPacket{
PktSeqHi: PacketID{Seq: 91},
}
pkt1.SetHeader(59, 100)
testPacket(pkt1, t)
}

View File

@@ -0,0 +1,50 @@
package packet
// Structure of packets and functions for writing/reading them
import (
"errors"
)
// MsgDropReqPacket is a UDT packet notifying the peer of expired packets not worth trying to send
type MsgDropReqPacket struct {
ctrlHeader
MsgID uint32 // Message ID
FirstSeq PacketID // First sequence number in the message
LastSeq PacketID // Last sequence number in the message
}
// WriteTo writes this packet to the provided buffer, returning the length of the packet
func (p *MsgDropReqPacket) WriteTo(buf []byte) (uint, error) {
l := len(buf)
if l < 24 {
return 0, errors.New("packet too small")
}
if _, err := p.writeHdrTo(buf, ptMsgDropReq, p.MsgID); err != nil {
return 0, err
}
endianness.PutUint32(buf[16:20], p.FirstSeq.Seq)
endianness.PutUint32(buf[20:24], p.LastSeq.Seq)
return 24, nil
}
func (p *MsgDropReqPacket) readFrom(data []byte) (err error) {
l := len(data)
if l < 24 {
return errors.New("packet too small")
}
if p.MsgID, err = p.readHdrFrom(data); err != nil {
return
}
p.FirstSeq = PacketID{endianness.Uint32(data[16:20])}
p.LastSeq = PacketID{endianness.Uint32(data[20:24])}
return
}
// PacketType returns the packetType associated with this packet
func (p *MsgDropReqPacket) PacketType() PacketType {
return ptMsgDropReq
}

View File

@@ -0,0 +1,15 @@
package packet
import (
"testing"
)
func TestMsgDropReqPacket(t *testing.T) {
pkt1 := &MsgDropReqPacket{
MsgID: 90,
FirstSeq: PacketID{Seq: 91},
LastSeq: PacketID{Seq: 92},
}
pkt1.SetHeader(59, 100)
testPacket(pkt1, t)
}

53
udt/packet/packet_nak.go Normal file
View File

@@ -0,0 +1,53 @@
package packet
// Structure of packets and functions for writing/reading them
import (
"errors"
)
// NakPacket is a UDT packet notifying the peer of lost packets
type NakPacket struct {
ctrlHeader
CmpLossInfo []uint32 // integer array of compressed loss information
}
// WriteTo writes this packet to the provided buffer, returning the length of the packet
func (p *NakPacket) WriteTo(buf []byte) (uint, error) {
off, err := p.writeHdrTo(buf, ptNak, 0)
if err != nil {
return 0, err
}
l := uint(len(buf))
if l < off+uint(4*len(p.CmpLossInfo)) {
return 0, errors.New("packet too small")
}
for _, elm := range p.CmpLossInfo {
endianness.PutUint32(buf[off:off+4], elm)
off = off + 4
}
return off, nil
}
func (p *NakPacket) readFrom(data []byte) error {
if _, err := p.readHdrFrom(data); err != nil {
return err
}
l := len(data)
numEntry := (l - 16) / 4
p.CmpLossInfo = make([]uint32, numEntry)
for idx := range p.CmpLossInfo {
st := 16 + 4*idx
p.CmpLossInfo[idx] = endianness.Uint32(data[st : st+4])
}
return nil
}
// PacketType returns the packetType associated with this packet
func (p *NakPacket) PacketType() PacketType {
return ptNak
}

View File

@@ -0,0 +1,13 @@
package packet
import (
"testing"
)
func TestNAKPacket(t *testing.T) {
pkt1 := &NakPacket{
CmpLossInfo: []uint32{90},
}
pkt1.SetHeader(59, 100)
testPacket(pkt1, t)
}

View File

@@ -0,0 +1,23 @@
package packet
// Structure of packets and functions for writing/reading them
// ShutdownPacket is a UDT packet notifying the peer of connection shutdown
type ShutdownPacket struct {
ctrlHeader
}
// WriteTo writes this packet to the provided buffer, returning the length of the packet
func (p *ShutdownPacket) WriteTo(buf []byte) (uint, error) {
return p.writeHdrTo(buf, ptShutdown, 0)
}
func (p *ShutdownPacket) readFrom(data []byte) (err error) {
_, err = p.readHdrFrom(data)
return
}
// PacketType returns the packetType associated with this packet
func (p *ShutdownPacket) PacketType() PacketType {
return ptShutdown
}

View File

@@ -0,0 +1,11 @@
package packet
import (
"testing"
)
func TestShutdownPacket(t *testing.T) {
pkt1 := &ShutdownPacket{}
pkt1.SetHeader(59, 100)
testPacket(pkt1, t)
}

23
udt/packet/packet_test.go Normal file
View File

@@ -0,0 +1,23 @@
package packet
import (
"reflect"
"testing"
)
func testPacket(p Packet, t *testing.T) (read Packet) {
buf := make([]byte, 1500)
n, err := p.WriteTo(buf)
if err != nil {
t.Errorf("Unable to write packet: %s", err)
}
if p2, err := DecodePacket(buf[0:n]); err != nil {
t.Errorf("Unable to read packet: %s", err)
} else {
if !reflect.DeepEqual(p, p2) {
t.Errorf("Read did not match written.\n\nWrote: %s\nRead: %s", p, p2)
}
read = p2
}
return
}

View File

@@ -0,0 +1,48 @@
package packet
import "errors"
// Structure of packets and functions for writing/reading them
// UserDefControlPacket is a UDT user-defined packet
type UserDefControlPacket struct {
ctrlHeader
msgType uint16 // user-defined message type
addtlInfo uint32
data []byte
}
// WriteTo writes this packet to the provided buffer, returning the length of the packet
func (p *UserDefControlPacket) WriteTo(buf []byte) (uint, error) {
l := len(buf)
ol := 16 + len(p.data)
if l < ol {
return 0, errors.New("packet too small")
}
// Sets the flag bit to indicate this is a control packet
endianness.PutUint16(buf[0:2], uint16(ptUserDefPkt)|flagBit16)
endianness.PutUint16(buf[2:4], p.msgType) // Write 16 bit reserved data
endianness.PutUint32(buf[4:8], p.addtlInfo)
endianness.PutUint32(buf[8:12], p.ts)
endianness.PutUint32(buf[12:16], p.DstSockID)
copy(buf[16:], p.data)
return uint(ol), nil
}
func (p *UserDefControlPacket) readFrom(data []byte) (err error) {
if p.addtlInfo, err = p.readHdrFrom(data); err != nil {
return err
}
p.data = data[16:]
return nil
}
// PacketType returns the packetType associated with this packet
func (p *UserDefControlPacket) PacketType() PacketType {
return ptUserDefPkt
}

31
udt/packet/pktseq.go Normal file
View File

@@ -0,0 +1,31 @@
package packet
// PacketID represents a UDT packet ID sequence
type PacketID struct {
Seq uint32
}
// Incr increments this packet ID
func (p *PacketID) Incr() {
p.Seq = (p.Seq + 1) & 0x7FFFFFFF
}
// Decr decrements this packet ID
func (p *PacketID) Decr() {
p.Seq = (p.Seq - 1) & 0x7FFFFFFF
}
// Add returns a packet ID after adding the specified offset
func (p PacketID) Add(off int32) PacketID {
newSeq := (p.Seq + 1) & 0x7FFFFFFF
return PacketID{newSeq}
}
// BlindDiff attempts to return the difference after subtracting the argument from itself
func (p PacketID) BlindDiff(rhs PacketID) int32 {
result := (p.Seq - rhs.Seq) & 0x7FFFFFFF
if result&0x40000000 != 0 {
result = result | 0x80000000
}
return int32(result)
}

92
udt/packetid_heap.go Normal file
View File

@@ -0,0 +1,92 @@
package udt
import "github.com/PeernetOfficial/core/udt/packet"
// packetIdHeap defines a list of sorted packet IDs
type packetIDHeap []packet.PacketID
func (h packetIDHeap) Len() int {
return len(h)
}
func (h packetIDHeap) Less(i, j int) bool {
return h[i].Seq < h[j].Seq
}
func (h packetIDHeap) Swap(i, j int) {
h[i], h[j] = h[j], h[i]
}
func (h *packetIDHeap) Push(x interface{}) { // Push and Pop use pointer receivers because they modify the slice's length, not just its contents.
*h = append(*h, x.(packet.PacketID))
}
func (h *packetIDHeap) Pop() interface{} {
old := *h
n := len(old)
x := old[n-1]
*h = old[0 : n-1]
return x
}
// Min does a binary search of the heap for the entry with the lowest packetID greater than or equal to the specified value
func (h packetIDHeap) Min(greaterEqual packet.PacketID, lessEqual packet.PacketID) (packet.PacketID, int) {
len := len(h)
idx := 0
wrapped := greaterEqual.Seq > lessEqual.Seq
for idx < len {
pid := h[idx]
var next int
if pid.Seq == greaterEqual.Seq {
return h[idx], idx
} else if pid.Seq >= greaterEqual.Seq {
next = idx * 2
} else {
next = idx*2 + 1
}
if next >= len && h[idx].Seq > greaterEqual.Seq && (wrapped || h[idx].Seq <= lessEqual.Seq) {
return h[idx], idx
}
idx = next
}
// can't find any packets with greater value, wrap around
if wrapped {
idx = 0
for {
next := idx * 2
if next >= len && h[idx].Seq <= lessEqual.Seq {
return h[idx], idx
}
idx = next
}
}
return packet.PacketID{Seq: 0}, -1
}
func (h packetIDHeap) compare(pktID packet.PacketID, idx int) int {
if pktID.Seq < h[idx].Seq {
return -1
}
if pktID.Seq > h[idx].Seq {
return +1
}
return 0
}
// Find does a binary search of the heap for the specified packetID which is returned
func (h packetIDHeap) Find(pktID packet.PacketID) (*packet.PacketID, int) {
len := len(h)
idx := 0
for idx < len {
cmp := h.compare(pktID, idx)
if cmp == 0 {
return &h[idx], idx
} else if cmp > 0 {
idx = idx * 2
} else {
idx = idx*2 + 1
}
}
return nil, -1
}

20
udt/readme.md Normal file
View File

@@ -0,0 +1,20 @@
# UDT: UDP-based Data Transfer Protocol
UDT (UDP-based Data Transfer Protocol) is a transfer protocol on top of UDP. See https://udt.sourceforge.io/ for the original spec and the reference implementation.
This project is a fork from https://github.com/odysseus654/go-udt which itself is a fork.
## Stream vs Datagram
```
// TypeSTREAM describes a reliable streaming protocol (e.g. TCP)
TypeSTREAM SocketType = 1
// TypeDGRAM describes a partially-reliable messaging protocol
TypeDGRAM SocketType = 2
UDT supports both reliable data streaming and partial reliable
messaging. The data streaming semantics is similar to that of TCP,
while the messaging semantics can be regarded as a subset of SCTP
[RFC4960].
```

111
udt/recvloss_heap.go Normal file
View File

@@ -0,0 +1,111 @@
package udt
import (
"container/heap"
"time"
"github.com/PeernetOfficial/core/udt/packet"
)
type recvLossEntry struct {
packetID packet.PacketID
lastFeedback time.Time
numNAK uint
}
// receiveLossList defines a list of recvLossEntry records sorted by their packet ID
type receiveLossHeap []recvLossEntry
func (h receiveLossHeap) Len() int {
return len(h)
}
func (h receiveLossHeap) Less(i, j int) bool {
return h[i].packetID.Seq < h[j].packetID.Seq
}
func (h receiveLossHeap) Swap(i, j int) {
h[i], h[j] = h[j], h[i]
}
func (h *receiveLossHeap) Push(x interface{}) { // Push and Pop use pointer receivers because they modify the slice's length, not just its contents.
*h = append(*h, x.(recvLossEntry))
}
func (h *receiveLossHeap) Pop() interface{} {
old := *h
n := len(old)
x := old[n-1]
*h = old[0 : n-1]
return x
}
// Min does a binary search of the heap for the entry with the lowest packetID greater than or equal to the specified value
func (h receiveLossHeap) Min(greaterEqual packet.PacketID, lessEqual packet.PacketID) (packet.PacketID, int) {
len := len(h)
idx := 0
wrapped := greaterEqual.Seq > lessEqual.Seq
for idx < len {
pid := h[idx].packetID
var next int
if pid.Seq == greaterEqual.Seq {
return h[idx].packetID, idx
} else if pid.Seq >= greaterEqual.Seq {
next = idx * 2
} else {
next = idx*2 + 1
}
if next >= len && h[idx].packetID.Seq > greaterEqual.Seq && (wrapped || h[idx].packetID.Seq <= lessEqual.Seq) {
return h[idx].packetID, idx
}
idx = next
}
// can't find any packets with greater value, wrap around
if wrapped {
idx = 0
for {
next := idx * 2
if next >= len && h[idx].packetID.Seq <= lessEqual.Seq {
return h[idx].packetID, idx
}
idx = next
}
}
return packet.PacketID{Seq: 0}, -1
}
// Find does a binary search of the heap for the specified packetID which is returned
func (h receiveLossHeap) Find(packetID packet.PacketID) (*recvLossEntry, int) {
len := len(h)
idx := 0
for idx < len {
pid := h[idx].packetID
if pid == packetID {
return &h[idx], idx
} else if pid.Seq > packetID.Seq {
idx = idx * 2
} else {
idx = idx*2 + 1
}
}
return nil, -1
}
// Remove does a binary search of the heap for the specified packetID, which is removed
func (h *receiveLossHeap) Remove(packetID packet.PacketID) bool {
len := len(*h)
idx := 0
for idx < len {
pid := (*h)[idx].packetID
if pid == packetID {
heap.Remove(h, idx)
return true
} else if pid.Seq > packetID.Seq {
idx = idx * 2
} else {
idx = idx*2 + 1
}
}
return false
}

111
udt/sendpacket_heap.go Normal file
View File

@@ -0,0 +1,111 @@
package udt
import (
"container/heap"
"time"
"github.com/PeernetOfficial/core/udt/packet"
)
type sendPacketEntry struct {
pkt *packet.DataPacket
tim time.Time
ttl time.Duration
}
// receiveLossList defines a list of recvLossEntry records sorted by their packet ID
type sendPacketHeap []sendPacketEntry
func (h sendPacketHeap) Len() int {
return len(h)
}
func (h sendPacketHeap) Less(i, j int) bool {
return h[i].pkt.Seq.Seq < h[j].pkt.Seq.Seq
}
func (h sendPacketHeap) Swap(i, j int) {
h[i], h[j] = h[j], h[i]
}
func (h *sendPacketHeap) Push(x interface{}) { // Push and Pop use pointer receivers because they modify the slice's length, not just its contents.
*h = append(*h, x.(sendPacketEntry))
}
func (h *sendPacketHeap) Pop() interface{} {
old := *h
n := len(old)
x := old[n-1]
*h = old[0 : n-1]
return x
}
// Find does a binary search of the heap for the specified packetID which is returned
func (h sendPacketHeap) Find(packetID packet.PacketID) (*sendPacketEntry, int) {
len := len(h)
idx := 0
for idx < len {
pid := h[idx].pkt.Seq
if pid == packetID {
return &h[idx], idx
} else if pid.Seq > packetID.Seq {
idx = idx * 2
} else {
idx = idx*2 + 1
}
}
return nil, -1
}
// Min does a binary search of the heap for the entry with the lowest packetID greater than or equal to the specified value
func (h sendPacketHeap) Min(greaterEqual packet.PacketID, lessEqual packet.PacketID) (*packet.DataPacket, int) {
len := len(h)
idx := 0
wrapped := greaterEqual.Seq > lessEqual.Seq
for idx < len {
pid := h[idx].pkt.Seq
var next int
if pid.Seq == greaterEqual.Seq {
return h[idx].pkt, idx
} else if pid.Seq >= greaterEqual.Seq {
next = idx * 2
} else {
next = idx*2 + 1
}
if next >= len && h[idx].pkt.Seq.Seq > greaterEqual.Seq && (wrapped || h[idx].pkt.Seq.Seq <= lessEqual.Seq) {
return h[idx].pkt, idx
}
idx = next
}
// can't find any packets with greater value, wrap around
if wrapped {
idx = 0
for {
next := idx * 2
if next >= len && h[idx].pkt.Seq.Seq <= lessEqual.Seq {
return h[idx].pkt, idx
}
idx = next
}
}
return nil, -1
}
// Remove does a binary search of the heap for the specified packetID, which is removed
func (h *sendPacketHeap) Remove(packetID packet.PacketID) bool {
len := len(*h)
idx := 0
for idx < len {
pid := (*h)[idx].pkt.Seq
if pid.Seq == packetID.Seq {
heap.Remove(h, idx)
return true
} else if pid.Seq > packetID.Seq {
idx = idx * 2
} else {
idx = idx*2 + 1
}
}
return false
}

26
udt/udt.go Normal file
View File

@@ -0,0 +1,26 @@
package udt
/*
Package udt provides a pure Go implementation of the UDT protocol per
http://udt.sourceforge.net/doc/draft-gg-udt-03.txt.
udt does not implement all of the spec. In particular, the following are not
implemented:
- STREAM mode (only UDP is supported)
*/
import (
"net"
)
// DialUDT establishes an outbound UDT connection using the existing provided packet connection. It creates a UDT client.
func DialUDT(config *Config, packetConn net.PacketConn, isStream bool) (net.Conn, error) {
m := newMultiplexer(packetConn, config.MTU)
s := m.newSocket(config, false, !isStream)
err := s.startConnect()
return s, err
}

710
udt/udtsocket.go Normal file
View File

@@ -0,0 +1,710 @@
package udt
import (
"errors"
"fmt"
"math/rand"
"net"
"sync"
"syscall"
"time"
"github.com/PeernetOfficial/core/udt/packet"
)
type sockState int
const (
sockStateInit sockState = iota // object is being constructed
sockStateInvalid // attempting to create a rendezvous connection
sockStateConnecting // attempting to create a connection
sockStateConnected // connection is established
sockStateClosed // connection has been closed (by either end)
sockStateRefused // connection rejected by remote host
sockStateCorrupted // peer behaved in an improper manner
sockStateTimeout // connection failed due to peer timeout
)
type recvPktEvent struct {
pkt packet.Packet
now time.Time
}
type sendMessage struct {
content []byte
tim time.Time // time message is submitted
ttl time.Duration // message dropped if it can't be sent in this timeframe
}
type shutdownMessage struct {
sockState sockState
permitLinger bool
err error
}
/*
udtSocket encapsulates a UDT socket between a local and remote address pair, as
defined by the UDT specification. udtSocket implements the net.Conn interface
so that it can be used anywhere that a stream-oriented network connection
(like TCP) would be used.
*/
type udtSocket struct {
// this data not changed after the socket is initialized and/or handshaked
m *multiplexer // the multiplexer that handles this socket
//raddr *net.UDPAddr // the remote address
created time.Time // the time that this socket was created
Config *Config // configuration parameters for this socket
udtVer int // UDT protcol version (normally 4. Will we be supporting others?)
isDatagram bool // if true then we're sending and receiving datagrams, otherwise we're a streaming socket
isServer bool // if true then we are behaving like a server, otherwise client (or rendezvous). Only useful during handshake
sockID uint32 // our sockID
farSockID uint32 // the peer's sockID
initPktSeq packet.PacketID // initial packet sequence to start the connection with
connectWait *sync.WaitGroup // released when connection is complete (or failed)
sockState sockState // socket state - used mostly during handshakes
mtu atomicUint32 // the negotiated maximum packet size
maxFlowWinSize uint // receiver: maximum unacknowledged packet count
currPartialRead []byte // stream connections: currently reading message (for partial reads). Owned by client caller (Read)
readDeadline *time.Timer // if set, then calls to Read() will return "timeout" after this time
readDeadlinePassed bool // if set, then calls to Read() will return "timeout"
writeDeadline *time.Timer // if set, then calls to Write() will return "timeout" after this time
writeDeadlinePassed bool // if set, then calls to Write() will return "timeout"
rttProt sync.RWMutex // lock must be held before referencing rtt/rttVar
rtt uint // receiver: estimated roundtrip time. (in microseconds)
rttVar uint // receiver: roundtrip variance. (in microseconds)
receiveRateProt sync.RWMutex // lock must be held before referencing deliveryRate/bandwidth
deliveryRate uint // delivery rate reported from peer (packets/sec)
bandwidth uint // bandwidth reported from peer (packets/sec)
// channels
messageIn chan []byte // inbound messages. Sender is goReceiveEvent->ingestData, Receiver is client caller (Read)
messageOut chan sendMessage // outbound messages. Sender is client caller (Write), Receiver is goSendEvent. Closed when socket is closed
recvEvent chan recvPktEvent // receiver: ingest the specified packet. Sender is readPacket, receiver is goReceiveEvent
sendEvent chan recvPktEvent // sender: ingest the specified packet. Sender is readPacket, receiver is goSendEvent
sendPacket chan packet.Packet // packets to send out on the wire (once goManageConnection is running)
shutdownEvent chan shutdownMessage // channel signals the connection to be shutdown
sockShutdown chan struct{} // closed when socket is shutdown
sockClosed chan struct{} // closed when socket is closed
// timers
connTimeout <-chan time.Time // connecting: fires when connection attempt times out
connRetry <-chan time.Time // connecting: fires when connection attempt to be retried
lingerTimer <-chan time.Time // after disconnection, fires once our linger timer runs out
send *udtSocketSend // reference to sending side of this socket
recv *udtSocketRecv // reference to receiving side of this socket
cong *udtSocketCc // reference to contestion control
// performance metrics
//PktSent uint64 // number of sent data packets, including retransmissions
//PktRecv uint64 // number of received packets
//PktSndLoss uint // number of lost packets (sender side)
//PktRcvLoss uint // number of lost packets (receiver side)
//PktRetrans uint // number of retransmitted packets
//PktSentACK uint // number of sent ACK packets
//PktRecvACK uint // number of received ACK packets
//PktSentNAK uint // number of sent NAK packets
//PktRecvNAK uint // number of received NAK packets
//MbpsSendRate float64 // sending rate in Mb/s
//MbpsRecvRate float64 // receiving rate in Mb/s
//SndDuration time.Duration // busy sending time (i.e., idle time exclusive)
// instant measurements
//PktSndPeriod time.Duration // packet sending period
//PktFlowWindow uint // flow window size, in number of packets
//PktCongestionWindow uint // congestion window size, in number of packets
//PktFlightSize uint // number of packets on flight
//MsRTT time.Duration // RTT
//MbpsBandwidth float64 // estimated bandwidth, in Mb/s
//ByteAvailSndBuf uint // available UDT sender buffer size
//ByteAvailRcvBuf uint // available UDT receiver buffer size
}
/*******************************************************************************
Implementation of net.Conn interface
*******************************************************************************/
// Grab the next data packet
func (s *udtSocket) fetchReadPacket(blocking bool) ([]byte, error) {
var result []byte
if blocking {
for {
if s.readDeadlinePassed {
return nil, syscall.ETIMEDOUT
}
var deadline <-chan time.Time
if s.readDeadline != nil {
deadline = s.readDeadline.C
}
select {
case result = <-s.messageIn:
return result, nil
case _, ok := <-deadline:
if !ok {
continue
}
s.readDeadlinePassed = true
return nil, syscall.ETIMEDOUT
}
}
}
select {
case result = <-s.messageIn:
// ok we have a message
default:
// ok we've read some stuff and there's nothing immediately available
return nil, nil
}
return result, nil
}
func (s *udtSocket) connectionError() error {
switch s.sockState {
case sockStateRefused:
return errors.New("Connection refused by remote host")
case sockStateCorrupted:
return errors.New("Connection closed due to protocol error")
case sockStateClosed:
return errors.New("Connection closed")
case sockStateTimeout:
return errors.New("Connection timed out")
}
return nil
}
// TODO: int sendmsg(const char* data, int len, int msttl, bool inorder)
// Read reads data from the connection.
// Read can be made to time out and return an Error with Timeout() == true
// after a fixed time limit; see SetDeadline and SetReadDeadline.
// (required for net.Conn implementation)
func (s *udtSocket) Read(p []byte) (n int, err error) {
connErr := s.connectionError()
if s.isDatagram {
// for datagram sockets, block until we have a message to return and then return it
// if the buffer isn't big enough, return a truncated message (discarding the rest) and return an error
msg, rerr := s.fetchReadPacket(connErr == nil)
if rerr != nil {
err = rerr
return
}
if msg == nil && connErr != nil {
err = connErr
return
}
n = copy(p, msg)
if n < len(msg) {
err = errors.New("Message truncated")
}
} else {
// for streaming sockets, block until we have at least something to return, then
// fill up the passed buffer as far as we can without blocking again
idx := 0
l := len(p)
n = 0
for idx < l {
if s.currPartialRead == nil {
// Grab the next data packet
currPartialRead, rerr := s.fetchReadPacket(n == 0 && connErr == nil)
s.currPartialRead = currPartialRead
if rerr != nil {
err = rerr
return
}
if s.currPartialRead == nil {
if n != 0 {
return
}
if connErr != nil {
err = connErr
return
}
}
}
thisN := copy(p[idx:], s.currPartialRead)
n = n + thisN
idx = idx + thisN
if n >= len(s.currPartialRead) {
// we've exhausted the current data packet, reset to nil
s.currPartialRead = nil
} else {
s.currPartialRead = s.currPartialRead[n:]
}
}
}
return
}
// Write writes data to the connection.
// Write can be made to time out and return an Error with Timeout() == true
// after a fixed time limit; see SetDeadline and SetWriteDeadline.
// (required for net.Conn implementation)
func (s *udtSocket) Write(p []byte) (n int, err error) {
// at the moment whatever we have right now we'll shove it into a channel and return
// on the other side:
// for datagram sockets: this is a distinct message to be broken into as few packets as possible
// for streaming sockets: collect as much as can fit into a packet and send them out
switch s.sockState {
case sockStateRefused:
err = errors.New("Connection refused by remote host")
return
case sockStateCorrupted:
err = errors.New("Connection closed due to protocol error")
return
case sockStateClosed:
err = errors.New("Connection closed")
return
}
n = len(p)
for {
if s.writeDeadlinePassed {
err = syscall.ETIMEDOUT
return
}
var deadline <-chan time.Time
if s.writeDeadline != nil {
deadline = s.writeDeadline.C
}
select {
case s.messageOut <- sendMessage{content: p, tim: time.Now()}:
// send successful
return
case _, ok := <-deadline:
if !ok {
continue
}
s.writeDeadlinePassed = true
err = syscall.ETIMEDOUT
return
}
}
}
// Close closes the connection.
// Any blocked Read or Write operations will be unblocked.
// Write operations will be permitted to send (initial packets)
// Read operations will return an error
// (required for net.Conn implementation)
func (s *udtSocket) Close() error {
if !s.isOpen() {
return nil // already closed
}
close(s.messageOut)
_, _ = <-s.shutdownEvent
return nil
}
func (s *udtSocket) isOpen() bool {
switch s.sockState {
case sockStateClosed, sockStateRefused, sockStateCorrupted, sockStateTimeout:
return false
default:
return true
}
}
// LocalAddr returns the local network address.
// (required for net.Conn implementation)
func (s *udtSocket) LocalAddr() net.Addr {
//return s.m.laddr
return nil
}
// RemoteAddr returns the remote network address.
// (required for net.Conn implementation)
func (s *udtSocket) RemoteAddr() net.Addr {
//return s.raddr
return nil
}
// SetDeadline sets the read and write deadlines associated
// with the connection. It is equivalent to calling both
// SetReadDeadline and SetWriteDeadline.
//
// A deadline is an absolute time after which I/O operations
// fail with a timeout (see type Error) instead of
// blocking. The deadline applies to all future and pending
// I/O, not just the immediately following call to Read or
// Write. After a deadline has been exceeded, the connection
// can be refreshed by setting a deadline in the future.
//
// An idle timeout can be implemented by repeatedly extending
// the deadline after successful Read or Write calls.
//
// A zero value for t means I/O operations will not time out.
//
// Note that if a TCP connection has keep-alive turned on,
// which is the default unless overridden by Dialer.KeepAlive
// or ListenConfig.KeepAlive, then a keep-alive failure may
// also return a timeout error. On Unix systems a keep-alive
// failure on I/O can be detected using
// errors.Is(err, syscall.ETIMEDOUT).
// (required for net.Conn implementation)
func (s *udtSocket) SetDeadline(t time.Time) error {
s.setDeadline(t, &s.readDeadline, &s.readDeadlinePassed)
s.setDeadline(t, &s.writeDeadline, &s.writeDeadlinePassed)
return nil
}
func (s *udtSocket) setDeadline(dl time.Time, timer **time.Timer, timerPassed *bool) {
if *timer == nil {
if !dl.IsZero() {
*timer = time.NewTimer(dl.Sub(time.Now()))
}
} else {
now := time.Now()
if !dl.IsZero() && dl.Before(now) {
*timerPassed = true
}
oldTime := *timer
if dl.IsZero() {
*timer = nil
}
oldTime.Stop()
_, _ = <-oldTime.C
if !dl.IsZero() && dl.After(now) {
*timerPassed = false
oldTime.Reset(dl.Sub(time.Now()))
}
}
}
// SetReadDeadline sets the deadline for future Read calls
// and any currently-blocked Read call.
// A zero value for t means Read will not time out.
// (required for net.Conn implementation)
func (s *udtSocket) SetReadDeadline(t time.Time) error {
s.setDeadline(t, &s.readDeadline, &s.readDeadlinePassed)
return nil
}
// SetWriteDeadline sets the deadline for future Write calls
// and any currently-blocked Write call.
// Even if write times out, it may return n > 0, indicating that
// some of the data was successfully written.
// A zero value for t means Write will not time out.
// (required for net.Conn implementation)
func (s *udtSocket) SetWriteDeadline(t time.Time) error {
s.setDeadline(t, &s.writeDeadline, &s.writeDeadlinePassed)
return nil
}
/*******************************************************************************
Private functions
*******************************************************************************/
// newSocket creates a new UDT socket, which will be configured afterwards as either an incoming our outgoing socket
func newSocket(m *multiplexer, config *Config, sockID uint32, isServer bool, isDatagram bool) (s *udtSocket) {
now := time.Now()
mtu := m.mtu
if config.MaxPacketSize > 0 && config.MaxPacketSize < mtu {
mtu = config.MaxPacketSize
}
maxFlowWinSize := config.MaxFlowWinSize
if maxFlowWinSize == 0 {
maxFlowWinSize = DefaultConfig().MaxFlowWinSize
}
if maxFlowWinSize < 32 {
maxFlowWinSize = 32
}
s = &udtSocket{
m: m,
Config: config,
//raddr: raddr,
created: now,
sockState: sockStateInit,
udtVer: 4,
isServer: isServer,
mtu: atomicUint32{val: uint32(mtu)},
maxFlowWinSize: maxFlowWinSize,
isDatagram: isDatagram,
sockID: sockID,
initPktSeq: packet.PacketID{Seq: rand.Uint32()},
messageIn: make(chan []byte, 256),
messageOut: make(chan sendMessage, 256),
recvEvent: make(chan recvPktEvent, 256),
sendEvent: make(chan recvPktEvent, 256),
sockClosed: make(chan struct{}, 1),
sockShutdown: make(chan struct{}, 1),
deliveryRate: 16,
bandwidth: 1,
sendPacket: make(chan packet.Packet, 256),
shutdownEvent: make(chan shutdownMessage, 5),
}
s.cong = newUdtSocketCc(s)
return
}
func (s *udtSocket) launchProcessors() {
s.send = newUdtSocketSend(s)
s.recv = newUdtSocketRecv(s)
s.cong.init(s.initPktSeq)
}
func (s *udtSocket) startConnect() error {
connectWait := &sync.WaitGroup{}
s.connectWait = connectWait
connectWait.Add(1)
s.sockState = sockStateConnecting
s.connTimeout = time.After(3 * time.Second)
s.connRetry = time.After(250 * time.Millisecond)
go s.goManageConnection()
s.sendHandshake(packet.HsRequest)
connectWait.Wait()
return s.connectionError()
}
func (s *udtSocket) goManageConnection() {
sockClosed := s.sockClosed
sockShutdown := s.sockShutdown
for {
select {
case <-s.lingerTimer: // linger timer expired, shut everything down
s.m.closeSocket(s.sockID)
close(s.sockClosed)
return
case _, _ = <-sockShutdown:
// catching this to force re-evaluation of this select (catching the linger timer)
case _, _ = <-sockClosed:
return
case p := <-s.sendPacket:
ts := uint32(time.Now().Sub(s.created) / time.Microsecond)
s.cong.onPktSent(p)
fmt.Printf("(id=%d) sending %s (id=%d)\n", s.sockID, packet.PacketTypeName(p.PacketType()), s.farSockID)
s.m.sendPacket(s.farSockID, ts, p)
case sd := <-s.shutdownEvent: // connection shut down
s.shutdown(sd.sockState, sd.permitLinger, sd.err)
case <-s.connTimeout: // connection timed out
s.shutdown(sockStateTimeout, true, nil)
case <-s.connRetry: // resend connection attempt
s.connRetry = nil
switch s.sockState {
case sockStateConnecting:
s.sendHandshake(packet.HsRequest)
s.connRetry = time.After(250 * time.Millisecond)
}
}
}
}
func (s *udtSocket) sendHandshake(reqType packet.HandshakeReqType) {
sockType := packet.TypeSTREAM
if s.isDatagram {
sockType = packet.TypeDGRAM
}
p := &packet.HandshakePacket{
UdtVer: uint32(s.udtVer),
SockType: sockType,
InitPktSeq: s.initPktSeq,
MaxPktSize: s.mtu.get(), // maximum packet size (including UDP/IP headers)
MaxFlowWinSize: uint32(s.maxFlowWinSize), // maximum flow window size
ReqType: reqType,
SockID: s.sockID,
}
ts := uint32(time.Now().Sub(s.created) / time.Microsecond)
s.cong.onPktSent(p)
fmt.Printf("(id=%d) sending handshake(%d) (id=%d)\n", s.sockID, int(reqType), s.farSockID)
s.m.sendPacket(s.farSockID, ts, p)
}
// checkValidHandshake checks to see if we want to accept a new connection with this handshake.
func (s *udtSocket) checkValidHandshake(m *multiplexer, p *packet.HandshakePacket) bool {
if s.udtVer != 4 {
return false
}
return true
}
// readHandshake is received when a handshake packet is received without a destination, either as part
// of a listening response or as a rendezvous connection
func (s *udtSocket) readHandshake(m *multiplexer, p *packet.HandshakePacket) bool {
switch s.sockState {
case sockStateInit: // server accepting a connection from a client
s.initPktSeq = p.InitPktSeq
s.udtVer = int(p.UdtVer)
s.farSockID = p.SockID
s.isDatagram = p.SockType == packet.TypeDGRAM
if s.mtu.get() > p.MaxPktSize {
s.mtu.set(p.MaxPktSize)
}
s.launchProcessors()
s.recv.configureHandshake(p)
s.send.configureHandshake(p, true)
s.sockState = sockStateConnected
s.connTimeout = nil
s.connRetry = nil
go s.goManageConnection()
s.sendHandshake(packet.HsResponse)
return true
case sockStateConnecting: // client attempting to connect to server
if p.ReqType == packet.HsRefused {
s.sockState = sockStateRefused
return true
}
if p.ReqType == packet.HsRequest {
if !s.checkValidHandshake(m, p) || p.InitPktSeq != s.initPktSeq || s.isDatagram != (p.SockType == packet.TypeDGRAM) {
// ignore, not a valid handshake request
return true
}
// handshake isn't done yet, send it back with the cookie we received
s.sendHandshake(packet.HsResponse)
return true
}
if p.ReqType != packet.HsResponse {
// unexpected packet type, ignore
return true
}
if !s.checkValidHandshake(m, p) || p.InitPktSeq != s.initPktSeq || s.isDatagram != (p.SockType == packet.TypeDGRAM) {
// ignore, not a valid handshake request
return true
}
s.farSockID = p.SockID
if s.mtu.get() > p.MaxPktSize {
s.mtu.set(p.MaxPktSize)
}
s.launchProcessors()
s.recv.configureHandshake(p)
s.send.configureHandshake(p, true)
s.connRetry = nil
s.sockState = sockStateConnected
s.connTimeout = nil
if s.connectWait != nil {
s.connectWait.Done()
s.connectWait = nil
}
return true
case sockStateConnected: // server repeating a handshake to a client
if s.isServer && p.ReqType == packet.HsRequest {
// client didn't receive our response handshake, resend it
s.sendHandshake(packet.HsResponse)
} else if !s.isServer && p.ReqType == packet.HsResponse {
// this is a rendezvous connection (re)send our response
s.sendHandshake(packet.HsResponse2)
}
return true
}
return false
}
func (s *udtSocket) shutdown(sockState sockState, permitLinger bool, err error) {
if !s.isOpen() {
return // already closed
}
if err != nil {
fmt.Printf("socket shutdown (type=%d), due to error: %s\n", int(sockState), err.Error())
} else {
fmt.Printf("socket shutdown (type=%d)\n", int(sockState))
}
if s.connectWait != nil {
s.connectWait.Done()
s.connectWait = nil
}
s.sockState = sockState
s.cong.close()
if permitLinger {
linger := s.Config.LingerTime
if linger == 0 {
linger = DefaultConfig().LingerTime
}
s.lingerTimer = time.After(linger)
}
s.connTimeout = nil
s.connRetry = nil
if permitLinger {
close(s.sockShutdown)
} else {
s.m.closeSocket(s.sockID)
close(s.sockClosed)
}
s.messageIn <- nil
}
func absdiff(a uint, b uint) uint {
if a < b {
return b - a
}
return a - b
}
func (s *udtSocket) applyRTT(rtt uint) {
s.rttProt.Lock()
s.rttVar = (s.rttVar*3 + absdiff(s.rtt, rtt)) >> 2
s.rtt = (s.rtt*7 + rtt) >> 3
s.rttProt.Unlock()
}
func (s *udtSocket) getRTT() (rtt, rttVar uint) {
s.rttProt.RLock()
rtt = s.rtt
rttVar = s.rttVar
s.rttProt.RUnlock()
return
}
// Update Estimated Bandwidth and packet delivery rate
func (s *udtSocket) applyReceiveRates(deliveryRate uint, bandwidth uint) {
s.receiveRateProt.Lock()
if deliveryRate > 0 {
s.deliveryRate = (s.deliveryRate*7 + deliveryRate) >> 3
}
if bandwidth > 0 {
s.bandwidth = (s.bandwidth*7 + bandwidth) >> 3
}
s.receiveRateProt.Unlock()
}
func (s *udtSocket) getRcvSpeeds() (deliveryRate uint, bandwidth uint) {
s.receiveRateProt.RLock()
deliveryRate = s.deliveryRate
bandwidth = s.bandwidth
s.receiveRateProt.RUnlock()
return
}
// called by the multiplexer read loop when a packet is received for this socket.
// Minimal processing is permitted but try not to stall the caller
func (s *udtSocket) readPacket(m *multiplexer, p packet.Packet) {
now := time.Now()
if s.sockState == sockStateClosed {
return
}
s.recvEvent <- recvPktEvent{pkt: p, now: now}
switch sp := p.(type) {
case *packet.HandshakePacket: // sent by both peers
s.readHandshake(m, sp)
case *packet.ShutdownPacket: // sent by either peer
s.shutdownEvent <- shutdownMessage{sockState: sockStateClosed, permitLinger: true}
case *packet.AckPacket, *packet.LightAckPacket, *packet.NakPacket: // receiver -> sender
s.sendEvent <- recvPktEvent{pkt: p, now: now}
case *packet.UserDefControlPacket:
s.cong.onCustomMsg(*sp)
}
}

227
udt/udtsocket_cc.go Normal file
View File

@@ -0,0 +1,227 @@
package udt
import (
"time"
"github.com/PeernetOfficial/core/udt/packet"
)
type congMsgType int
const (
congInit congMsgType = iota
congClose
congOnACK
congOnNAK
congOnTimeout
congOnDataPktSent
congOnPktSent
congOnPktRecv
congOnCustomMsg
)
type congMsg struct {
mtyp congMsgType
pktID packet.PacketID
arg interface{}
}
type udtSocketCc struct {
// channels
sockClosed <-chan struct{} // closed when socket is closed
socket *udtSocket
congestion CongestionControl // congestion control object for this socket
msgs chan congMsg
sendPktSeq packet.PacketID // packetID of most recently sent packet
congWindow uint // size of congestion window (in packets)
sndPeriod time.Duration // delay between sending packets
}
func newUdtSocketCc(s *udtSocket) *udtSocketCc {
newCongestion := s.Config.CongestionForSocket
if newCongestion == nil {
newCongestion = DefaultConfig().CongestionForSocket
}
sc := &udtSocketCc{
socket: s,
sockClosed: s.sockClosed,
congestion: newCongestion(s),
msgs: make(chan congMsg, 100),
}
go sc.goCongestionEvent()
return sc
}
func (s *udtSocketCc) goCongestionEvent() {
msgs := s.msgs
sockClosed := s.sockClosed
for {
select {
case evt, ok := <-msgs:
if !ok {
return
}
switch evt.mtyp {
case congInit:
s.sendPktSeq = evt.pktID
s.congestion.Init(s, s.socket.Config.SynTime)
case congClose:
s.congestion.Close(s)
case congOnACK:
s.congestion.OnACK(s, evt.pktID)
case congOnNAK:
s.congestion.OnNAK(s, evt.arg.([]packet.PacketID))
case congOnTimeout:
s.congestion.OnTimeout(s)
case congOnDataPktSent:
s.sendPktSeq = evt.pktID
case congOnPktSent:
s.congestion.OnPktSent(s, evt.arg.(packet.Packet))
case congOnPktRecv:
s.congestion.OnPktRecv(s, evt.arg.(packet.DataPacket))
case congOnCustomMsg:
s.congestion.OnCustomMsg(s, evt.arg.(packet.UserDefControlPacket))
}
case _, _ = <-sockClosed:
return
}
}
}
// Init to be called (only) at the start of a UDT connection.
func (s *udtSocketCc) init(sendPktSeq packet.PacketID) {
s.msgs <- congMsg{
mtyp: congInit,
pktID: sendPktSeq,
}
}
// Close to be called when a UDT connection is closed.
func (s *udtSocketCc) close() {
s.msgs <- congMsg{
mtyp: congClose,
}
}
// OnACK to be called when an ACK packet is received
func (s *udtSocketCc) onACK(pktID packet.PacketID) {
s.msgs <- congMsg{
mtyp: congOnACK,
pktID: pktID,
}
}
// OnNAK to be called when a loss report is received
func (s *udtSocketCc) onNAK(loss []packet.PacketID) {
var ourLoss = make([]packet.PacketID, len(loss))
copy(ourLoss, loss)
s.msgs <- congMsg{
mtyp: congOnNAK,
arg: ourLoss,
}
}
// OnTimeout to be called when a timeout event occurs
func (s *udtSocketCc) onTimeout() {
s.msgs <- congMsg{
mtyp: congOnTimeout,
}
}
// OnPktSent to be called when data is sent
func (s *udtSocketCc) onDataPktSent(pktID packet.PacketID) {
s.msgs <- congMsg{
mtyp: congOnDataPktSent,
pktID: pktID,
}
}
// OnPktSent to be called when data is sent
func (s *udtSocketCc) onPktSent(p packet.Packet) {
s.msgs <- congMsg{
mtyp: congOnPktSent,
arg: p,
}
}
// OnPktRecv to be called when data is received
func (s *udtSocketCc) onPktRecv(p packet.DataPacket) {
s.msgs <- congMsg{
mtyp: congOnPktRecv,
arg: p,
}
}
// OnCustomMsg to process a user-defined packet
func (s *udtSocketCc) onCustomMsg(p packet.UserDefControlPacket) {
s.msgs <- congMsg{
mtyp: congOnCustomMsg,
arg: p,
}
}
// GetSndCurrSeqNo is the most recently sent packet ID
func (s *udtSocketCc) GetSndCurrSeqNo() packet.PacketID {
return s.sendPktSeq
}
// SetCongestionWindowSize sets the size of the congestion window (in packets)
func (s *udtSocketCc) SetCongestionWindowSize(pkt uint) {
s.congWindow = pkt
s.socket.send.congestWindow.set(uint32(pkt))
}
// GetCongestionWindowSize gets the size of the congestion window (in packets)
func (s *udtSocketCc) GetCongestionWindowSize() uint {
return s.congWindow
}
// GetPacketSendPeriod gets the current delay between sending packets
func (s *udtSocketCc) GetPacketSendPeriod() time.Duration {
return s.sndPeriod
}
// SetPacketSendPeriod sets the current delay between sending packets
func (s *udtSocketCc) SetPacketSendPeriod(snd time.Duration) {
s.sndPeriod = snd
s.socket.send.SetPacketSendPeriod(snd)
}
// GetMaxFlowWindow is the largest number of unacknowledged packets we can receive (in packets)
func (s *udtSocketCc) GetMaxFlowWindow() uint {
return s.socket.maxFlowWinSize
}
// GetReceiveRates is the current calculated receive rate and bandwidth (in packets/sec)
func (s *udtSocketCc) GetReceiveRates() (uint, uint) {
return s.socket.getRcvSpeeds()
}
// GetRTT is the current calculated roundtrip time between peers
func (s *udtSocketCc) GetRTT() time.Duration {
rtt, _ := s.socket.getRTT()
return time.Duration(rtt) * time.Microsecond
}
// GetMSS is the largest packet size we can currently send (in bytes)
func (s *udtSocketCc) GetMSS() uint {
return uint(s.socket.mtu.get())
}
// SetACKPerid sets the time between ACKs sent to the peer
func (s *udtSocketCc) SetACKPeriod(ack time.Duration) {
s.socket.recv.ackPeriod.set(ack)
}
// SetACKInterval sets the number of packets sent to the peer before sending an ACK
func (s *udtSocketCc) SetACKInterval(ack uint) {
s.socket.recv.ackInterval.set(uint32(ack))
}
// SetRTOPeriod overrides the default EXP timeout calculations waiting for data from the peer
func (s *udtSocketCc) SetRTOPeriod(rto time.Duration) {
s.socket.send.rtoPeriod.set(rto)
}

592
udt/udtsocket_recv.go Normal file
View File

@@ -0,0 +1,592 @@
package udt
import (
"container/heap"
"fmt"
"time"
"github.com/PeernetOfficial/core/udt/packet"
)
const (
ackSelfClockInterval = 64
)
type udtSocketRecv struct {
// channels
sockClosed <-chan struct{} // closed when socket is closed
sockShutdown <-chan struct{} // closed when socket is shutdown
recvEvent <-chan recvPktEvent // receiver: ingest the specified packet. Sender is readPacket, receiver is goReceiveEvent
messageIn chan<- []byte // inbound messages. Sender is goReceiveEvent->ingestData, Receiver is client caller (Read)
sendPacket chan<- packet.Packet // send a packet out on the wire
socket *udtSocket
farNextPktSeq packet.PacketID // the peer's next largest packet ID expected.
farRecdPktSeq packet.PacketID // the peer's last "received" packet ID (before any loss events)
lastACK uint32 // last ACK packet we've sent
largestACK uint32 // largest ACK packet we've sent that has been acknowledged (by an ACK2).
recvPktPend dataPacketHeap // list of packets that are waiting to be processed.
recvLossList receiveLossHeap // loss list.
ackHistory ackHistoryHeap // list of sent ACKs.
sentAck packet.PacketID // largest packetID we've sent an ACK regarding
recvAck2 packet.PacketID // largest packetID we've received an ACK2 from
recvLastArrival time.Time // time of the most recent data packet arrival
recvLastProbe time.Time // time of the most recent data packet probe packet
ackPeriod atomicDuration // (set by congestion control) delay between sending ACKs
ackInterval atomicUint32 // (set by congestion control) number of data packets to send before sending an ACK
unackPktCount uint // number of packets we've received that we haven't sent an ACK for
lightAckCount uint // number of "light ACK" packets we've sent since the last ACK
recvPktHistory []time.Duration // list of recently received packets.
recvPktPairHistory []time.Duration // probing packet window.
// timers
ackSentEvent2 <-chan time.Time // if an ACK packet has recently sent, don't include link information in the next one
ackSentEvent <-chan time.Time // if an ACK packet has recently sent, wait before resending it
ackTimerEvent <-chan time.Time // controls when to send an ACK to our peer
}
func newUdtSocketRecv(s *udtSocket) *udtSocketRecv {
sr := &udtSocketRecv{
socket: s,
sockClosed: s.sockClosed,
sockShutdown: s.sockShutdown,
recvEvent: s.recvEvent,
messageIn: s.messageIn,
sendPacket: s.sendPacket,
ackTimerEvent: time.After(s.Config.SynTime),
}
go sr.goReceiveEvent()
return sr
}
func (s *udtSocketRecv) configureHandshake(p *packet.HandshakePacket) {
s.farNextPktSeq = p.InitPktSeq
s.farRecdPktSeq = p.InitPktSeq.Add(-1)
s.sentAck = p.InitPktSeq
s.recvAck2 = p.InitPktSeq
}
func (s *udtSocketRecv) goReceiveEvent() {
recvEvent := s.recvEvent
sockClosed := s.sockClosed
sockShutdown := s.sockShutdown
for {
select {
case evt, ok := <-recvEvent:
if !ok {
return
}
switch sp := evt.pkt.(type) {
case *packet.Ack2Packet:
s.ingestAck2(sp, evt.now)
case *packet.MsgDropReqPacket:
s.ingestMsgDropReq(sp, evt.now)
case *packet.DataPacket:
s.ingestData(sp, evt.now)
case *packet.ErrPacket:
s.ingestError(sp)
}
case _, _ = <-sockShutdown: // socket is shut down, no need to receive any further data
return
case _, _ = <-sockClosed: // socket is closed, leave now
return
case <-s.ackSentEvent:
s.ackSentEvent = nil
case <-s.ackSentEvent2:
s.ackSentEvent2 = nil
case <-s.ackTimerEvent:
s.ackEvent()
}
}
}
/*
ACK is used to trigger an acknowledgement (ACK). Its period is set by
the congestion control module. However, UDT will send an ACK no
longer than every 0.01 second, even though the congestion control
does not need timer-based ACK. Here, 0.01 second is defined as the
SYN time, or synchronization time, and it affects many of the other
timers used in UDT.
NAK is used to trigger a negative acknowledgement (NAK). Its period
is dynamically updated to 4 * RTT_+ RTTVar + SYN, where RTTVar is the
variance of RTT samples.
EXP is used to trigger data packets retransmission and maintain
connection status. Its period is dynamically updated to N * (4 * RTT
+ RTTVar + SYN), where N is the number of continuous timeouts. To
avoid unnecessary timeout, a minimum threshold (e.g., 0.5 second)
should be used in the implementation.
*/
// ingestAck2 is called to process an ACK2 packet
func (s *udtSocketRecv) ingestAck2(p *packet.Ack2Packet, now time.Time) {
ackSeq := p.AckSeqNo
if s.ackHistory == nil {
return // no ACKs to search
}
ackHistEntry, ackIdx := s.ackHistory.Find(ackSeq)
if ackHistEntry == nil {
return // this ACK not found
}
if s.recvAck2.BlindDiff(ackHistEntry.lastPacket) < 0 {
s.recvAck2 = ackHistEntry.lastPacket
}
heap.Remove(&s.ackHistory, ackIdx)
// Update the largest ACK number ever been acknowledged.
if s.largestACK < ackSeq {
s.largestACK = ackSeq
}
s.socket.applyRTT(uint(now.Sub(ackHistEntry.sendTime) / time.Microsecond))
//s.rto = 4 * s.rtt + s.rttVar
}
// ingestMsgDropReq is called to process an message drop request packet
func (s *udtSocketRecv) ingestMsgDropReq(p *packet.MsgDropReqPacket, now time.Time) {
stopSeq := p.LastSeq.Add(1)
for pktID := p.FirstSeq; pktID != stopSeq; pktID.Incr() {
// remove all these packets from the loss list
if s.recvLossList != nil {
if lossEntry, idx := s.recvLossList.Find(pktID); lossEntry != nil {
heap.Remove(&s.recvLossList, idx)
}
}
// remove all pending packets with this message
if s.recvPktPend != nil {
if lossEntry, idx := s.recvPktPend.Find(pktID); lossEntry != nil {
heap.Remove(&s.recvPktPend, idx)
}
}
}
if p.FirstSeq == s.farRecdPktSeq.Add(1) {
s.farRecdPktSeq = p.LastSeq
}
if s.recvLossList != nil && len(s.recvLossList) == 0 {
s.farRecdPktSeq = s.farNextPktSeq.Add(-1)
s.recvLossList = nil
}
if s.recvPktPend != nil && len(s.recvPktPend) == 0 {
s.recvPktPend = nil
}
// try to push any pending packets out, now that we have dropped any blocking packets
for s.recvPktPend != nil && stopSeq != s.farNextPktSeq {
nextPkt, _ := s.recvPktPend.Min(stopSeq, s.farNextPktSeq)
if nextPkt == nil || !s.attemptProcessPacket(nextPkt, false) {
break
}
}
}
// ingestData is called to process a data packet
func (s *udtSocketRecv) ingestData(p *packet.DataPacket, now time.Time) {
s.socket.cong.onPktRecv(*p)
seq := p.Seq
/* If the sequence number of the current data packet is 16n + 1,
where n is an integer, record the time interval between this
packet and the last data packet in the Packet Pair Window. */
if (seq.Seq-1)&0xf == 0 {
if !s.recvLastProbe.IsZero() {
if s.recvPktPairHistory == nil {
s.recvPktPairHistory = []time.Duration{now.Sub(s.recvLastProbe)}
} else {
s.recvPktPairHistory = append(s.recvPktPairHistory, now.Sub(s.recvLastProbe))
if len(s.recvPktPairHistory) > 16 {
s.recvPktPairHistory = s.recvPktPairHistory[len(s.recvPktPairHistory)-16:]
}
}
}
s.recvLastProbe = now
}
// Record the packet arrival time in PKT History Window.
if !s.recvLastArrival.IsZero() {
if s.recvPktHistory == nil {
s.recvPktHistory = []time.Duration{now.Sub(s.recvLastArrival)}
} else {
s.recvPktHistory = append(s.recvPktHistory, now.Sub(s.recvLastArrival))
if len(s.recvPktHistory) > 16 {
s.recvPktHistory = s.recvPktHistory[len(s.recvPktHistory)-16:]
}
}
}
s.recvLastArrival = now
/* If the sequence number of the current data packet is greater
than LRSN + 1, put all the sequence numbers between (but
excluding) these two values into the receiver's loss list and
send them to the sender in an NAK packet. */
seqDiff := seq.BlindDiff(s.farNextPktSeq)
if seqDiff > 0 {
newLoss := make(receiveLossHeap, 0, seqDiff)
for idx := s.farNextPktSeq; idx != seq; idx.Incr() {
newLoss = append(newLoss, recvLossEntry{packetID: seq})
}
if s.recvLossList == nil {
s.recvLossList = newLoss
heap.Init(&s.recvLossList)
} else {
for idx := s.farNextPktSeq; idx != seq; idx.Incr() {
heap.Push(&s.recvLossList, recvLossEntry{packetID: seq})
}
heap.Init(&newLoss)
}
s.sendNAK(newLoss)
s.farNextPktSeq = seq.Add(1)
} else if seqDiff < 0 {
// If the sequence number is less than LRSN, remove it from the receiver's loss list.
if !s.recvLossList.Remove(seq) {
return // already previously received packet -- ignore
}
if len(s.recvLossList) == 0 {
s.farRecdPktSeq = s.farNextPktSeq.Add(-1)
s.recvLossList = nil
} else {
s.farRecdPktSeq, _ = s.recvLossList.Min(s.farRecdPktSeq, s.farNextPktSeq)
}
}
s.attemptProcessPacket(p, true)
}
func (s *udtSocketRecv) attemptProcessPacket(p *packet.DataPacket, isNew bool) bool {
seq := p.Seq
// can we process this packet?
boundary, mustOrder, msgID := p.GetMessageData()
if s.recvLossList != nil && mustOrder && s.farRecdPktSeq.Add(1) != seq {
// we're required to order these packets and we're missing prior packets, so push and return
if isNew {
if s.recvPktPend == nil {
s.recvPktPend = dataPacketHeap{p}
heap.Init(&s.recvPktPend)
} else {
heap.Push(&s.recvPktPend, p)
}
}
return false
}
// can we find the start of this message?
pieces := make([]*packet.DataPacket, 0)
cannotContinue := false
switch boundary {
case packet.MbLast, packet.MbMiddle:
// we need prior packets, let's make sure we have them
if s.recvPktPend != nil {
pieceSeq := seq.Add(-1)
for {
prevPiece, _ := s.recvPktPend.Find(pieceSeq)
if prevPiece == nil {
// we don't have the previous piece, is it missing?
if s.recvLossList != nil {
if lossEntry, _ := s.recvLossList.Find(pieceSeq); lossEntry != nil {
// it's missing, stop processing
cannotContinue = true
}
}
// in any case we can't continue with this
fmt.Printf("Message with id %d appears to be a broken fragment\n", msgID)
break
}
prevBoundary, _, prevMsg := prevPiece.GetMessageData()
if prevMsg != msgID {
// ...oops? previous piece isn't in the same message
fmt.Printf("Message with id %d appears to be a broken fragment\n", msgID)
break
}
pieces = append([]*packet.DataPacket{prevPiece}, pieces...)
if prevBoundary == packet.MbFirst {
break
}
pieceSeq.Decr()
}
}
}
if !cannotContinue {
pieces = append(pieces, p)
switch boundary {
case packet.MbFirst, packet.MbMiddle:
// we need following packets, let's make sure we have them
if s.recvPktPend != nil {
pieceSeq := seq.Add(1)
for {
nextPiece, _ := s.recvPktPend.Find(pieceSeq)
if nextPiece == nil {
// we don't have the previous piece, is it missing?
if pieceSeq == s.farNextPktSeq {
// hasn't been received yet
cannotContinue = true
} else if s.recvLossList != nil {
if lossEntry, _ := s.recvLossList.Find(pieceSeq); lossEntry != nil {
// it's missing, stop processing
cannotContinue = true
}
} else {
fmt.Printf("Message with id %d appears to be a broken fragment\n", msgID)
}
// in any case we can't continue with this
break
}
nextBoundary, _, nextMsg := nextPiece.GetMessageData()
if nextMsg != msgID {
// ...oops? previous piece isn't in the same message
fmt.Printf("Message with id %d appears to be a broken fragment\n", msgID)
break
}
pieces = append(pieces, nextPiece)
if nextBoundary == packet.MbLast {
break
}
}
}
}
}
// we've received a data packet, do we need to send an ACK for it?
s.unackPktCount++
ackInterval := uint(s.ackInterval.get())
if (ackInterval > 0) && (ackInterval <= s.unackPktCount) {
// ACK timer expired or ACK interval is reached
s.ackEvent()
} else if ackSelfClockInterval*s.lightAckCount <= s.unackPktCount {
//send a "light" ACK
s.sendLightACK()
s.lightAckCount++
}
if cannotContinue {
// we need to wait for more packets, store and return
if isNew {
if s.recvPktPend == nil {
s.recvPktPend = dataPacketHeap{p}
heap.Init(&s.recvPktPend)
} else {
heap.Push(&s.recvPktPend, p)
}
}
return false
}
// we have a message, pull it from the pending heap (if necessary), assemble it into a message, and return it
if s.recvPktPend != nil {
for _, piece := range pieces {
s.recvPktPend.Remove(piece.Seq)
}
if len(s.recvPktPend) == 0 {
s.recvPktPend = nil
}
}
msg := make([]byte, 0)
for _, piece := range pieces {
msg = append(msg, piece.Data...)
}
s.messageIn <- msg
return true
}
func (s *udtSocketRecv) sendLightACK() {
var ack packet.PacketID
// If there is no loss, the ACK is the current largest sequence number plus 1;
// Otherwise it is the smallest sequence number in the receiver loss list.
if s.recvLossList == nil {
ack = s.farNextPktSeq
} else {
ack = s.farRecdPktSeq.Add(1)
}
if ack != s.recvAck2 {
// send out a lite ACK
// to save time on buffer processing and bandwidth/AS measurement, a lite ACK only feeds back an ACK number
s.sendPacket <- &packet.LightAckPacket{PktSeqHi: ack}
}
}
func (s *udtSocketRecv) getRcvSpeeds() (recvSpeed, bandwidth int) {
// get median value, but cannot change the original value order in the window
if s.recvPktHistory != nil {
ourPktHistory := make(sortableDurnArray, len(s.recvPktHistory))
copy(ourPktHistory, s.recvPktHistory)
n := len(ourPktHistory)
cutPos := n / 2
FloydRivestBuckets(ourPktHistory, cutPos)
median := ourPktHistory[cutPos]
upper := median << 3 // upper bounds
lower := median >> 3 // lower bounds
count := 0 // number of entries inside bounds
var sum time.Duration // sum of values inside bounds
// median filtering
idx := 0
for i := 0; i < n; i++ {
if (ourPktHistory[idx] < upper) && (ourPktHistory[idx] > lower) {
count++
sum += ourPktHistory[idx]
}
idx++
}
// do we have enough valid values to return a value?
// calculate speed
if count > (n >> 1) {
recvSpeed = int(time.Second * time.Duration(count) / sum)
}
}
// get median value, but cannot change the original value order in the window
if s.recvPktPairHistory == nil {
ourProbeHistory := make(sortableDurnArray, len(s.recvPktPairHistory))
copy(ourProbeHistory, s.recvPktPairHistory)
n := len(ourProbeHistory)
cutPos := n / 2
FloydRivestBuckets(ourProbeHistory, cutPos)
median := ourProbeHistory[cutPos]
upper := median << 3 // upper bounds
lower := median >> 3 // lower bounds
count := 1 // number of entries inside bounds
sum := median // sum of values inside bounds
// median filtering
idx := 0
for i := 0; i < n; i++ {
if (ourProbeHistory[idx] < upper) && (ourProbeHistory[idx] > lower) {
count++
sum += ourProbeHistory[idx]
}
idx++
}
bandwidth = int(time.Second * time.Duration(count) / sum)
}
return
}
func (s *udtSocketRecv) sendACK() {
var ack packet.PacketID
// If there is no loss, the ACK is the current largest sequence number plus 1;
// Otherwise it is the smallest sequence number in the receiver loss list.
if s.recvLossList == nil {
ack = s.farNextPktSeq
} else {
ack = s.farRecdPktSeq.Add(1)
}
if ack == s.recvAck2 {
return
}
// only send out an ACK if we either are saying something new or the ackSentEvent has expired
if ack == s.sentAck && s.ackSentEvent != nil {
return
}
s.sentAck = ack
s.lastACK++
ackHist := &ackHistoryEntry{
ackID: s.lastACK,
lastPacket: ack,
sendTime: time.Now(),
}
if s.ackHistory == nil {
s.ackHistory = ackHistoryHeap{ackHist}
heap.Init(&s.ackHistory)
} else {
heap.Push(&s.ackHistory, ackHist)
}
rtt, rttVar := s.socket.getRTT()
numPendPackets := int(s.farNextPktSeq.BlindDiff(s.farRecdPktSeq) - 1)
availWindow := int(s.socket.maxFlowWinSize) - numPendPackets
if availWindow < 2 {
availWindow = 2
}
p := &packet.AckPacket{
AckSeqNo: s.lastACK,
PktSeqHi: ack,
Rtt: uint32(rtt),
RttVar: uint32(rttVar),
BuffAvail: uint32(availWindow),
}
if s.ackSentEvent2 == nil {
recvSpeed, bandwidth := s.getRcvSpeeds()
p.IncludeLink = true
p.PktRecvRate = uint32(recvSpeed)
p.EstLinkCap = uint32(bandwidth)
s.ackSentEvent2 = time.After(s.socket.Config.SynTime)
}
s.sendPacket <- p
s.ackSentEvent = time.After(time.Duration(rtt+4*rttVar) * time.Microsecond)
}
func (s *udtSocketRecv) sendNAK(rl receiveLossHeap) {
lossInfo := make([]uint32, 0)
curPkt := s.farRecdPktSeq
for curPkt != s.farNextPktSeq {
minPkt, idx := rl.Min(curPkt, s.farRecdPktSeq)
if idx < 0 {
break
}
lastPkt := minPkt
for {
nextPkt := lastPkt.Add(1)
_, idx = rl.Find(nextPkt)
if idx < 0 {
break
}
lastPkt = nextPkt
}
if lastPkt == minPkt {
lossInfo = append(lossInfo, minPkt.Seq&0x7FFFFFFF)
} else {
lossInfo = append(lossInfo, minPkt.Seq|0x80000000, lastPkt.Seq&0x7FFFFFFF)
}
}
s.sendPacket <- &packet.NakPacket{CmpLossInfo: lossInfo}
}
// ingestData is called to process an (undocumented) OOB error packet
func (s *udtSocketRecv) ingestError(p *packet.ErrPacket) {
// TODO: umm something
}
// assuming some condition has occured (ACK timer expired, ACK interval), send an ACK and reset the appropriate timer
func (s *udtSocketRecv) ackEvent() {
s.sendACK()
ackTime := s.socket.Config.SynTime
ackPeriod := s.ackPeriod.get()
if ackPeriod > 0 {
ackTime = ackPeriod
}
s.ackTimerEvent = time.After(ackTime)
s.unackPktCount = 0
s.lightAckCount = 1
}

577
udt/udtsocket_send.go Normal file
View File

@@ -0,0 +1,577 @@
package udt
import (
"container/heap"
"fmt"
"time"
"github.com/PeernetOfficial/core/udt/packet"
)
type sendState int
const (
sendStateIdle sendState = iota // not waiting for anything, can send immediately
sendStateSending // recently sent something, waiting for SND before sending more
sendStateWaiting // destination is full, waiting for them to process something and come back
sendStateProcessDrop // immediately re-process any drop list requests
sendStateShutdown // connection is shutdown
)
const (
minEXPinterval time.Duration = 300 * time.Millisecond
)
type udtSocketSend struct {
// channels
sockClosed <-chan struct{} // closed when socket is closed
sockShutdown <-chan struct{} // closed when socket is shutdown
sendEvent <-chan recvPktEvent // sender: ingest the specified packet. Sender is readPacket, receiver is goSendEvent
messageOut <-chan sendMessage // outbound messages. Sender is client caller (Write), Receiver is goSendEvent. Closed when socket is closed
sendPacket chan<- packet.Packet // send a packet out on the wire
shutdownEvent chan<- shutdownMessage // channel signals the connection to be shutdown
socket *udtSocket
sendState sendState // current sender state
sendPktPend sendPacketHeap // list of packets that have been sent but not yet acknoledged
sendPktSeq packet.PacketID // the current packet sequence number
msgPartialSend *sendMessage // when a message can only partially fit in a socket, this is the remainder
msgSeq uint32 // the current message sequence number
expCount uint // number of continuous EXP timeouts.
lastRecvTime time.Time // the last time we've heard something from the remote system
recvAckSeq packet.PacketID // largest packetID we've received an ACK from
sentAck2 uint32 // largest ACK2 packet we've sent
sendLossList packetIDHeap // loss list
sndPeriod atomicDuration // (set by congestion control) delay between sending packets
rtoPeriod atomicDuration // (set by congestion control) override of EXP timer calculations
congestWindow atomicUint32 // (set by congestion control) size of the current congestion window (in packets)
flowWindowSize uint // negotiated maximum number of unacknowledged packets (in packets)
// timers
sndEvent <-chan time.Time // if a packet is recently sent, this timer fires when SND completes
ack2SentEvent <-chan time.Time // if an ACK2 packet has recently sent, wait SYN before sending another one
expTimerEvent <-chan time.Time // Fires when we haven't heard from the peer in a while
}
func newUdtSocketSend(s *udtSocket) *udtSocketSend {
ss := &udtSocketSend{
socket: s,
expCount: 1,
sendPktSeq: s.initPktSeq,
sockClosed: s.sockClosed,
sockShutdown: s.sockShutdown,
sendEvent: s.sendEvent,
messageOut: s.messageOut,
congestWindow: atomicUint32{val: 16},
flowWindowSize: s.maxFlowWinSize,
sendPacket: s.sendPacket,
shutdownEvent: s.shutdownEvent,
}
ss.resetEXP(s.created)
go ss.goSendEvent()
return ss
}
func (s *udtSocketSend) configureHandshake(p *packet.HandshakePacket, resetSeq bool) {
if resetSeq {
s.recvAckSeq = p.InitPktSeq
s.sendPktSeq = p.InitPktSeq
}
s.flowWindowSize = uint(p.MaxFlowWinSize)
}
func (s *udtSocketSend) SetPacketSendPeriod(snd time.Duration) {
// check to see if we have a bandwidth limit here
maxBandwidth := s.socket.Config.MaxBandwidth
if maxBandwidth > 0 {
minSP := time.Second / time.Duration(float64(maxBandwidth)/float64(s.socket.mtu.get()))
if snd < minSP {
snd = minSP
}
}
s.sndPeriod.set(snd)
}
func (s *udtSocketSend) goSendEvent() {
sendEvent := s.sendEvent
messageOut := s.messageOut
sockClosed := s.sockClosed
for {
thisMsgChan := messageOut
sockShutdown := s.sockShutdown
switch s.sendState {
case sendStateIdle: // not waiting for anything, can send immediately
if s.msgPartialSend != nil { // we have a partial message waiting, try to send more of it now
s.processDataMsg(false, messageOut)
continue
}
case sendStateProcessDrop: // immediately re-process any drop list requests
s.sendState = s.reevalSendState() // try to reconstruct what our state should be if it wasn't sendStateProcessDrop
if !s.processSendLoss() || s.sendPktSeq.Seq%16 == 0 {
s.processSendExpire()
}
continue
case sendStateShutdown:
sockShutdown = nil
thisMsgChan = nil
default:
thisMsgChan = nil
}
select {
case _, _ = <-sockShutdown:
s.sendState = sendStateShutdown
s.expTimerEvent = nil // don't process EXP events if we're shutting down
case msg, ok := <-thisMsgChan: // nil if we can't process outgoing messages right now
if !ok {
s.sendPacket <- &packet.ShutdownPacket{}
s.shutdownEvent <- shutdownMessage{sockState: sockStateClosed, permitLinger: true}
return
}
s.msgPartialSend = &msg
s.processDataMsg(true, messageOut)
case evt, ok := <-sendEvent:
if !ok {
return
}
s.expCount = 1
s.resetEXP(evt.now)
switch sp := evt.pkt.(type) {
case *packet.AckPacket:
s.ingestAck(sp, evt.now)
case *packet.LightAckPacket:
s.ingestLightAck(sp, evt.now)
case *packet.NakPacket:
s.ingestNak(sp, evt.now)
case *packet.CongestionPacket:
s.ingestCongestion(sp, evt.now)
}
s.sendState = s.reevalSendState()
case _, _ = <-sockClosed:
return
case <-s.ack2SentEvent: // ACK2 unlocked
s.ack2SentEvent = nil
case now := <-s.expTimerEvent: // EXP event
s.expEvent(now)
case <-s.sndEvent: // SND event
s.sndEvent = nil
if s.sendState == sendStateSending {
s.sendState = s.reevalSendState()
if !s.processSendLoss() || s.sendPktSeq.Seq%16 == 0 {
s.processSendExpire()
}
}
}
}
}
func (s *udtSocketSend) reevalSendState() sendState {
if s.sndEvent != nil {
return sendStateSending
}
// Do we have too many unacknowledged packets for us to send any more?
if s.sendPktPend != nil {
congestWindow := uint(s.congestWindow.get())
cwnd := s.flowWindowSize
if cwnd > congestWindow {
cwnd = congestWindow
}
if cwnd >= uint(len(s.sendPktPend)) {
return sendStateWaiting
}
}
return sendStateIdle
}
// try to pack a new data packet and send it
func (s *udtSocketSend) processDataMsg(isFirst bool, inChan <-chan sendMessage) {
for s.msgPartialSend != nil {
partialSend := s.msgPartialSend
state := packet.MbOnly
if s.socket.isDatagram {
if isFirst {
state = packet.MbFirst
} else {
state = packet.MbMiddle
}
}
if isFirst || !s.socket.isDatagram {
s.msgSeq++
}
mtu := int(s.socket.mtu.get())
msgLen := len(partialSend.content)
if msgLen >= mtu {
// we are full -- send what we can and leave the rest
var dp *packet.DataPacket
if msgLen == mtu {
dp = &packet.DataPacket{
Seq: s.sendPktSeq,
Data: partialSend.content,
}
s.msgPartialSend = nil
} else {
dp = &packet.DataPacket{
Seq: s.sendPktSeq,
Data: partialSend.content[0:mtu],
}
s.msgPartialSend = &sendMessage{content: partialSend.content[mtu:], tim: partialSend.tim, ttl: partialSend.ttl}
}
s.sendPktSeq.Incr()
dp.SetMessageData(state, !s.socket.isDatagram, s.msgSeq)
s.sendDataPacket(sendPacketEntry{pkt: dp, tim: partialSend.tim, ttl: partialSend.ttl}, false)
return
}
// we are not full -- send only if this is a datagram or there's nothing obvious left
if s.socket.isDatagram {
if isFirst {
state = packet.MbOnly
} else {
state = packet.MbLast
}
} else {
select {
case morePartialSend, ok := <-inChan:
if ok {
// we have more data, concat and try again
s.msgPartialSend = &sendMessage{
content: append(s.msgPartialSend.content, morePartialSend.content...),
tim: s.msgPartialSend.tim,
ttl: s.msgPartialSend.ttl,
}
continue
}
default:
// nothing immediately available, just send what we have
}
}
partialSend = s.msgPartialSend
dp := &packet.DataPacket{
Seq: s.sendPktSeq,
Data: partialSend.content,
}
s.msgPartialSend = nil
s.sendPktSeq.Incr()
dp.SetMessageData(state, !s.socket.isDatagram, s.msgSeq)
s.sendDataPacket(sendPacketEntry{pkt: dp, tim: partialSend.tim, ttl: partialSend.ttl}, false)
return
}
}
// If the sender's loss list is not empty, retransmit the first packet in the list and remove it from the list.
func (s *udtSocketSend) processSendLoss() bool {
if s.sendLossList == nil || s.sendPktPend == nil {
return false
}
var dp *sendPacketEntry
for {
minLoss, minLossIdx := s.sendLossList.Min(s.recvAckSeq, s.sendPktSeq)
if minLossIdx < 0 {
// empty loss list? shouldn't really happen as we don't keep empty lists, but check for it anyhow
return false
}
heap.Remove(&s.sendLossList, minLossIdx)
if len(s.sendLossList) == 0 {
s.sendLossList = nil
}
dp, _ = s.sendPktPend.Find(minLoss)
if dp == nil {
// can't find record of this packet, not much we can do really
continue
}
if dp.ttl != 0 && time.Now().Add(dp.ttl).After(dp.tim) {
// this packet has expired, ignore
continue
}
break
}
s.sendDataPacket(*dp, true)
return true
}
// evaluate our pending packet list to see if we have any expired messages
func (s *udtSocketSend) processSendExpire() bool {
if s.sendPktPend == nil {
return false
}
pktPend := make([]sendPacketEntry, len(s.sendPktPend))
copy(pktPend, s.sendPktPend)
for _, p := range pktPend {
if p.ttl != 0 && time.Now().Add(p.ttl).After(p.tim) {
// this message has expired, drop it
_, _, msgNo := p.pkt.GetMessageData()
dropMsg := &packet.MsgDropReqPacket{
MsgID: msgNo,
FirstSeq: p.pkt.Seq,
LastSeq: p.pkt.Seq,
}
// find the other packets in this message
for _, op := range pktPend {
_, _, otherMsgNo := op.pkt.GetMessageData()
if otherMsgNo == msgNo {
if dropMsg.FirstSeq.BlindDiff(p.pkt.Seq) > 0 {
dropMsg.FirstSeq = p.pkt.Seq
}
if dropMsg.LastSeq.BlindDiff(p.pkt.Seq) < 0 {
dropMsg.LastSeq = p.pkt.Seq
}
}
if s.sendLossList != nil {
if _, slIdx := s.sendLossList.Find(p.pkt.Seq); slIdx >= 0 {
heap.Remove(&s.sendLossList, slIdx)
}
}
}
if s.sendLossList != nil && len(s.sendLossList) == 0 {
s.sendLossList = nil
}
s.sendPacket <- dropMsg
return true
}
}
return false
}
// we have a packed packet and a green light to send, so lets send this and mark it
func (s *udtSocketSend) sendDataPacket(dp sendPacketEntry, isResend bool) {
if s.sendPktPend == nil {
s.sendPktPend = sendPacketHeap{dp}
heap.Init(&s.sendPktPend)
} else {
heap.Push(&s.sendPktPend, dp)
}
s.socket.cong.onDataPktSent(dp.pkt.Seq)
s.sendPacket <- dp.pkt
// have we exceeded our recipient's window size?
s.sendState = s.reevalSendState()
if s.sendState == sendStateWaiting {
return
}
if !isResend && dp.pkt.Seq.Seq%16 == 0 {
s.processSendExpire()
return
}
snd := s.sndPeriod.get()
if snd > 0 {
s.sndEvent = time.After(snd)
s.sendState = sendStateSending
}
}
// ingestLightAck is called to process a "light" ACK packet
func (s *udtSocketSend) ingestLightAck(p *packet.LightAckPacket, now time.Time) {
// Update the largest acknowledged sequence number.
pktSeqHi := p.PktSeqHi
diff := pktSeqHi.BlindDiff(s.recvAckSeq)
if diff > 0 {
s.flowWindowSize += uint(diff)
s.recvAckSeq = pktSeqHi
}
}
func (s *udtSocketSend) assertValidSentPktID(pktType string, pktSeq packet.PacketID) bool {
if s.sendPktSeq.BlindDiff(pktSeq) < 0 {
s.shutdownEvent <- shutdownMessage{sockState: sockStateCorrupted, permitLinger: false,
err: fmt.Errorf("FAULT: Received an %s for packet %d, but the largest packet we've sent has been %d", pktType, pktSeq.Seq, s.sendPktSeq.Seq)}
return false
}
return true
}
// ingestAck is called to process an ACK packet
func (s *udtSocketSend) ingestAck(p *packet.AckPacket, now time.Time) {
// Update the largest acknowledged sequence number.
// Send back an ACK2 with the same ACK sequence number in this ACK.
if s.ack2SentEvent == nil && p.AckSeqNo == s.sentAck2 {
s.sentAck2 = p.AckSeqNo
s.sendPacket <- &packet.Ack2Packet{AckSeqNo: p.AckSeqNo}
s.ack2SentEvent = time.After(s.socket.Config.SynTime)
}
pktSeqHi := p.PktSeqHi
if !s.assertValidSentPktID("ACK", pktSeqHi) {
return
}
diff := pktSeqHi.BlindDiff(s.recvAckSeq)
if diff <= 0 {
return
}
oldAckSeq := s.recvAckSeq
s.flowWindowSize = uint(p.BuffAvail)
s.recvAckSeq = pktSeqHi
// Update RTT and RTTVar.
s.socket.applyRTT(uint(p.Rtt))
// Update flow window size.
if p.IncludeLink {
s.socket.applyReceiveRates(uint(p.PktRecvRate), uint(p.EstLinkCap))
}
s.socket.cong.onACK(pktSeqHi)
// Update packet arrival rate: A = (A * 7 + a) / 8, where a is the value carried in the ACK.
// Update estimated link capacity: B = (B * 7 + b) / 8, where b is the value carried in the ACK.
// Update sender's buffer (by releasing the buffer that has been acknowledged).
if s.sendPktPend != nil {
for {
minLoss, minLossIdx := s.sendPktPend.Min(oldAckSeq, s.sendPktSeq)
if pktSeqHi.BlindDiff(minLoss.Seq) >= 0 || minLossIdx < 0 {
break
}
heap.Remove(&s.sendPktPend, minLossIdx)
}
if len(s.sendPktPend) == 0 {
s.sendPktPend = nil
}
}
// Update sender's loss list (by removing all those that has been acknowledged).
if s.sendLossList != nil {
for {
minLoss, minLossIdx := s.sendLossList.Min(oldAckSeq, s.sendPktSeq)
if pktSeqHi.BlindDiff(minLoss) >= 0 || minLossIdx < 0 {
break
}
heap.Remove(&s.sendLossList, minLossIdx)
}
if len(s.sendLossList) == 0 {
s.sendLossList = nil
}
}
}
// ingestNak is called to process an NAK packet
func (s *udtSocketSend) ingestNak(p *packet.NakPacket, now time.Time) {
newLossList := make([]packet.PacketID, 0)
clen := len(p.CmpLossInfo)
for idx := 0; idx < clen; idx++ {
thisEntry := p.CmpLossInfo[idx]
if thisEntry&0x80000000 != 0 {
thisPktID := packet.PacketID{Seq: thisEntry & 0x7FFFFFFF}
if idx+1 == clen {
s.shutdownEvent <- shutdownMessage{sockState: sockStateCorrupted, permitLinger: false,
err: fmt.Errorf("FAULT: While unpacking a NAK, the last entry (%x) was describing a start-of-range", thisEntry)}
return
}
if !s.assertValidSentPktID("NAK", thisPktID) {
return
}
lastEntry := p.CmpLossInfo[idx+1]
if lastEntry&0x80000000 != 0 {
s.shutdownEvent <- shutdownMessage{sockState: sockStateCorrupted, permitLinger: false,
err: fmt.Errorf("FAULT: While unpacking a NAK, a start-of-range (%x) was followed by another start-of-range (%x)", thisEntry, lastEntry)}
return
}
lastPktID := packet.PacketID{Seq: lastEntry}
if !s.assertValidSentPktID("NAK", lastPktID) {
return
}
idx++
for span := thisPktID; span != lastPktID; span.Incr() {
newLossList = append(newLossList, span)
}
} else {
thisPktID := packet.PacketID{Seq: thisEntry}
if !s.assertValidSentPktID("NAK", thisPktID) {
return
}
newLossList = append(newLossList, thisPktID)
}
}
s.socket.cong.onNAK(newLossList)
if s.sendLossList == nil {
s.sendLossList = newLossList
heap.Init(&s.sendLossList)
} else {
llen := len(newLossList)
for idx := 0; idx < llen; idx++ {
heap.Push(&s.sendLossList, newLossList[idx])
}
}
s.sendState = sendStateProcessDrop // immediately restart transmission
}
// ingestCongestion is called to process a (retired?) Congestion packet
func (s *udtSocketSend) ingestCongestion(p *packet.CongestionPacket, now time.Time) {
// One way packet delay is increasing, so decrease the sending rate
// this is very rough (not atomic, doesn't inform congestion) but this is a deprecated message in any case
s.sndPeriod.set(s.sndPeriod.get() * 1125 / 1000)
//m_iLastDecSeq = s.sendPktSeq
}
func (s *udtSocketSend) resetEXP(now time.Time) {
s.lastRecvTime = now
var nextExpDurn time.Duration
rtoPeriod := s.rtoPeriod.get()
if rtoPeriod > 0 {
nextExpDurn = rtoPeriod
} else {
rtt, rttVar := s.socket.getRTT()
nextExpDurn = (time.Duration(s.expCount*(rtt+4*rttVar))*time.Microsecond + s.socket.Config.SynTime)
minExpTime := time.Duration(s.expCount) * minEXPinterval
if nextExpDurn < minExpTime {
nextExpDurn = minExpTime
}
}
s.expTimerEvent = time.After(nextExpDurn)
}
// we've just had the EXP timer expire, see what we can do to recover this
func (s *udtSocketSend) expEvent(currTime time.Time) {
// Haven't receive any information from the peer, is it dead?!
// timeout: at least 16 expirations and must be greater than 10 seconds
if (s.expCount > 16) && (currTime.Sub(s.lastRecvTime) > 5*time.Second) {
// Connection is broken.
s.shutdownEvent <- shutdownMessage{sockState: sockStateTimeout, permitLinger: true}
return
}
// sender: Insert all the packets sent after last received acknowledgement into the sender loss list.
// recver: Send out a keep-alive packet
if s.sendPktPend != nil {
if s.sendPktPend != nil && s.sendLossList == nil {
// resend all unacknowledged packets on timeout, but only if there is no packet in the loss list
newLossList := make([]packet.PacketID, 0)
for span := s.recvAckSeq.Add(1); span != s.sendPktSeq.Add(1); span.Incr() {
newLossList = append(newLossList, span)
}
s.sendLossList = newLossList
heap.Init(&s.sendLossList)
}
s.socket.cong.onTimeout()
s.sendState = sendStateProcessDrop // immediately restart transmission
} else {
s.sendPacket <- &packet.KeepAlivePacket{}
}
s.expCount++
// Reset last response time since we just sent a heart-beat.
s.resetEXP(currTime)
}