gr1ffon/quic-go
1
0
Fork 0
forked from quic-go/quic-go
Code Pull Requests Activity

implement a StopWaitingManager

Browse Source 
fixes #30
This commit is contained in:
Marten Seemann
2016-04-29 16:53:47 +07:00
parent 88a82be2fa
commit 74f7576391
8 changed files with 195 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
ackhandler/interfaces.go
View File

@@ -24,3 +24,10 @@ type ReceivedPacketHandler interface {
DequeueAckFrame() *frames.AckFrame
}
type StopWaitingManager interface {
RegisterPacketForRetransmission(packet *Packet)
GetStopWaitingFrame() *frames.StopWaitingFrame
SentStopWaitingWithPacket(packetNumber protocol.PacketNumber)
ReceivedAckForPacketNumber(packetNumber protocol.PacketNumber)
}

9
ackhandler/sent_packet_handler.go
View File

@@ -33,14 +33,16 @@ type sentPacketHandler struct {
packetHistory map[protocol.PacketNumber]*Packet
retransmissionQueue []*Packet // ToDo: use better data structure
stopWaitingManager StopWaitingManager
bytesInFlight uint64
}
// NewSentPacketHandler creates a new sentPacketHandler
func NewSentPacketHandler() SentPacketHandler {
func NewSentPacketHandler(stopWaitingManager StopWaitingManager) SentPacketHandler {
return &sentPacketHandler{
packetHistory: make(map[protocol.PacketNumber]*Packet),
packetHistory: make(map[protocol.PacketNumber]*Packet),
stopWaitingManager: stopWaitingManager,
}
}
@@ -49,6 +51,9 @@ func (h *sentPacketHandler) ackPacket(packetNumber protocol.PacketNumber) {
h.bytesInFlight -= packet.Length
}
delete(h.packetHistory, packetNumber)
// TODO: add tests
h.stopWaitingManager.ReceivedAckForPacketNumber(packetNumber)
}
func (h *sentPacketHandler) nackPacket(packetNumber protocol.PacketNumber) error {

3
ackhandler/sent_packet_handler_test.go
View File

@@ -13,7 +13,8 @@ var _ = Describe("SentPacketHandler", func() {
var handler *sentPacketHandler
var streamFrame frames.StreamFrame
BeforeEach(func() {
handler = NewSentPacketHandler().(*sentPacketHandler)
stopWaitingManager := NewStopWaitingManager()
handler = NewSentPacketHandler(stopWaitingManager).(*sentPacketHandler)
streamFrame = frames.StreamFrame{
StreamID: 5,
Data: []byte{0x13, 0x37},

52
ackhandler/stop_waiting_manager.go Normal file
View File

@@ -0,0 +1,52 @@
package ackhandler
import (
"github.com/lucas-clemente/quic-go/frames"
"github.com/lucas-clemente/quic-go/protocol"
)
// StopWaitingManager manages StopWaitingFrames
type stopWaitingManager struct {
// sentStopWaitings map[protocol.PacketNumber]protocol.PacketNumber // map[LeastUnacked]stopWaitingSentWithPacketNumber
lastNewStopWaitingFirstSentWithPacketNumber protocol.PacketNumber
currentStopWaitingFrame *frames.StopWaitingFrame
currentStopWaitingFrameSent bool
}
// NewStopWaitingManager creates a new StopWaitingManager
func NewStopWaitingManager() StopWaitingManager {
return &stopWaitingManager{
currentStopWaitingFrame: nil,
}
}
// RegisterPacketForRetransmission prepares the StopWaitingFrame, if neccessary
func (h *stopWaitingManager) RegisterPacketForRetransmission(packet *Packet) {
if h.currentStopWaitingFrame == nil || h.currentStopWaitingFrame.LeastUnacked <= packet.PacketNumber { // <= because for StopWaitingFrames LeastUnacked = packet.PacketNumber + 1
h.currentStopWaitingFrame = &frames.StopWaitingFrame{
LeastUnacked: packet.PacketNumber + 1,
Entropy: byte(packet.Entropy), // TODO: do we have to send out the entropy of this packet or of the next packet, possible fix for #29
}
h.currentStopWaitingFrameSent = false
}
}
// GetStopWaitingFrame gets the StopWaitingFrame that needs to be sent. It returns nil if no StopWaitingFrame needs to be sent
func (h *stopWaitingManager) GetStopWaitingFrame() *frames.StopWaitingFrame {
return h.currentStopWaitingFrame
}
// SentStopWaitingWithPacket must be called after sending out a StopWaitingFrame with a packet
func (h *stopWaitingManager) SentStopWaitingWithPacket(packetNumber protocol.PacketNumber) {
if !h.currentStopWaitingFrameSent {
h.lastNewStopWaitingFirstSentWithPacketNumber = packetNumber
}
h.currentStopWaitingFrameSent = true
}
// ReceivedAckForPacketNumber should be called after receiving an ACK
func (h *stopWaitingManager) ReceivedAckForPacketNumber(packetNumber protocol.PacketNumber) {
if packetNumber >= h.lastNewStopWaitingFirstSentWithPacketNumber {
h.currentStopWaitingFrame = nil
}
}

77
ackhandler/stop_waiting_manager_test.go Normal file
View File

@@ -0,0 +1,77 @@
package ackhandler
import (
"github.com/lucas-clemente/quic-go/protocol"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
)
var _ = Describe("StopWaitingManager", func() {
var manager *stopWaitingManager
BeforeEach(func() {
manager = NewStopWaitingManager().(*stopWaitingManager)
})
It("returns nil in the beginning", func() {
Expect(manager.GetStopWaitingFrame()).To(BeNil())
})
It("gets a StopWaitingFrame after a packet has been registered for retransmission", func() {
leastUnacked := protocol.PacketNumber(10)
manager.RegisterPacketForRetransmission(&Packet{PacketNumber: leastUnacked, Entropy: 10})
swf := manager.GetStopWaitingFrame()
Expect(swf).ToNot(BeNil())
Expect(swf.LeastUnacked).To(Equal(leastUnacked + 1))
Expect(swf.Entropy).To(Equal(byte(10)))
})
It("always gets the StopWaitingFrame for the highest retransmitted packet number", func() {
leastUnacked := protocol.PacketNumber(10)
manager.RegisterPacketForRetransmission(&Packet{PacketNumber: leastUnacked, Entropy: 10})
manager.RegisterPacketForRetransmission(&Packet{PacketNumber: leastUnacked - 1, Entropy: 8})
swf := manager.GetStopWaitingFrame()
Expect(swf).ToNot(BeNil())
Expect(swf.LeastUnacked).To(Equal(leastUnacked + 1))
Expect(swf.Entropy).To(Equal(byte(10)))
})
It("updates the StopWaitingFrame when a packet with a higher packet number is retransmitted", func() {
leastUnacked := protocol.PacketNumber(10)
manager.RegisterPacketForRetransmission(&Packet{PacketNumber: leastUnacked - 1, Entropy: 10})
manager.RegisterPacketForRetransmission(&Packet{PacketNumber: leastUnacked, Entropy: 8})
swf := manager.GetStopWaitingFrame()
Expect(swf).ToNot(BeNil())
Expect(swf.LeastUnacked).To(Equal(leastUnacked + 1))
Expect(swf.Entropy).To(Equal(byte(8)))
})
Context("ACK handling", func() {
It("removes the current StopWaitingFrame when the first packet it was sent with is ACKed", func() {
manager.RegisterPacketForRetransmission(&Packet{PacketNumber: 10})
manager.SentStopWaitingWithPacket(13)
manager.SentStopWaitingWithPacket(14)
manager.SentStopWaitingWithPacket(15)
Expect(manager.GetStopWaitingFrame()).ToNot(BeNil())
manager.ReceivedAckForPacketNumber(13)
Expect(manager.GetStopWaitingFrame()).To(BeNil())
})
It("removes the current StopWaitingFrame when any packet it was sent with is ACKed", func() {
manager.RegisterPacketForRetransmission(&Packet{PacketNumber: 10})
manager.SentStopWaitingWithPacket(13)
manager.SentStopWaitingWithPacket(14)
manager.SentStopWaitingWithPacket(15)
Expect(manager.GetStopWaitingFrame()).ToNot(BeNil())
manager.ReceivedAckForPacketNumber(14)
Expect(manager.GetStopWaitingFrame()).To(BeNil())
})
It("does not remove the current StopWaitingFrame when a packet before the one containing the StopWaitingFrame is ACKed", func() {
manager.RegisterPacketForRetransmission(&Packet{PacketNumber: 10})
manager.SentStopWaitingWithPacket(13)
Expect(manager.GetStopWaitingFrame()).ToNot(BeNil())
manager.ReceivedAckForPacketNumber(12)
Expect(manager.GetStopWaitingFrame()).ToNot(BeNil())
})
})
})

11
packet_packer.go
View File

@@ -34,10 +34,10 @@ func (p *packetPacker) AddHighPrioStreamFrame(f frames.StreamFrame) {
p.streamFrameQueue.Push(&f, true)
}
func (p *packetPacker) PackPacket(controlFrames []frames.Frame, includeStreamFrames bool) (*packedPacket, error) {
func (p *packetPacker) PackPacket(stopWaitingFrame *frames.StopWaitingFrame, controlFrames []frames.Frame, includeStreamFrames bool) (*packedPacket, error) {
// TODO: save controlFrames as a member variable, makes it easier to handle in the unlikely event that there are more controlFrames than you can put into on packet
payloadFrames, err := p.composeNextPacket(controlFrames, includeStreamFrames)
payloadFrames, err := p.composeNextPacket(stopWaitingFrame, controlFrames, includeStreamFrames)
if err != nil {
return nil, err
}
@@ -97,11 +97,16 @@ func (p *packetPacker) getPayload(frames []frames.Frame, currentPacketNumber pro
return payload.Bytes(), nil
}
func (p *packetPacker) composeNextPacket(controlFrames []frames.Frame, includeStreamFrames bool) ([]frames.Frame, error) {
func (p *packetPacker) composeNextPacket(stopWaitingFrame *frames.StopWaitingFrame, controlFrames []frames.Frame, includeStreamFrames bool) ([]frames.Frame, error) {
payloadLength := 0
var payloadFrames []frames.Frame
// TODO: handle the case where there are more controlFrames than we can put into one packet
if stopWaitingFrame != nil {
payloadFrames = append(payloadFrames, stopWaitingFrame)
payloadLength += stopWaitingFrame.MinLength()
}
for len(controlFrames) > 0 {
frame := controlFrames[0]
payloadFrames = append(payloadFrames, frame)

53
packet_packer_test.go
View File

@@ -21,7 +21,7 @@ var _ = Describe("Packet packer", func() {
})
It("returns nil when no packet is queued", func() {
p, err := packer.PackPacket([]frames.Frame{}, true)
p, err := packer.PackPacket(nil, []frames.Frame{}, true)
Expect(p).To(BeNil())
Expect(err).ToNot(HaveOccurred())
})
@@ -32,7 +32,7 @@ var _ = Describe("Packet packer", func() {
Data: []byte{0xDE, 0xCA, 0xFB, 0xAD},
}
packer.AddStreamFrame(f)
p, err := packer.PackPacket([]frames.Frame{}, true)
p, err := packer.PackPacket(nil, []frames.Frame{}, true)
Expect(p).ToNot(BeNil())
Expect(err).ToNot(HaveOccurred())
b := &bytes.Buffer{}
@@ -47,19 +47,28 @@ var _ = Describe("Packet packer", func() {
Data: []byte{0xDE, 0xCA, 0xFB, 0xAD},
}
packer.AddStreamFrame(f)
p, err := packer.PackPacket([]frames.Frame{}, false)
p, err := packer.PackPacket(nil, []frames.Frame{}, false)
Expect(err).ToNot(HaveOccurred())
Expect(p).To(BeNil())
})
It("packs only control frames", func() {
p, err := packer.PackPacket([]frames.Frame{&frames.ConnectionCloseFrame{}}, false)
p, err := packer.PackPacket(nil, []frames.Frame{&frames.ConnectionCloseFrame{}}, false)
Expect(p).ToNot(BeNil())
Expect(err).ToNot(HaveOccurred())
Expect(len(p.frames)).To(Equal(1))
Expect(p.raw).NotTo(HaveLen(0))
})
It("packs a StopWaitingFrame first", func() {
swf := &frames.StopWaitingFrame{LeastUnacked: 10}
p, err := packer.PackPacket(swf, []frames.Frame{&frames.ConnectionCloseFrame{}}, false)
Expect(p).ToNot(BeNil())
Expect(err).ToNot(HaveOccurred())
Expect(len(p.frames)).To(Equal(2))
Expect(p.frames[0]).To(Equal(swf))
})
It("packs many control frames into 1 packets", func() {
f := &frames.AckFrame{LargestObserved: 1}
b := &bytes.Buffer{}
@@ -69,10 +78,10 @@ var _ = Describe("Packet packer", func() {
for i := 0; i < maxFramesPerPacket; i++ {
controlFrames = append(controlFrames, f)
}
payloadFrames, err := packer.composeNextPacket(controlFrames, true)
payloadFrames, err := packer.composeNextPacket(nil, controlFrames, true)
Expect(err).ToNot(HaveOccurred())
Expect(len(payloadFrames)).To(Equal(maxFramesPerPacket))
payloadFrames, err = packer.composeNextPacket([]frames.Frame{}, true)
payloadFrames, err = packer.composeNextPacket(nil, []frames.Frame{}, true)
Expect(err).ToNot(HaveOccurred())
Expect(len(payloadFrames)).To(BeZero())
})
@@ -83,16 +92,16 @@ var _ = Describe("Packet packer", func() {
Data: []byte{0xDE, 0xCA, 0xFB, 0xAD},
}
packer.AddStreamFrame(f)
p, err := packer.PackPacket([]frames.Frame{}, true)
p, err := packer.PackPacket(nil, []frames.Frame{}, true)
Expect(p).ToNot(BeNil())
Expect(err).ToNot(HaveOccurred())
Expect(packer.lastPacketNumber).To(Equal(protocol.PacketNumber(1)))
p, err = packer.PackPacket([]frames.Frame{}, true)
p, err = packer.PackPacket(nil, []frames.Frame{}, true)
Expect(p).To(BeNil())
Expect(err).ToNot(HaveOccurred())
Expect(packer.lastPacketNumber).To(Equal(protocol.PacketNumber(1)))
packer.AddStreamFrame(f)
p, err = packer.PackPacket([]frames.Frame{}, true)
p, err = packer.PackPacket(nil, []frames.Frame{}, true)
Expect(p).ToNot(BeNil())
Expect(err).ToNot(HaveOccurred())
Expect(packer.lastPacketNumber).To(Equal(protocol.PacketNumber(2)))
@@ -106,10 +115,10 @@ var _ = Describe("Packet packer", func() {
Offset: 1,
}
packer.AddStreamFrame(f)
payloadFrames, err := packer.composeNextPacket([]frames.Frame{}, true)
payloadFrames, err := packer.composeNextPacket(nil, []frames.Frame{}, true)
Expect(err).ToNot(HaveOccurred())
Expect(len(payloadFrames)).To(Equal(1))
payloadFrames, err = packer.composeNextPacket([]frames.Frame{}, true)
payloadFrames, err = packer.composeNextPacket(nil, []frames.Frame{}, true)
Expect(err).ToNot(HaveOccurred())
Expect(len(payloadFrames)).To(Equal(0))
})
@@ -125,7 +134,7 @@ var _ = Describe("Packet packer", func() {
}
packer.AddStreamFrame(f1)
packer.AddStreamFrame(f2)
p, err := packer.PackPacket([]frames.Frame{}, true)
p, err := packer.PackPacket(nil, []frames.Frame{}, true)
Expect(p).ToNot(BeNil())
Expect(err).ToNot(HaveOccurred())
b := &bytes.Buffer{}
@@ -142,15 +151,15 @@ var _ = Describe("Packet packer", func() {
Offset: 1,
}
packer.AddStreamFrame(f)
payloadFrames, err := packer.composeNextPacket([]frames.Frame{}, true)
payloadFrames, err := packer.composeNextPacket(nil, []frames.Frame{}, true)
Expect(err).ToNot(HaveOccurred())
Expect(len(payloadFrames)).To(Equal(1))
Expect(len(payloadFrames[0].(*frames.StreamFrame).Data)).To(Equal(maxStreamFrameDataLen))
payloadFrames, err = packer.composeNextPacket([]frames.Frame{}, true)
payloadFrames, err = packer.composeNextPacket(nil, []frames.Frame{}, true)
Expect(err).ToNot(HaveOccurred())
Expect(len(payloadFrames)).To(Equal(1))
Expect(len(payloadFrames[0].(*frames.StreamFrame).Data)).To(Equal(200))
payloadFrames, err = packer.composeNextPacket([]frames.Frame{}, true)
payloadFrames, err = packer.composeNextPacket(nil, []frames.Frame{}, true)
Expect(err).ToNot(HaveOccurred())
Expect(len(payloadFrames)).To(Equal(0))
})
@@ -167,16 +176,16 @@ var _ = Describe("Packet packer", func() {
}
packer.AddStreamFrame(f1)
packer.AddStreamFrame(f2)
p, err := packer.PackPacket([]frames.Frame{}, true)
p, err := packer.PackPacket(nil, []frames.Frame{}, true)
Expect(err).ToNot(HaveOccurred())
Expect(len(p.raw)).To(Equal(protocol.MaxPacketSize))
p, err = packer.PackPacket([]frames.Frame{}, true)
p, err = packer.PackPacket(nil, []frames.Frame{}, true)
Expect(err).ToNot(HaveOccurred())
Expect(len(p.raw)).To(Equal(protocol.MaxPacketSize))
p, err = packer.PackPacket([]frames.Frame{}, true)
p, err = packer.PackPacket(nil, []frames.Frame{}, true)
Expect(err).ToNot(HaveOccurred())
Expect(p).ToNot(BeNil())
p, err = packer.PackPacket([]frames.Frame{}, true)
p, err = packer.PackPacket(nil, []frames.Frame{}, true)
Expect(err).ToNot(HaveOccurred())
Expect(p).To(BeNil())
})
@@ -187,7 +196,7 @@ var _ = Describe("Packet packer", func() {
Offset: 1,
}
packer.AddStreamFrame(f)
p, err := packer.PackPacket([]frames.Frame{}, true)
p, err := packer.PackPacket(nil, []frames.Frame{}, true)
Expect(err).ToNot(HaveOccurred())
Expect(p).ToNot(BeNil())
Expect(len(p.raw)).To(Equal(protocol.MaxPacketSize))
@@ -199,10 +208,10 @@ var _ = Describe("Packet packer", func() {
Offset: 1,
}
packer.AddStreamFrame(f)
payloadFrames, err := packer.composeNextPacket([]frames.Frame{}, true)
payloadFrames, err := packer.composeNextPacket(nil, []frames.Frame{}, true)
Expect(err).ToNot(HaveOccurred())
Expect(len(payloadFrames)).To(Equal(1))
payloadFrames, err = packer.composeNextPacket([]frames.Frame{}, true)
payloadFrames, err = packer.composeNextPacket(nil, []frames.Frame{}, true)
Expect(err).ToNot(HaveOccurred())
Expect(len(payloadFrames)).To(Equal(1))
})

21
session.go
View File

@@ -41,6 +41,7 @@ type Session struct {
sentPacketHandler ackhandler.SentPacketHandler
receivedPacketHandler ackhandler.ReceivedPacketHandler
stopWaitingManager ackhandler.StopWaitingManager
unpacker *packetUnpacker
packer *packetPacker
@@ -55,13 +56,15 @@ type Session struct {
// NewSession makes a new session
func NewSession(conn connection, v protocol.VersionNumber, connectionID protocol.ConnectionID, sCfg *handshake.ServerConfig, streamCallback StreamCallback) PacketHandler {
stopWaitingManager := ackhandler.NewStopWaitingManager()
session := &Session{
connectionID: connectionID,
conn: conn,
streamCallback: streamCallback,
streams: make(map[protocol.StreamID]*stream),
sentPacketHandler: ackhandler.NewSentPacketHandler(),
sentPacketHandler: ackhandler.NewSentPacketHandler(stopWaitingManager),
receivedPacketHandler: ackhandler.NewReceivedPacketHandler(),
stopWaitingManager: stopWaitingManager,
receivedPackets: make(chan receivedPacket),
closeChan: make(chan struct{}, 1),
}
@@ -222,7 +225,7 @@ func (s *Session) Close(e error) error {
errorCode = quicError.ErrorCode
}
s.closeStreamsWithError(e)
packet, err := s.packer.PackPacket([]frames.Frame{
packet, err := s.packer.PackPacket(nil, []frames.Frame{
&frames.ConnectionCloseFrame{ErrorCode: errorCode, ReasonPhrase: reasonPhrase},
}, false)
if err != nil {
@@ -252,25 +255,21 @@ func (s *Session) sendPacket() error {
// TODO: handle multiple packets retransmissions
retransmitPacket := s.sentPacketHandler.DequeuePacketForRetransmission()
if retransmitPacket != nil {
swf := &frames.StopWaitingFrame{
LeastUnacked: retransmitPacket.PacketNumber + 1,
Entropy: byte(retransmitPacket.Entropy),
}
controlFrames = append(controlFrames, swf)
s.stopWaitingManager.RegisterPacketForRetransmission(retransmitPacket)
// resend the frames that were in the packet
controlFrames = append(controlFrames, retransmitPacket.GetControlFramesForRetransmission()...)
for _, streamFrame := range retransmitPacket.GetStreamFramesForRetransmission() {
// TODO: add these stream frames with a higher priority
s.packer.AddHighPrioStreamFrame(*streamFrame)
}
}
stopWaitingFrame := s.stopWaitingManager.GetStopWaitingFrame()
ack := s.receivedPacketHandler.DequeueAckFrame()
if ack != nil {
controlFrames = append(controlFrames, ack)
}
packet, err := s.packer.PackPacket(controlFrames, true)
packet, err := s.packer.PackPacket(stopWaitingFrame, controlFrames, true)
if err != nil {
return err
@@ -289,6 +288,8 @@ func (s *Session) sendPacket() error {
return err
}
s.stopWaitingManager.SentStopWaitingWithPacket(packet.number)
fmt.Printf("-> Sending packet %d (%d bytes)\n", packet.number, len(packet.raw))
err = s.conn.write(packet.raw)
if err != nil {