Skip to content

Commit

Permalink
Adjust connection windows size dynamically according to ACK received
Browse files Browse the repository at this point in the history 
Signed-off-by: billfort <fxbao@hotmail.com>
  • Loading branch information
@billfort
billfort committed Jan 26, 2023
1 parent 7ebb3f7 commit b397087
Show file tree
Hide file tree
Showing 12 changed files with 1,041 additions and 29 deletions.
33 changes: 25 additions & 8 deletions connection.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ type Connection struct {
session *Session
localClientID string
remoteClientID string
windowSize uint32
windowSize float64
sendWindowUpdate chan struct{}

sync.RWMutex
Expand All @@ -31,7 +31,7 @@ func NewConnection(session *Session, localClientID, remoteClientID string) (*Con
session: session,
localClientID: localClientID,
remoteClientID: remoteClientID,
windowSize: uint32(session.config.InitialConnectionWindowSize),
windowSize: float64(session.sendWindowPacketSize) / float64(len(session.localClientIDs)),
retransmissionTimeout: time.Duration(session.config.InitialRetransmissionTimeout) * time.Millisecond,
sendWindowUpdate: make(chan struct{}, 1),
timeSentSeq: make(map[uint32]time.Time),
Expand Down Expand Up @@ -77,9 +77,6 @@ func (conn *Connection) ReceiveAck(sequenceID uint32, isSentByMe bool) {

if _, ok := conn.resentSeq[sequenceID]; !ok {
conn.windowSize++
if conn.windowSize > uint32(conn.session.config.MaxConnectionWindowSize) {
conn.windowSize = uint32(conn.session.config.MaxConnectionWindowSize)
}
}

if isSentByMe {
Expand All @@ -100,7 +97,7 @@ func (conn *Connection) ReceiveAck(sequenceID uint32, isSentByMe bool) {
}

func (conn *Connection) waitForSendWindow(ctx context.Context) error {
for conn.SendWindowUsed() >= conn.windowSize {
for float64(conn.SendWindowUsed()) >= conn.windowSize {
select {
case <-conn.sendWindowUpdate:
case <-time.After(maxWait):
Expand Down Expand Up @@ -163,6 +160,16 @@ func (conn *Connection) tx() error {
return err
}
log.Println(err)

// reduce window size
conn.Lock()
conn.windowSize /= 2
if conn.windowSize < float64(conn.session.config.MinConnectionWindowSize) {
conn.windowSize = float64(conn.session.config.MinConnectionWindowSize)
}
conn.Unlock()
conn.session.updateConnWindowSize()

select {
case conn.session.resendChan <- seq:
seq = 0
Expand Down Expand Up @@ -268,9 +275,10 @@ func (conn *Connection) checkTimeout() error {
case conn.session.resendChan <- seq:
conn.resentSeq[seq] = struct{}{}
conn.windowSize /= 2
if conn.windowSize < uint32(conn.session.config.MinConnectionWindowSize) {
conn.windowSize = uint32(conn.session.config.MinConnectionWindowSize)
if conn.windowSize < float64(conn.session.config.MinConnectionWindowSize) {
conn.windowSize = float64(conn.session.config.MinConnectionWindowSize)
}
conn.session.updateConnWindowSize()
case <-conn.session.context.Done():
return conn.session.context.Err()
}
Expand All @@ -279,3 +287,12 @@ func (conn *Connection) checkTimeout() error {
conn.Unlock()
}
}

func (conn *Connection) setWindowSize(n float64) {
if n < float64(conn.session.config.MinConnectionWindowSize) {
n = float64(conn.session.config.MinConnectionWindowSize)
}
conn.Lock()
defer conn.Unlock()
conn.windowSize = n
}
5 changes: 4 additions & 1 deletion go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,9 @@ go 1.13
require (
github.com/golang/protobuf v1.4.1
github.com/imdario/mergo v0.3.8
github.com/stretchr/testify v1.4.0 // indirect
github.com/nknorg/mockconn-go v0.0.0-20230125231524-d664e728352a // indirect
github.com/stretchr/testify v1.8.1
github.com/wcharczuk/go-chart/v2 v2.1.0
golang.org/x/time v0.3.0
gopkg.in/yaml.v2 v2.2.8 // indirect
)
19 changes: 19 additions & 0 deletions go.sum
View file
Original file line number Diff line number Diff line change
@@ -1,5 +1,8 @@
github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/golang/freetype v0.0.0-20170609003504-e2365dfdc4a0 h1:DACJavvAHhabrF08vX0COfcOBJRhZ8lUbR+ZWIs0Y5g=
github.com/golang/freetype v0.0.0-20170609003504-e2365dfdc4a0/go.mod h1:E/TSTwGwJL78qG/PmXZO1EjYhfJinVAhrmmHX6Z8B9k=
github.com/golang/protobuf v1.4.0-rc.1/go.mod h1:ceaxUfeHdC40wWswd/P6IGgMaK3YpKi5j83Wpe3EHw8=
github.com/golang/protobuf v1.4.0-rc.1.0.20200221234624-67d41d38c208/go.mod h1:xKAWHe0F5eneWXFV3EuXVDTCmh+JuBKY0li0aMyXATA=
github.com/golang/protobuf v1.4.0-rc.2/go.mod h1:LlEzMj4AhA7rCAGe4KMBDvJI+AwstrUpVNzEA03Pprs=
Expand All @@ -13,11 +16,25 @@ github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4=
github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
github.com/imdario/mergo v0.3.8 h1:CGgOkSJeqMRmt0D9XLWExdT4m4F1vd3FV3VPt+0VxkQ=
github.com/imdario/mergo v0.3.8/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJh5FfA=
github.com/nknorg/mockconn-go v0.0.0-20230125231524-d664e728352a h1:/SGWoUyHDfOgF1G/BQ9x9h1hVj3+mw150Aw4NlWLgtg=
github.com/nknorg/mockconn-go v0.0.0-20230125231524-d664e728352a/go.mod h1:/SvBORYxt9wlm8ZbaEFEri6ooOSDcU3ovU0L2eRRdS4=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk=
github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
github.com/wcharczuk/go-chart/v2 v2.1.0 h1:tY2slqVQ6bN+yHSnDYwZebLQFkphK4WNrVwnt7CJZ2I=
github.com/wcharczuk/go-chart/v2 v2.1.0/go.mod h1:yx7MvAVNcP/kN9lKXM/NTce4au4DFN99j6i1OwDclNA=
golang.org/x/image v0.0.0-20200927104501-e162460cd6b5 h1:QelT11PB4FXiDEXucrfNckHoFxwt8USGY1ajP1ZF5lM=
golang.org/x/image v0.0.0-20200927104501-e162460cd6b5/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/time v0.3.0 h1:rg5rLMjNzMS1RkNLzCG38eapWhnYLFYXDXj2gOlr8j4=
golang.org/x/time v0.3.0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
Expand All @@ -32,3 +49,5 @@ gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8
gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v2 v2.2.8 h1:obN1ZagJSUGI0Ek/LBmuj4SNLPfIny3KsKFopxRdj10=
gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
61 changes: 41 additions & 20 deletions session.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,8 @@ type Session struct {
bytesReadSentTime time.Time
bytesReadUpdateTime time.Time
remoteBytesRead uint64

sendWindowPacketSize float64 // Equal to sendWindowsSize / sendMtu
}

type SendWithFunc func(localClientID, remoteClientID string, buf []byte, writeTimeout time.Duration) error
Expand All @@ -72,26 +74,27 @@ func NewSession(localAddr, remoteAddr net.Addr, localClientIDs, remoteClientIDs
}

session := &Session{
config: config,
localAddr: localAddr,
remoteAddr: remoteAddr,
localClientIDs: localClientIDs,
remoteClientIDs: remoteClientIDs,
sendWith: sendWith,
sendWindowSize: uint32(config.SessionWindowSize),
recvWindowSize: uint32(config.SessionWindowSize),
sendMtu: uint32(config.MTU),
recvMtu: uint32(config.MTU),
sendWindowStartSeq: MinSequenceID,
sendWindowEndSeq: MinSequenceID,
recvWindowStartSeq: MinSequenceID,
recvWindowUsed: 0,
bytesWrite: 0,
bytesRead: 0,
bytesReadSentTime: time.Now(),
bytesReadUpdateTime: time.Now(),
remoteBytesRead: 0,
onAccept: make(chan struct{}, 1),
config: config,
localAddr: localAddr,
remoteAddr: remoteAddr,
localClientIDs: localClientIDs,
remoteClientIDs: remoteClientIDs,
sendWith: sendWith,
sendWindowSize: uint32(config.SessionWindowSize),
recvWindowSize: uint32(config.SessionWindowSize),
sendMtu: uint32(config.MTU),
recvMtu: uint32(config.MTU),
sendWindowStartSeq: MinSequenceID,
sendWindowEndSeq: MinSequenceID,
recvWindowStartSeq: MinSequenceID,
recvWindowUsed: 0,
bytesWrite: 0,
bytesRead: 0,
bytesReadSentTime: time.Now(),
bytesReadUpdateTime: time.Now(),
remoteBytesRead: 0,
onAccept: make(chan struct{}, 1),
sendWindowPacketSize: float64(config.SessionWindowSize) / float64(config.MTU),
}

session.context, session.cancel = context.WithCancel(context.Background())
Expand Down Expand Up @@ -256,6 +259,8 @@ func (session *Session) ReceiveWith(localClientID, remoteClientID string, buf []
}
}
}

session.updateConnWindowSize()
}

if isEstablished && packet.BytesRead > session.remoteBytesRead {
Expand Down Expand Up @@ -519,6 +524,7 @@ func (session *Session) handleHandshakePacket(packet *pb.Packet) error {
if packet.Mtu < session.sendMtu {
session.sendMtu = packet.Mtu
}
session.sendWindowPacketSize = float64(session.sendWindowSize) / float64(session.sendMtu)

if len(packet.ClientIds) == 0 {
return ErrInvalidPacket
Expand Down Expand Up @@ -943,3 +949,18 @@ func (session *Session) SetWriteDeadline(t time.Time) error {
func (session *Session) SetLinger(t int32) {
session.config.Linger = t
}

func (session *Session) updateConnWindowSize() {
totalSize := 0.0
for _, conn := range session.connections {
totalSize += float64(conn.windowSize)
}
if totalSize <= 0 {
return
}

for _, conn := range session.connections {
n := session.sendWindowPacketSize * (conn.windowSize / totalSize)
conn.setWindowSize(n)
}
}
Loading

0 comments on commit b397087

Please sign in to comment.