TUN-4597: Add a QUIC server skeleton

- Added a QUIC server to accept streams
- Unit test for this server also tests ALPN
- Temporary echo capability for HTTP ConnectionType

vendor/github.com/lucas-clemente/quic-go/send_stream.go

@@ -0,0 +1,488 @@
+package quic
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/lucas-clemente/quic-go/internal/ackhandler"
+	"github.com/lucas-clemente/quic-go/internal/flowcontrol"
+	"github.com/lucas-clemente/quic-go/internal/protocol"
+	"github.com/lucas-clemente/quic-go/internal/qerr"
+	"github.com/lucas-clemente/quic-go/internal/utils"
+	"github.com/lucas-clemente/quic-go/internal/wire"
+)
+
+type sendStreamI interface {
+	SendStream
+	handleStopSendingFrame(*wire.StopSendingFrame)
+	hasData() bool
+	popStreamFrame(maxBytes protocol.ByteCount) (*ackhandler.Frame, bool)
+	closeForShutdown(error)
+	updateSendWindow(protocol.ByteCount)
+}
+
+type sendStream struct {
+	mutex sync.Mutex
+
+	numOutstandingFrames int64
+	retransmissionQueue  []*wire.StreamFrame
+
+	ctx       context.Context
+	ctxCancel context.CancelFunc
+
+	streamID protocol.StreamID
+	sender   streamSender
+
+	writeOffset protocol.ByteCount
+
+	cancelWriteErr      error
+	closeForShutdownErr error
+
+	closedForShutdown bool // set when CloseForShutdown() is called
+	finishedWriting   bool // set once Close() is called
+	canceledWrite     bool // set when CancelWrite() is called, or a STOP_SENDING frame is received
+	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
+	nextFrame      *wire.StreamFrame
+
+	writeChan chan struct{}
+	deadline  time.Time
+
+	flowController flowcontrol.StreamFlowController
+
+	version protocol.VersionNumber
+}
+
+var (
+	_ SendStream  = &sendStream{}
+	_ sendStreamI = &sendStream{}
+)
+
+func newSendStream(
+	streamID protocol.StreamID,
+	sender streamSender,
+	flowController flowcontrol.StreamFlowController,
+	version protocol.VersionNumber,
+) *sendStream {
+	s := &sendStream{
+		streamID:       streamID,
+		sender:         sender,
+		flowController: flowController,
+		writeChan:      make(chan struct{}, 1),
+		version:        version,
+	}
+	s.ctx, s.ctxCancel = context.WithCancel(context.Background())
+	return s
+}
+
+func (s *sendStream) StreamID() protocol.StreamID {
+	return s.streamID // same for receiveStream and sendStream
+}
+
+func (s *sendStream) Write(p []byte) (int, error) {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	if s.finishedWriting {
+		return 0, fmt.Errorf("write on closed stream %d", s.streamID)
+	}
+	if s.canceledWrite {
+		return 0, s.cancelWriteErr
+	}
+	if s.closeForShutdownErr != nil {
+		return 0, s.closeForShutdownErr
+	}
+	if !s.deadline.IsZero() && !time.Now().Before(s.deadline) {
+		return 0, errDeadline
+	}
+	if len(p) == 0 {
+		return 0, nil
+	}
+
+	s.dataForWriting = p
+
+	var (
+		deadlineTimer  *utils.Timer
+		bytesWritten   int
+		notifiedSender bool
+	)
+	for {
+		var copied bool
+		var deadline time.Time
+		// As soon as dataForWriting becomes smaller than a certain size x, we copy all the data to a STREAM frame (s.nextFrame),
+		// which can the be popped the next time we assemble a packet.
+		// This allows us to return Write() when all data but x bytes have been sent out.
+		// When the user now calls Close(), this is much more likely to happen before we popped that last STREAM frame,
+		// allowing us to set the FIN bit on that frame (instead of sending an empty STREAM frame with FIN).
+		if s.canBufferStreamFrame() && len(s.dataForWriting) > 0 {
+			if s.nextFrame == nil {
+				f := wire.GetStreamFrame()
+				f.Offset = s.writeOffset
+				f.StreamID = s.streamID
+				f.DataLenPresent = true
+				f.Data = f.Data[:len(s.dataForWriting)]
+				copy(f.Data, s.dataForWriting)
+				s.nextFrame = f
+			} else {
+				l := len(s.nextFrame.Data)
+				s.nextFrame.Data = s.nextFrame.Data[:l+len(s.dataForWriting)]
+				copy(s.nextFrame.Data[l:], s.dataForWriting)
+			}
+			s.dataForWriting = nil
+			bytesWritten = len(p)
+			copied = true
+		} else {
+			bytesWritten = len(p) - len(s.dataForWriting)
+			deadline = s.deadline
+			if !deadline.IsZero() {
+				if !time.Now().Before(deadline) {
+					s.dataForWriting = nil
+					return bytesWritten, errDeadline
+				}
+				if deadlineTimer == nil {
+					deadlineTimer = utils.NewTimer()
+					defer deadlineTimer.Stop()
+				}
+				deadlineTimer.Reset(deadline)
+			}
+			if s.dataForWriting == nil || s.canceledWrite || s.closedForShutdown {
+				break
+			}
+		}
+
+		s.mutex.Unlock()
+		if !notifiedSender {
+			s.sender.onHasStreamData(s.streamID) // must be called without holding the mutex
+			notifiedSender = true
+		}
+		if copied {
+			s.mutex.Lock()
+			break
+		}
+		if deadline.IsZero() {
+			<-s.writeChan
+		} else {
+			select {
+			case <-s.writeChan:
+			case <-deadlineTimer.Chan():
+				deadlineTimer.SetRead()
+			}
+		}
+		s.mutex.Lock()
+	}
+
+	if bytesWritten == len(p) {
+		return bytesWritten, nil
+	}
+	if s.closeForShutdownErr != nil {
+		return bytesWritten, s.closeForShutdownErr
+	} else if s.cancelWriteErr != nil {
+		return bytesWritten, s.cancelWriteErr
+	}
+	return bytesWritten, nil
+}
+
+func (s *sendStream) canBufferStreamFrame() bool {
+	var l protocol.ByteCount
+	if s.nextFrame != nil {
+		l = s.nextFrame.DataLen()
+	}
+	return l+protocol.ByteCount(len(s.dataForWriting)) <= protocol.MaxPacketBufferSize
+}
+
+// popStreamFrame returns the next STREAM frame that is supposed to be sent on this stream
+// maxBytes is the maximum length this frame (including frame header) will have.
+func (s *sendStream) popStreamFrame(maxBytes protocol.ByteCount) (*ackhandler.Frame, bool /* has more data to send */) {
+	s.mutex.Lock()
+	f, hasMoreData := s.popNewOrRetransmittedStreamFrame(maxBytes)
+	if f != nil {
+		s.numOutstandingFrames++
+	}
+	s.mutex.Unlock()
+
+	if f == nil {
+		return nil, hasMoreData
+	}
+	return &ackhandler.Frame{Frame: f, OnLost: s.queueRetransmission, OnAcked: s.frameAcked}, hasMoreData
+}
+
+func (s *sendStream) popNewOrRetransmittedStreamFrame(maxBytes protocol.ByteCount) (*wire.StreamFrame, bool /* has more data to send */) {
+	if s.canceledWrite || s.closeForShutdownErr != nil {
+		return nil, false
+	}
+
+	if len(s.retransmissionQueue) > 0 {
+		f, hasMoreRetransmissions := s.maybeGetRetransmission(maxBytes)
+		if f != nil || hasMoreRetransmissions {
+			if f == nil {
+				return nil, true
+			}
+			// We always claim that we have more data to send.
+			// This might be incorrect, in which case there'll be a spurious call to popStreamFrame in the future.
+			return f, true
+		}
+	}
+
+	if len(s.dataForWriting) == 0 && s.nextFrame == nil {
+		if s.finishedWriting && !s.finSent {
+			s.finSent = true
+			return &wire.StreamFrame{
+				StreamID:       s.streamID,
+				Offset:         s.writeOffset,
+				DataLenPresent: true,
+				Fin:            true,
+			}, false
+		}
+		return nil, false
+	}
+
+	sendWindow := s.flowController.SendWindowSize()
+	if sendWindow == 0 {
+		if isBlocked, offset := s.flowController.IsNewlyBlocked(); isBlocked {
+			s.sender.queueControlFrame(&wire.StreamDataBlockedFrame{
+				StreamID:          s.streamID,
+				MaximumStreamData: offset,
+			})
+			return nil, false
+		}
+		return nil, true
+	}
+
+	f, hasMoreData := s.popNewStreamFrame(maxBytes, sendWindow)
+	if dataLen := f.DataLen(); dataLen > 0 {
+		s.writeOffset += f.DataLen()
+		s.flowController.AddBytesSent(f.DataLen())
+	}
+	f.Fin = s.finishedWriting && s.dataForWriting == nil && s.nextFrame == nil && !s.finSent
+	if f.Fin {
+		s.finSent = true
+	}
+	return f, hasMoreData
+}
+
+func (s *sendStream) popNewStreamFrame(maxBytes, sendWindow protocol.ByteCount) (*wire.StreamFrame, bool) {
+	if s.nextFrame != nil {
+		nextFrame := s.nextFrame
+		s.nextFrame = nil
+
+		maxDataLen := utils.MinByteCount(sendWindow, nextFrame.MaxDataLen(maxBytes, s.version))
+		if nextFrame.DataLen() > maxDataLen {
+			s.nextFrame = wire.GetStreamFrame()
+			s.nextFrame.StreamID = s.streamID
+			s.nextFrame.Offset = s.writeOffset + maxDataLen
+			s.nextFrame.Data = s.nextFrame.Data[:nextFrame.DataLen()-maxDataLen]
+			s.nextFrame.DataLenPresent = true
+			copy(s.nextFrame.Data, nextFrame.Data[maxDataLen:])
+			nextFrame.Data = nextFrame.Data[:maxDataLen]
+		} else {
+			s.signalWrite()
+		}
+		return nextFrame, s.nextFrame != nil || s.dataForWriting != nil
+	}
+
+	f := wire.GetStreamFrame()
+	f.Fin = false
+	f.StreamID = s.streamID
+	f.Offset = s.writeOffset
+	f.DataLenPresent = true
+	f.Data = f.Data[:0]
+
+	hasMoreData := s.popNewStreamFrameWithoutBuffer(f, maxBytes, sendWindow)
+	if len(f.Data) == 0 && !f.Fin {
+		f.PutBack()
+		return nil, hasMoreData
+	}
+	return f, hasMoreData
+}
+
+func (s *sendStream) popNewStreamFrameWithoutBuffer(f *wire.StreamFrame, maxBytes, sendWindow protocol.ByteCount) bool {
+	maxDataLen := f.MaxDataLen(maxBytes, s.version)
+	if maxDataLen == 0 { // a STREAM frame must have at least one byte of data
+		return s.dataForWriting != nil || s.nextFrame != nil || s.finishedWriting
+	}
+	s.getDataForWriting(f, utils.MinByteCount(maxDataLen, sendWindow))
+
+	return s.dataForWriting != nil || s.nextFrame != nil || s.finishedWriting
+}
+
+func (s *sendStream) maybeGetRetransmission(maxBytes protocol.ByteCount) (*wire.StreamFrame, bool /* has more retransmissions */) {
+	f := s.retransmissionQueue[0]
+	newFrame, needsSplit := f.MaybeSplitOffFrame(maxBytes, s.version)
+	if needsSplit {
+		return newFrame, true
+	}
+	s.retransmissionQueue = s.retransmissionQueue[1:]
+	return f, len(s.retransmissionQueue) > 0
+}
+
+func (s *sendStream) hasData() bool {
+	s.mutex.Lock()
+	hasData := len(s.dataForWriting) > 0
+	s.mutex.Unlock()
+	return hasData
+}
+
+func (s *sendStream) getDataForWriting(f *wire.StreamFrame, maxBytes protocol.ByteCount) {
+	if protocol.ByteCount(len(s.dataForWriting)) <= maxBytes {
+		f.Data = f.Data[:len(s.dataForWriting)]
+		copy(f.Data, s.dataForWriting)
+		s.dataForWriting = nil
+		s.signalWrite()
+		return
+	}
+	f.Data = f.Data[:maxBytes]
+	copy(f.Data, s.dataForWriting)
+	s.dataForWriting = s.dataForWriting[maxBytes:]
+	if s.canBufferStreamFrame() {
+		s.signalWrite()
+	}
+}
+
+func (s *sendStream) frameAcked(f wire.Frame) {
+	f.(*wire.StreamFrame).PutBack()
+
+	s.mutex.Lock()
+	if s.canceledWrite {
+		s.mutex.Unlock()
+		return
+	}
+	s.numOutstandingFrames--
+	if s.numOutstandingFrames < 0 {
+		panic("numOutStandingFrames negative")
+	}
+	newlyCompleted := s.isNewlyCompleted()
+	s.mutex.Unlock()
+
+	if newlyCompleted {
+		s.sender.onStreamCompleted(s.streamID)
+	}
+}
+
+func (s *sendStream) isNewlyCompleted() bool {
+	completed := (s.finSent || s.canceledWrite) && s.numOutstandingFrames == 0 && len(s.retransmissionQueue) == 0
+	if completed && !s.completed {
+		s.completed = true
+		return true
+	}
+	return false
+}
+
+func (s *sendStream) queueRetransmission(f wire.Frame) {
+	sf := f.(*wire.StreamFrame)
+	sf.DataLenPresent = true
+	s.mutex.Lock()
+	if s.canceledWrite {
+		s.mutex.Unlock()
+		return
+	}
+	s.retransmissionQueue = append(s.retransmissionQueue, sf)
+	s.numOutstandingFrames--
+	if s.numOutstandingFrames < 0 {
+		panic("numOutStandingFrames negative")
+	}
+	s.mutex.Unlock()
+
+	s.sender.onHasStreamData(s.streamID)
+}
+
+func (s *sendStream) Close() error {
+	s.mutex.Lock()
+	if s.closedForShutdown {
+		s.mutex.Unlock()
+		return nil
+	}
+	if s.canceledWrite {
+		s.mutex.Unlock()
+		return fmt.Errorf("close called for canceled stream %d", s.streamID)
+	}
+	s.ctxCancel()
+	s.finishedWriting = true
+	s.mutex.Unlock()
+
+	s.sender.onHasStreamData(s.streamID) // need to send the FIN, must be called without holding the mutex
+	return nil
+}
+
+func (s *sendStream) CancelWrite(errorCode StreamErrorCode) {
+	s.cancelWriteImpl(errorCode, fmt.Errorf("Write on stream %d canceled with error code %d", s.streamID, errorCode))
+}
+
+// must be called after locking the mutex
+func (s *sendStream) cancelWriteImpl(errorCode qerr.StreamErrorCode, writeErr error) {
+	s.mutex.Lock()
+	if s.canceledWrite {
+		s.mutex.Unlock()
+		return
+	}
+	s.ctxCancel()
+	s.canceledWrite = true
+	s.cancelWriteErr = writeErr
+	s.numOutstandingFrames = 0
+	s.retransmissionQueue = nil
+	newlyCompleted := s.isNewlyCompleted()
+	s.mutex.Unlock()
+
+	s.signalWrite()
+	s.sender.queueControlFrame(&wire.ResetStreamFrame{
+		StreamID:  s.streamID,
+		FinalSize: s.writeOffset,
+		ErrorCode: errorCode,
+	})
+	if newlyCompleted {
+		s.sender.onStreamCompleted(s.streamID)
+	}
+}
+
+func (s *sendStream) updateSendWindow(limit protocol.ByteCount) {
+	s.mutex.Lock()
+	hasStreamData := s.dataForWriting != nil || s.nextFrame != nil
+	s.mutex.Unlock()
+
+	s.flowController.UpdateSendWindow(limit)
+	if hasStreamData {
+		s.sender.onHasStreamData(s.streamID)
+	}
+}
+
+func (s *sendStream) handleStopSendingFrame(frame *wire.StopSendingFrame) {
+	s.cancelWriteImpl(frame.ErrorCode, &StreamError{
+		StreamID:  s.streamID,
+		ErrorCode: frame.ErrorCode,
+	})
+}
+
+func (s *sendStream) Context() context.Context {
+	return s.ctx
+}
+
+func (s *sendStream) SetWriteDeadline(t time.Time) error {
+	s.mutex.Lock()
+	s.deadline = t
+	s.mutex.Unlock()
+	s.signalWrite()
+	return nil
+}
+
+// CloseForShutdown closes a stream abruptly.
+// It makes Write unblock (and return the error) immediately.
+// The peer will NOT be informed about this: the stream is closed without sending a FIN or RST.
+func (s *sendStream) closeForShutdown(err error) {
+	s.mutex.Lock()
+	s.ctxCancel()
+	s.closedForShutdown = true
+	s.closeForShutdownErr = err
+	s.mutex.Unlock()
+	s.signalWrite()
+}
+
+// signalWrite performs a non-blocking send on the writeChan
+func (s *sendStream) signalWrite() {
+	select {
+	case s.writeChan <- struct{}{}:
+	default:
+	}
+}