mirrors/uquic
1
0
Fork 0
mirror of https://github.com/refraction-networking/uquic.git synced 2025-04-04 04:37:36 +03:00
Code Issues Projects Releases Packages Wiki Activity

remove unneeded tracking variables from streams, optimize memory layout (#3699)

Browse source 
* remove the closedForShutdown boolean in the send stream

* remove the canceledWrite boolean in the send stream

* remove the closedForShutdown boolean in the receive stream

* remove the canceledRead boolean in the receive stream

* remove the resetRemotely boolean in the receive stream

* optimize memory layout of the receiveStream

This brings it down from 200 to 192 bytes.
This commit is contained in:
Marten Seemann 2023-02-14 10:57:00 +13:00 • committed by GitHub
parent 0ac5d1c149
commit aa091fe672
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 28 additions and 38 deletions
Download patch file Download diff file Expand all files Collapse all files

36
receive_stream.go
View file

@ -34,18 +34,14 @@ type receiveStream struct {
currentFrame []byte currentFrame []byte
currentFrameDone func() currentFrameDone func()
currentFrameIsLast bool // is the currentFrame the last frame on this stream
readPosInFrame int readPosInFrame int
currentFrameIsLast bool // is the currentFrame the last frame on this stream
finRead bool // set once we read a frame with a Fin
closeForShutdownErr error closeForShutdownErr error
cancelReadErr error cancelReadErr error
resetRemotelyErr *StreamError resetRemotelyErr *StreamError
closedForShutdown bool // set when CloseForShutdown() is called
finRead bool // set once we read a frame with a Fin
canceledRead bool // set when CancelRead() is called
resetRemotely bool // set when handleResetStreamFrame() is called
readChan chan struct{} readChan chan struct{}
readOnce chan struct{} // cap: 1, to protect against concurrent use of Read readOnce chan struct{} // cap: 1, to protect against concurrent use of Read
deadline time.Time deadline time.Time
@ -100,13 +96,13 @@ func (s *receiveStream) readImpl(p []byte) (bool /*stream completed */, int, err
if s.finRead { if s.finRead {
return false, 0, io.EOF return false, 0, io.EOF
} }
if s.canceledRead { if s.cancelReadErr != nil {
return false, 0, s.cancelReadErr return false, 0, s.cancelReadErr
} }
if s.resetRemotely { if s.resetRemotelyErr != nil {
return false, 0, s.resetRemotelyErr return false, 0, s.resetRemotelyErr
} }
if s.closedForShutdown { if s.closeForShutdownErr != nil {
return false, 0, s.closeForShutdownErr return false, 0, s.closeForShutdownErr
} }
@ -122,13 +118,13 @@ func (s *receiveStream) readImpl(p []byte) (bool /*stream completed */, int, err
for { for {
// Stop waiting on errors // Stop waiting on errors
if s.closedForShutdown { if s.closeForShutdownErr != nil {
return false, bytesRead, s.closeForShutdownErr return false, bytesRead, s.closeForShutdownErr
} }
if s.canceledRead { if s.cancelReadErr != nil {
return false, bytesRead, s.cancelReadErr return false, bytesRead, s.cancelReadErr
} }
if s.resetRemotely { if s.resetRemotelyErr != nil {
return false, bytesRead, s.resetRemotelyErr return false, bytesRead, s.resetRemotelyErr
} }
@ -175,8 +171,9 @@ func (s *receiveStream) readImpl(p []byte) (bool /*stream completed */, int, err
s.readPosInFrame += m s.readPosInFrame += m
bytesRead += m bytesRead += m
// when a RESET_STREAM was received, the was already informed about the final byteOffset for this stream // when a RESET_STREAM was received, the flow controller was already
if !s.resetRemotely { // informed about the final byteOffset for this stream
if s.resetRemotelyErr == nil {
s.flowController.AddBytesRead(protocol.ByteCount(m)) s.flowController.AddBytesRead(protocol.ByteCount(m))
} }
@ -211,10 +208,9 @@ func (s *receiveStream) CancelRead(errorCode StreamErrorCode) {
} }
func (s *receiveStream) cancelReadImpl(errorCode qerr.StreamErrorCode) bool /* completed */ { func (s *receiveStream) cancelReadImpl(errorCode qerr.StreamErrorCode) bool /* completed */ {
if s.finRead || s.canceledRead || s.resetRemotely { if s.finRead || s.cancelReadErr != nil || s.resetRemotelyErr != nil {
return false return false
} }
s.canceledRead = true
s.cancelReadErr = &StreamError{StreamID: s.streamID, ErrorCode: errorCode, Remote: false} s.cancelReadErr = &StreamError{StreamID: s.streamID, ErrorCode: errorCode, Remote: false}
s.signalRead() s.signalRead()
s.sender.queueControlFrame(&wire.StopSendingFrame{ s.sender.queueControlFrame(&wire.StopSendingFrame{
@ -247,7 +243,7 @@ func (s *receiveStream) handleStreamFrameImpl(frame *wire.StreamFrame) (bool /*
newlyRcvdFinalOffset = s.finalOffset == protocol.MaxByteCount newlyRcvdFinalOffset = s.finalOffset == protocol.MaxByteCount
s.finalOffset = maxOffset s.finalOffset = maxOffset
} }
if s.canceledRead { if s.cancelReadErr != nil {
return newlyRcvdFinalOffset, nil return newlyRcvdFinalOffset, nil
} }
if err := s.frameQueue.Push(frame.Data, frame.Offset, frame.PutBack); err != nil { if err := s.frameQueue.Push(frame.Data, frame.Offset, frame.PutBack); err != nil {
@ -270,7 +266,7 @@ func (s *receiveStream) handleResetStreamFrame(frame *wire.ResetStreamFrame) err
} }
func (s *receiveStream) handleResetStreamFrameImpl(frame *wire.ResetStreamFrame) (bool /*completed */, error) { func (s *receiveStream) handleResetStreamFrameImpl(frame *wire.ResetStreamFrame) (bool /*completed */, error) {
if s.closedForShutdown { if s.closeForShutdownErr != nil {
return false, nil return false, nil
} }
if err := s.flowController.UpdateHighestReceived(frame.FinalSize, true); err != nil { if err := s.flowController.UpdateHighestReceived(frame.FinalSize, true); err != nil {
@ -280,10 +276,9 @@ func (s *receiveStream) handleResetStreamFrameImpl(frame *wire.ResetStreamFrame)
s.finalOffset = frame.FinalSize s.finalOffset = frame.FinalSize
// ignore duplicate RESET_STREAM frames for this stream (after checking their final offset) // ignore duplicate RESET_STREAM frames for this stream (after checking their final offset)
if s.resetRemotely { if s.resetRemotelyErr != nil {
return false, nil return false, nil
} }
s.resetRemotely = true
s.resetRemotelyErr = &StreamError{ s.resetRemotelyErr = &StreamError{
StreamID: s.streamID, StreamID: s.streamID,
ErrorCode: frame.ErrorCode, ErrorCode: frame.ErrorCode,
@ -310,7 +305,6 @@ func (s *receiveStream) SetReadDeadline(t time.Time) error {
// The peer will NOT be informed about this: the stream is closed without sending a FIN or RESET. // The peer will NOT be informed about this: the stream is closed without sending a FIN or RESET.
func (s *receiveStream) closeForShutdown(err error) { func (s *receiveStream) closeForShutdown(err error) {
s.mutex.Lock() s.mutex.Lock()
s.closedForShutdown = true
s.closeForShutdownErr = err s.closeForShutdownErr = err
s.mutex.Unlock() s.mutex.Unlock()
s.signalRead() s.signalRead()

2
receive_stream_test.go
View file

@ -682,7 +682,7 @@ var _ = Describe("Receive Stream", func() {
}) })
It("doesn't do anything when it was closed for shutdown", func() { It("doesn't do anything when it was closed for shutdown", func() {
str.closeForShutdown(nil) str.closeForShutdown(errors.New("shutdown"))
err := str.handleResetStreamFrame(rst) err := str.handleResetStreamFrame(rst)
Expect(err).ToNot(HaveOccurred()) Expect(err).ToNot(HaveOccurred())
}) })

28
send_stream.go
View file

@ -40,11 +40,9 @@ type sendStream struct {
cancelWriteErr error cancelWriteErr error
closeForShutdownErr error closeForShutdownErr error
closedForShutdown bool // set when CloseForShutdown() is called finishedWriting bool // set once Close() is called
finishedWriting bool // set once Close() is called finSent bool // set when a STREAM_FRAME with FIN bit has been sent
canceledWrite bool // set when CancelWrite() is called, or a STOP_SENDING frame is received completed bool // set when this stream has been reported to the streamSender as completed
finSent bool // set when a STREAM_FRAME with FIN bit has been sent
completed bool // set when this stream has been reported to the streamSender as completed
dataForWriting []byte // during a Write() call, this slice is the part of p that still needs to be sent out dataForWriting []byte // during a Write() call, this slice is the part of p that still needs to be sent out
nextFrame *wire.StreamFrame nextFrame *wire.StreamFrame
@ -94,7 +92,7 @@ func (s *sendStream) Write(p []byte) (int, error) {
if s.finishedWriting { if s.finishedWriting {
return 0, fmt.Errorf("write on closed stream %d", s.streamID) return 0, fmt.Errorf("write on closed stream %d", s.streamID)
} }
if s.canceledWrite { if s.cancelWriteErr != nil {
return 0, s.cancelWriteErr return 0, s.cancelWriteErr
} }
if s.closeForShutdownErr != nil { if s.closeForShutdownErr != nil {
@ -153,7 +151,7 @@ func (s *sendStream) Write(p []byte) (int, error) {
} }
deadlineTimer.Reset(deadline) deadlineTimer.Reset(deadline)
} }
if s.dataForWriting == nil || s.canceledWrite || s.closedForShutdown { if s.dataForWriting == nil || s.cancelWriteErr != nil || s.closeForShutdownErr != nil {
break break
} }
} }
@ -219,7 +217,7 @@ func (s *sendStream) popStreamFrame(maxBytes protocol.ByteCount, v protocol.Vers
} }
func (s *sendStream) popNewOrRetransmittedStreamFrame(maxBytes protocol.ByteCount, v protocol.VersionNumber) (*wire.StreamFrame, bool /* has more data to send */) { func (s *sendStream) popNewOrRetransmittedStreamFrame(maxBytes protocol.ByteCount, v protocol.VersionNumber) (*wire.StreamFrame, bool /* has more data to send */) {
if s.canceledWrite || s.closeForShutdownErr != nil { if s.cancelWriteErr != nil || s.closeForShutdownErr != nil {
return nil, false return nil, false
} }
@ -354,7 +352,7 @@ func (s *sendStream) frameAcked(f wire.Frame) {
f.(*wire.StreamFrame).PutBack() f.(*wire.StreamFrame).PutBack()
s.mutex.Lock() s.mutex.Lock()
if s.canceledWrite { if s.cancelWriteErr != nil {
s.mutex.Unlock() s.mutex.Unlock()
return return
} }
@ -371,7 +369,7 @@ func (s *sendStream) frameAcked(f wire.Frame) {
} }
func (s *sendStream) isNewlyCompleted() bool { func (s *sendStream) isNewlyCompleted() bool {
completed := (s.finSent || s.canceledWrite) && s.numOutstandingFrames == 0 && len(s.retransmissionQueue) == 0 completed := (s.finSent || s.cancelWriteErr != nil) && s.numOutstandingFrames == 0 && len(s.retransmissionQueue) == 0
if completed && !s.completed { if completed && !s.completed {
s.completed = true s.completed = true
return true return true
@ -383,7 +381,7 @@ func (s *sendStream) queueRetransmission(f wire.Frame) {
sf := f.(*wire.StreamFrame) sf := f.(*wire.StreamFrame)
sf.DataLenPresent = true sf.DataLenPresent = true
s.mutex.Lock() s.mutex.Lock()
if s.canceledWrite { if s.cancelWriteErr != nil {
s.mutex.Unlock() s.mutex.Unlock()
return return
} }
@ -399,11 +397,11 @@ func (s *sendStream) queueRetransmission(f wire.Frame) {
func (s *sendStream) Close() error { func (s *sendStream) Close() error {
s.mutex.Lock() s.mutex.Lock()
if s.closedForShutdown { if s.closeForShutdownErr != nil {
s.mutex.Unlock() s.mutex.Unlock()
return nil return nil
} }
if s.canceledWrite { if s.cancelWriteErr != nil {
s.mutex.Unlock() s.mutex.Unlock()
return fmt.Errorf("close called for canceled stream %d", s.streamID) return fmt.Errorf("close called for canceled stream %d", s.streamID)
} }
@ -422,12 +420,11 @@ func (s *sendStream) CancelWrite(errorCode StreamErrorCode) {
// must be called after locking the mutex // must be called after locking the mutex
func (s *sendStream) cancelWriteImpl(errorCode qerr.StreamErrorCode, remote bool) { func (s *sendStream) cancelWriteImpl(errorCode qerr.StreamErrorCode, remote bool) {
s.mutex.Lock() s.mutex.Lock()
if s.canceledWrite { if s.cancelWriteErr != nil {
s.mutex.Unlock() s.mutex.Unlock()
return return
} }
s.ctxCancel() s.ctxCancel()
s.canceledWrite = true
s.cancelWriteErr = &StreamError{StreamID: s.streamID, ErrorCode: errorCode, Remote: remote} s.cancelWriteErr = &StreamError{StreamID: s.streamID, ErrorCode: errorCode, Remote: remote}
s.numOutstandingFrames = 0 s.numOutstandingFrames = 0
s.retransmissionQueue = nil s.retransmissionQueue = nil
@ -478,7 +475,6 @@ func (s *sendStream) SetWriteDeadline(t time.Time) error {
func (s *sendStream) closeForShutdown(err error) { func (s *sendStream) closeForShutdown(err error) {
s.mutex.Lock() s.mutex.Lock()
s.ctxCancel() s.ctxCancel()
s.closedForShutdown = true
s.closeForShutdownErr = err s.closeForShutdownErr = err
s.mutex.Unlock() s.mutex.Unlock()
s.signalWrite() s.signalWrite()