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

improve garbage collection of stream

Browse Source 
This removes the Iterate method of the streamsMap, and moves the garbage
collection of closed streams from the session to the streamsMap.
This commit is contained in:
Marten Seemann
2017-10-03 04:04:21 +08:00
parent 474b036474
commit 709eec55b5
6 changed files with 161 additions and 250 deletions
Download Patch File Download Diff File Expand all files Collapse all files

169
streams_map_test.go
View File

@@ -27,16 +27,26 @@ var _ = Describe("Streams Map", func() {
mockPn.EXPECT().GetMaxOutgoingStreams().AnyTimes().Return(uint32(maxOutgoingStreams))
mockPn.EXPECT().GetMaxIncomingStreams().AnyTimes().Return(uint32(maxIncomingStreams))
m = newStreamsMap(nil, p, mockPn)
m.newStream = func(id protocol.StreamID) *stream {
return newStream(id, nil, nil, nil)
newStream := func(id protocol.StreamID) *stream {
return newStream(id, func() {}, nil, nil)
}
removeStreamCallback := func(protocol.StreamID) {}
m = newStreamsMap(newStream, removeStreamCallback, p, mockPn)
}
AfterEach(func() {
Expect(m.openStreams).To(HaveLen(len(m.streams)))
})
deleteStream := func(id protocol.StreamID) {
str := m.streams[id]
Expect(str).ToNot(BeNil())
str.cancelled.Set(true)
Expect(str.finished()).To(BeTrue())
err := m.DeleteClosedStreams()
Expect(err).ToNot(HaveOccurred())
}
Context("getting and creating streams", func() {
Context("as a server", func() {
BeforeEach(func() {
@@ -77,8 +87,7 @@ var _ = Describe("Streams Map", func() {
It("returns nil for closed streams", func() {
_, err := m.GetOrOpenStream(5)
Expect(err).NotTo(HaveOccurred())
err = m.RemoveStream(5)
Expect(err).NotTo(HaveOccurred())
deleteStream(5)
s, err := m.GetOrOpenStream(5)
Expect(err).NotTo(HaveOccurred())
Expect(s).To(BeNil())
@@ -95,7 +104,7 @@ var _ = Describe("Streams Map", func() {
It("doesn't reopen an already closed stream", func() {
_, err := m.GetOrOpenStream(5)
Expect(err).ToNot(HaveOccurred())
err = m.RemoveStream(5)
deleteStream(5)
Expect(err).ToNot(HaveOccurred())
str, err := m.GetOrOpenStream(5)
Expect(err).ToNot(HaveOccurred())
@@ -119,9 +128,9 @@ var _ = Describe("Streams Map", func() {
It("does not error when many streams are opened and closed", func() {
for i := 2; i < 10*maxIncomingStreams; i++ {
_, err := m.GetOrOpenStream(protocol.StreamID(i*2 + 1))
str, err := m.GetOrOpenStream(protocol.StreamID(i*2 + 1))
Expect(err).NotTo(HaveOccurred())
m.RemoveStream(protocol.StreamID(i*2 + 1))
deleteStream(str.StreamID())
}
})
})
@@ -148,7 +157,7 @@ var _ = Describe("Streams Map", func() {
str, err := m.OpenStream()
Expect(err).ToNot(HaveOccurred())
Expect(str.StreamID()).To(Equal(protocol.StreamID(2)))
err = m.RemoveStream(2)
deleteStream(2)
Expect(err).ToNot(HaveOccurred())
str, err = m.GetOrOpenStream(2)
Expect(err).ToNot(HaveOccurred())
@@ -169,7 +178,7 @@ var _ = Describe("Streams Map", func() {
for i := 2; i < 10*maxOutgoingStreams; i++ {
str, err := m.OpenStream()
Expect(err).NotTo(HaveOccurred())
m.RemoveStream(str.StreamID())
deleteStream(str.StreamID())
}
})
@@ -208,8 +217,7 @@ var _ = Describe("Streams Map", func() {
}()
Consistently(func() bool { return returned }).Should(BeFalse())
err := m.RemoveStream(6)
Expect(err).ToNot(HaveOccurred())
deleteStream(6)
Eventually(func() bool { return returned }).Should(BeTrue())
Expect(str.StreamID()).To(Equal(protocol.StreamID(2*maxOutgoingStreams + 2)))
})
@@ -404,7 +412,7 @@ var _ = Describe("Streams Map", func() {
str, err := m.OpenStream()
Expect(err).ToNot(HaveOccurred())
Expect(str.StreamID()).To(Equal(protocol.StreamID(1)))
err = m.RemoveStream(1)
deleteStream(1)
Expect(err).ToNot(HaveOccurred())
str, err = m.GetOrOpenStream(1)
Expect(err).ToNot(HaveOccurred())
@@ -433,7 +441,7 @@ var _ = Describe("Streams Map", func() {
It("doesn't reopen an already closed stream", func() {
_, err := m.GetOrOpenStream(4)
Expect(err).ToNot(HaveOccurred())
err = m.RemoveStream(4)
deleteStream(4)
Expect(err).ToNot(HaveOccurred())
str, err := m.GetOrOpenStream(4)
Expect(err).ToNot(HaveOccurred())
@@ -464,6 +472,14 @@ var _ = Describe("Streams Map", func() {
setNewStreamsMap(protocol.PerspectiveServer)
})
closeStream := func(id protocol.StreamID) {
str := m.streams[id]
Expect(str).ToNot(BeNil())
Expect(str.finished()).To(BeFalse())
str.cancelled.Set(true)
Expect(str.finished()).To(BeTrue())
}
Context("deleting streams", func() {
BeforeEach(func() {
for i := 1; i <= 5; i++ {
@@ -473,27 +489,36 @@ var _ = Describe("Streams Map", func() {
Expect(m.openStreams).To(Equal([]protocol.StreamID{1, 2, 3, 4, 5}))
})
It("errors when removing non-existing stream", func() {
err := m.RemoveStream(1337)
Expect(err).To(MatchError("attempted to remove non-existing stream: 1337"))
It("does not delete streams with Close()", func() {
str, err := m.GetOrOpenStream(55)
Expect(err).ToNot(HaveOccurred())
str.Close()
err = m.DeleteClosedStreams()
Expect(err).ToNot(HaveOccurred())
str, err = m.GetOrOpenStream(55)
Expect(err).ToNot(HaveOccurred())
Expect(str).ToNot(BeNil())
})
It("removes the first stream", func() {
err := m.RemoveStream(1)
closeStream(1)
err := m.DeleteClosedStreams()
Expect(err).ToNot(HaveOccurred())
Expect(m.openStreams).To(HaveLen(4))
Expect(m.openStreams).To(Equal([]protocol.StreamID{2, 3, 4, 5}))
})
It("removes a stream in the middle", func() {
err := m.RemoveStream(3)
closeStream(3)
err := m.DeleteClosedStreams()
Expect(err).ToNot(HaveOccurred())
Expect(m.openStreams).To(HaveLen(4))
Expect(m.streams).To(HaveLen(4))
Expect(m.openStreams).To(Equal([]protocol.StreamID{1, 2, 4, 5}))
})
It("removes a stream at the end", func() {
err := m.RemoveStream(5)
closeStream(5)
err := m.DeleteClosedStreams()
Expect(err).ToNot(HaveOccurred())
Expect(m.openStreams).To(HaveLen(4))
Expect(m.openStreams).To(Equal([]protocol.StreamID{1, 2, 3, 4}))
@@ -501,64 +526,15 @@ var _ = Describe("Streams Map", func() {
It("removes all streams", func() {
for i := 1; i <= 5; i++ {
err := m.RemoveStream(protocol.StreamID(i))
Expect(err).ToNot(HaveOccurred())
closeStream(protocol.StreamID(i))
}
err := m.DeleteClosedStreams()
Expect(err).ToNot(HaveOccurred())
Expect(m.streams).To(BeEmpty())
Expect(m.openStreams).To(BeEmpty())
})
})
Context("Iterate", func() {
// create 3 streams, ids 1 to 3
BeforeEach(func() {
for i := 1; i <= 3; i++ {
err := m.putStream(&stream{streamID: protocol.StreamID(i)})
Expect(err).NotTo(HaveOccurred())
}
})
It("executes the lambda exactly once for every stream", func() {
var numIterations int
callbackCalled := make(map[protocol.StreamID]bool)
fn := func(str *stream) (bool, error) {
callbackCalled[str.StreamID()] = true
numIterations++
return true, nil
}
err := m.Iterate(fn)
Expect(err).ToNot(HaveOccurred())
Expect(callbackCalled).To(HaveKey(protocol.StreamID(1)))
Expect(callbackCalled).To(HaveKey(protocol.StreamID(2)))
Expect(callbackCalled).To(HaveKey(protocol.StreamID(3)))
Expect(numIterations).To(Equal(3))
})
It("stops iterating when the callback returns false", func() {
var numIterations int
fn := func(str *stream) (bool, error) {
numIterations++
return false, nil
}
err := m.Iterate(fn)
Expect(err).ToNot(HaveOccurred())
// due to map access randomization, we don't know for which stream the callback was executed
// but it must only be executed once
Expect(numIterations).To(Equal(1))
})
It("returns the error, if the lambda returns one", func() {
var numIterations int
expectedError := errors.New("test")
fn := func(str *stream) (bool, error) {
numIterations++
return true, expectedError
}
err := m.Iterate(fn)
Expect(err).To(MatchError(expectedError))
Expect(numIterations).To(Equal(1))
})
})
Context("RoundRobinIterate", func() {
// create 5 streams, ids 4 to 8
var lambdaCalledForStream []protocol.StreamID
@@ -630,22 +606,39 @@ var _ = Describe("Streams Map", func() {
Expect(lambdaCalledForStream).To(Equal([]protocol.StreamID{6, 7}))
})
It("adjust the RoundRobinIndex when deleting an element in front", func() {
m.roundRobinIndex = 3 // stream 7
m.RemoveStream(5)
Expect(m.roundRobinIndex).To(Equal(uint32(2)))
})
Context("adjusting the RoundRobinIndex when deleting streams", func() {
/*
Index: 0 1 2 3 4
StreamID: [ 4, 5, 6, 7, 8 ]
*/
It("doesn't adjust the RoundRobinIndex when deleting an element at the back", func() {
m.roundRobinIndex = 1 // stream 5
m.RemoveStream(7)
Expect(m.roundRobinIndex).To(BeEquivalentTo(1))
})
It("adjusts when deleting an element in front", func() {
m.roundRobinIndex = 3 // stream 7
deleteStream(5)
Expect(m.roundRobinIndex).To(Equal(uint32(2)))
})
It("doesn't adjust the RoundRobinIndex when deleting the element it is pointing to", func() {
m.roundRobinIndex = 3 // stream 7
m.RemoveStream(7)
Expect(m.roundRobinIndex).To(Equal(uint32(3)))
It("doesn't adjust when deleting an element at the back", func() {
m.roundRobinIndex = 1 // stream 5
deleteStream(7)
Expect(m.roundRobinIndex).To(BeEquivalentTo(1))
})
It("doesn't adjust when deleting the element it is pointing to", func() {
m.roundRobinIndex = 3 // stream 7
deleteStream(7)
Expect(m.roundRobinIndex).To(Equal(uint32(3)))
})
It("adjusts when deleting multiple elements", func() {
m.roundRobinIndex = 3 // stream 7
closeStream(5)
closeStream(6)
closeStream(8)
err := m.DeleteClosedStreams()
Expect(err).ToNot(HaveOccurred())
Expect(m.roundRobinIndex).To(BeEquivalentTo(1))
})
})
Context("Prioritizing crypto- and header streams", func() {