split the stream into a receive and a send stream

crypto_stream.go

@@ -37,13 +37,13 @@ func newCryptoStream(onData func(), flowController flowcontrol.StreamFlowControl
 // It is only needed for the crypto stream.
 // It must not be called concurrently with any other stream methods, especially Read and Write.
 func (s *cryptoStream) SetReadOffset(offset protocol.ByteCount) {
-	s.readOffset = offset
-	s.frameQueue.readPosition = offset
+	s.receiveStream.readOffset = offset
+	s.receiveStream.frameQueue.readPosition = offset
 }
 
 func (s *cryptoStream) HasDataForWriting() bool {
-	s.mutex.Lock()
-	hasData := s.dataForWriting != nil
-	s.mutex.Unlock()
+	s.sendStream.mutex.Lock()
+	hasData := s.sendStream.dataForWriting != nil
+	s.sendStream.mutex.Unlock()
 	return hasData
 }

crypto_stream_test.go

@@ -14,8 +14,8 @@ var _ = Describe("Stream", func() {
 
 	It("sets the read offset", func() {
 		str.SetReadOffset(0x42)
-		Expect(str.readOffset).To(Equal(protocol.ByteCount(0x42)))
-		Expect(str.frameQueue.readPosition).To(Equal(protocol.ByteCount(0x42)))
+		Expect(str.receiveStream.readOffset).To(Equal(protocol.ByteCount(0x42)))
+		Expect(str.receiveStream.frameQueue.readPosition).To(Equal(protocol.ByteCount(0x42)))
 	})
 
 	It("says if it has data for writing", func() {

interface.go

@@ -24,6 +24,8 @@ type ErrorCode = protocol.ApplicationErrorCode
 
 // Stream is the interface implemented by QUIC streams
 type Stream interface {
+	// StreamID returns the stream ID.
+	StreamID() StreamID
 	// Read reads data from the stream.
 	// Read can be made to time out and return a net.Error with Timeout() == true
 	// after a fixed time limit; see SetDeadline and SetReadDeadline.
@@ -41,7 +43,6 @@ type Stream interface {
 	// It must not be called concurrently with Write.
 	// It must not be called after calling CancelWrite.
 	io.Closer
-	StreamID() StreamID
 	// CancelWrite aborts sending on this stream.
 	// It must not be called after Close.
 	// Data already written, but not yet delivered to the peer is not guaranteed to be delivered reliably.
@@ -71,6 +72,34 @@ type Stream interface {
 	SetDeadline(t time.Time) error
 }
 
+// A ReceiveStream is a unidirectional Receive Stream.
+type ReceiveStream interface {
+	// see Stream.StreamID
+	StreamID() StreamID
+	// see Stream.Read
+	io.Reader
+	// see Stream.CancelRead
+	CancelRead(ErrorCode) error
+	// see Stream.SetReadDealine
+	SetReadDeadline(t time.Time) error
+}
+
+// A SendStream is a unidirectional Send Stream.
+type SendStream interface {
+	// see Stream.StreamID
+	StreamID() StreamID
+	// see Stream.Write
+	io.Writer
+	// see Stream.Close
+	io.Closer
+	// see Stream.CancelWrite
+	CancelWrite(ErrorCode) error
+	// see Stream.Context
+	Context() context.Context
+	// see Stream.SetWriteDeadline
+	SetWriteDeadline(t time.Time) error
+}
+
 // StreamError is returned by Read and Write when the peer cancels the stream.
 type StreamError interface {
 	error

receive_stream.go

@@ -0,0 +1,284 @@
+package quic
+
+import (
+	"fmt"
+	"io"
+	"sync"
+	"time"
+
+	"github.com/lucas-clemente/quic-go/internal/flowcontrol"
+	"github.com/lucas-clemente/quic-go/internal/protocol"
+	"github.com/lucas-clemente/quic-go/internal/utils"
+	"github.com/lucas-clemente/quic-go/internal/wire"
+)
+
+type receiveStream struct {
+	mutex sync.Mutex
+
+	streamID protocol.StreamID
+
+	// onData tells the session that there's stuff to pack into a new packet
+	onData func()
+	// queueControlFrame queues a new control frame for sending
+	// it does not call onData
+	queueControlFrame func(wire.Frame)
+
+	frameQueue     *streamFrameSorter
+	readPosInFrame int
+	readOffset     protocol.ByteCount
+
+	closeForShutdownErr error
+	cancelReadErr       error
+	resetRemotelyErr    StreamError
+
+	closedForShutdown bool // set when CloseForShutdown() is called
+	finRead           bool // set once we read a frame with a FinBit
+	canceledRead      bool // set when CancelRead() is called
+	resetRemotely     bool // set when HandleRstStreamFrame() is called
+
+	readChan     chan struct{}
+	readDeadline time.Time
+
+	flowController flowcontrol.StreamFlowController
+	version        protocol.VersionNumber
+}
+
+var _ ReceiveStream = &receiveStream{}
+
+func newReceiveStream(
+	streamID protocol.StreamID,
+	onData func(),
+	queueControlFrame func(wire.Frame),
+	flowController flowcontrol.StreamFlowController,
+) *receiveStream {
+	return &receiveStream{
+		streamID:          streamID,
+		onData:            onData,
+		queueControlFrame: queueControlFrame,
+		flowController:    flowController,
+		frameQueue:        newStreamFrameSorter(),
+		readChan:          make(chan struct{}, 1),
+	}
+}
+
+func (s *receiveStream) StreamID() protocol.StreamID {
+	return s.streamID
+}
+
+// Read implements io.Reader. It is not thread safe!
+func (s *receiveStream) Read(p []byte) (int, error) {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	if s.finRead {
+		return 0, io.EOF
+	}
+	if s.canceledRead {
+		return 0, s.cancelReadErr
+	}
+	if s.resetRemotely {
+		return 0, s.resetRemotelyErr
+	}
+	if s.closedForShutdown {
+		return 0, s.closeForShutdownErr
+	}
+
+	bytesRead := 0
+	for bytesRead < len(p) {
+		frame := s.frameQueue.Head()
+		if frame == nil && bytesRead > 0 {
+			return bytesRead, s.closeForShutdownErr
+		}
+
+		for {
+			// Stop waiting on errors
+			if s.closedForShutdown {
+				return bytesRead, s.closeForShutdownErr
+			}
+			if s.canceledRead {
+				return bytesRead, s.cancelReadErr
+			}
+			if s.resetRemotely {
+				return bytesRead, s.resetRemotelyErr
+			}
+
+			deadline := s.readDeadline
+			if !deadline.IsZero() && !time.Now().Before(deadline) {
+				return bytesRead, errDeadline
+			}
+
+			if frame != nil {
+				s.readPosInFrame = int(s.readOffset - frame.Offset)
+				break
+			}
+
+			s.mutex.Unlock()
+			if deadline.IsZero() {
+				<-s.readChan
+			} else {
+				select {
+				case <-s.readChan:
+				case <-time.After(deadline.Sub(time.Now())):
+				}
+			}
+			s.mutex.Lock()
+			frame = s.frameQueue.Head()
+		}
+
+		if bytesRead > len(p) {
+			return bytesRead, fmt.Errorf("BUG: bytesRead (%d) > len(p) (%d) in stream.Read", bytesRead, len(p))
+		}
+		if s.readPosInFrame > int(frame.DataLen()) {
+			return bytesRead, fmt.Errorf("BUG: readPosInFrame (%d) > frame.DataLen (%d) in stream.Read", s.readPosInFrame, frame.DataLen())
+		}
+
+		s.mutex.Unlock()
+
+		copy(p[bytesRead:], frame.Data[s.readPosInFrame:])
+		m := utils.Min(len(p)-bytesRead, int(frame.DataLen())-s.readPosInFrame)
+		s.readPosInFrame += m
+		bytesRead += m
+		s.readOffset += protocol.ByteCount(m)
+
+		s.mutex.Lock()
+		// when a RST_STREAM was received, the was already informed about the final byteOffset for this stream
+		if !s.resetRemotely {
+			s.flowController.AddBytesRead(protocol.ByteCount(m))
+		}
+		s.onData() // so that a possible WINDOW_UPDATE is sent
+
+		if s.readPosInFrame >= int(frame.DataLen()) {
+			s.frameQueue.Pop()
+			s.finRead = frame.FinBit
+			if frame.FinBit {
+				return bytesRead, io.EOF
+			}
+		}
+	}
+	return bytesRead, nil
+}
+
+func (s *receiveStream) CancelRead(errorCode protocol.ApplicationErrorCode) error {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	if s.finRead {
+		return nil
+	}
+	if s.canceledRead {
+		return nil
+	}
+	s.canceledRead = true
+	s.cancelReadErr = fmt.Errorf("Read on stream %d canceled with error code %d", s.streamID, errorCode)
+	s.signalRead()
+	if s.version.UsesIETFFrameFormat() {
+		s.queueControlFrame(&wire.StopSendingFrame{
+			StreamID:  s.streamID,
+			ErrorCode: errorCode,
+		})
+	}
+	return nil
+}
+
+func (s *receiveStream) HandleStreamFrame(frame *wire.StreamFrame) error {
+	maxOffset := frame.Offset + frame.DataLen()
+	if err := s.flowController.UpdateHighestReceived(maxOffset, frame.FinBit); err != nil {
+		return err
+	}
+
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+	if err := s.frameQueue.Push(frame); err != nil && err != errDuplicateStreamData {
+		return err
+	}
+	s.signalRead()
+	return nil
+}
+
+func (s *receiveStream) HandleRstStreamFrame(frame *wire.RstStreamFrame) error {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	if s.closedForShutdown {
+		return nil
+	}
+	if err := s.flowController.UpdateHighestReceived(frame.ByteOffset, true); err != nil {
+		return err
+	}
+	// In gQUIC, error code 0 has a special meaning.
+	// The peer will reliably continue transmitting, but is not interested in reading from the stream.
+	// We should therefore just continue reading from the stream, until we encounter the FIN bit.
+	if !s.version.UsesIETFFrameFormat() && frame.ErrorCode == 0 {
+		return nil
+	}
+
+	// ignore duplicate RST_STREAM frames for this stream (after checking their final offset)
+	if s.resetRemotely {
+		return nil
+	}
+	s.resetRemotely = true
+	s.resetRemotelyErr = streamCanceledError{
+		errorCode: frame.ErrorCode,
+		error:     fmt.Errorf("Stream %d was reset with error code %d", s.streamID, frame.ErrorCode),
+	}
+	s.signalRead()
+	return nil
+}
+
+func (s *receiveStream) CloseRemote(offset protocol.ByteCount) {
+	s.HandleStreamFrame(&wire.StreamFrame{FinBit: true, Offset: offset})
+}
+
+func (s *receiveStream) onClose(offset protocol.ByteCount) {
+	if s.canceledRead && !s.version.UsesIETFFrameFormat() {
+		s.queueControlFrame(&wire.RstStreamFrame{
+			StreamID:   s.streamID,
+			ByteOffset: offset,
+			ErrorCode:  0,
+		})
+		s.onData()
+	}
+}
+
+func (s *receiveStream) SetReadDeadline(t time.Time) error {
+	s.mutex.Lock()
+	oldDeadline := s.readDeadline
+	s.readDeadline = t
+	s.mutex.Unlock()
+	// if the new deadline is before the currently set deadline, wake up Read()
+	if t.Before(oldDeadline) {
+		s.signalRead()
+	}
+	return nil
+}
+
+// CloseForShutdown closes a stream abruptly.
+// It makes Read 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 *receiveStream) CloseForShutdown(err error) {
+	s.mutex.Lock()
+	s.closedForShutdown = true
+	s.closeForShutdownErr = err
+	s.mutex.Unlock()
+	s.signalRead()
+}
+
+func (s *receiveStream) Finished() bool {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	return s.closedForShutdown || // if the stream was abruptly closed for shutting down
+		s.finRead || s.resetRemotely
+}
+
+func (s *receiveStream) GetWindowUpdate() protocol.ByteCount {
+	return s.flowController.GetWindowUpdate()
+}
+
+// signalRead performs a non-blocking send on the readChan
+func (s *receiveStream) signalRead() {
+	select {
+	case s.readChan <- struct{}{}:
+	default:
+	}
+}

receive_stream_test.go

@@ -0,0 +1,662 @@
+package quic
+
+import (
+	"errors"
+	"io"
+	"runtime"
+	"time"
+
+	"github.com/lucas-clemente/quic-go/internal/mocks"
+	"github.com/lucas-clemente/quic-go/internal/protocol"
+	"github.com/lucas-clemente/quic-go/internal/wire"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	"github.com/onsi/gomega/gbytes"
+)
+
+var _ = Describe("Receive Stream", func() {
+	const streamID protocol.StreamID = 1337
+
+	var (
+		str                 *receiveStream
+		strWithTimeout      io.Reader // str wrapped with gbytes.TimeoutReader
+		onDataCalled        bool
+		queuedControlFrames []wire.Frame
+		mockFC              *mocks.MockStreamFlowController
+	)
+
+	onData := func() { onDataCalled = true }
+	queueControlFrame := func(f wire.Frame) { queuedControlFrames = append(queuedControlFrames, f) }
+
+	BeforeEach(func() {
+		queuedControlFrames = queuedControlFrames[:0]
+		onDataCalled = false
+		mockFC = mocks.NewMockStreamFlowController(mockCtrl)
+		str = newReceiveStream(streamID, onData, queueControlFrame, mockFC)
+
+		timeout := scaleDuration(250 * time.Millisecond)
+		strWithTimeout = gbytes.TimeoutReader(str, timeout)
+	})
+
+	It("gets stream id", func() {
+		Expect(str.StreamID()).To(Equal(protocol.StreamID(1337)))
+	})
+
+	Context("reading", func() {
+		It("reads a single STREAM frame", func() {
+			mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(4), false)
+			mockFC.EXPECT().AddBytesRead(protocol.ByteCount(4))
+			frame := wire.StreamFrame{
+				Offset: 0,
+				Data:   []byte{0xDE, 0xAD, 0xBE, 0xEF},
+			}
+			err := str.HandleStreamFrame(&frame)
+			Expect(err).ToNot(HaveOccurred())
+			b := make([]byte, 4)
+			n, err := strWithTimeout.Read(b)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(n).To(Equal(4))
+			Expect(b).To(Equal([]byte{0xDE, 0xAD, 0xBE, 0xEF}))
+		})
+
+		It("reads a single STREAM frame in multiple goes", func() {
+			mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(4), false)
+			mockFC.EXPECT().AddBytesRead(protocol.ByteCount(2))
+			mockFC.EXPECT().AddBytesRead(protocol.ByteCount(2))
+			frame := wire.StreamFrame{
+				Offset: 0,
+				Data:   []byte{0xDE, 0xAD, 0xBE, 0xEF},
+			}
+			err := str.HandleStreamFrame(&frame)
+			Expect(err).ToNot(HaveOccurred())
+			b := make([]byte, 2)
+			n, err := strWithTimeout.Read(b)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(n).To(Equal(2))
+			Expect(b).To(Equal([]byte{0xDE, 0xAD}))
+			n, err = strWithTimeout.Read(b)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(n).To(Equal(2))
+			Expect(b).To(Equal([]byte{0xBE, 0xEF}))
+		})
+
+		It("reads all data available", func() {
+			mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(2), false)
+			mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(4), false)
+			mockFC.EXPECT().AddBytesRead(protocol.ByteCount(2)).Times(2)
+			frame1 := wire.StreamFrame{
+				Offset: 0,
+				Data:   []byte{0xDE, 0xAD},
+			}
+			frame2 := wire.StreamFrame{
+				Offset: 2,
+				Data:   []byte{0xBE, 0xEF},
+			}
+			err := str.HandleStreamFrame(&frame1)
+			Expect(err).ToNot(HaveOccurred())
+			err = str.HandleStreamFrame(&frame2)
+			Expect(err).ToNot(HaveOccurred())
+			b := make([]byte, 6)
+			n, err := strWithTimeout.Read(b)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(n).To(Equal(4))
+			Expect(b).To(Equal([]byte{0xDE, 0xAD, 0xBE, 0xEF, 0x00, 0x00}))
+		})
+
+		It("assembles multiple StreamFrames", func() {
+			mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(2), false)
+			mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(4), false)
+			mockFC.EXPECT().AddBytesRead(protocol.ByteCount(2)).Times(2)
+			frame1 := wire.StreamFrame{
+				Offset: 0,
+				Data:   []byte{0xDE, 0xAD},
+			}
+			frame2 := wire.StreamFrame{
+				Offset: 2,
+				Data:   []byte{0xBE, 0xEF},
+			}
+			err := str.HandleStreamFrame(&frame1)
+			Expect(err).ToNot(HaveOccurred())
+			err = str.HandleStreamFrame(&frame2)
+			Expect(err).ToNot(HaveOccurred())
+			b := make([]byte, 4)
+			n, err := strWithTimeout.Read(b)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(n).To(Equal(4))
+			Expect(b).To(Equal([]byte{0xDE, 0xAD, 0xBE, 0xEF}))
+		})
+
+		It("waits until data is available", func() {
+			mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(2), false)
+			mockFC.EXPECT().AddBytesRead(protocol.ByteCount(2))
+			go func() {
+				defer GinkgoRecover()
+				frame := wire.StreamFrame{Data: []byte{0xDE, 0xAD}}
+				time.Sleep(10 * time.Millisecond)
+				err := str.HandleStreamFrame(&frame)
+				Expect(err).ToNot(HaveOccurred())
+			}()
+			b := make([]byte, 2)
+			n, err := strWithTimeout.Read(b)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(n).To(Equal(2))
+		})
+
+		It("handles STREAM frames in wrong order", func() {
+			mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(2), false)
+			mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(4), false)
+			mockFC.EXPECT().AddBytesRead(protocol.ByteCount(2)).Times(2)
+			frame1 := wire.StreamFrame{
+				Offset: 2,
+				Data:   []byte{0xBE, 0xEF},
+			}
+			frame2 := wire.StreamFrame{
+				Offset: 0,
+				Data:   []byte{0xDE, 0xAD},
+			}
+			err := str.HandleStreamFrame(&frame1)
+			Expect(err).ToNot(HaveOccurred())
+			err = str.HandleStreamFrame(&frame2)
+			Expect(err).ToNot(HaveOccurred())
+			b := make([]byte, 4)
+			n, err := strWithTimeout.Read(b)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(n).To(Equal(4))
+			Expect(b).To(Equal([]byte{0xDE, 0xAD, 0xBE, 0xEF}))
+		})
+
+		It("ignores duplicate STREAM frames", func() {
+			mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(2), false)
+			mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(2), false)
+			mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(4), false)
+			mockFC.EXPECT().AddBytesRead(protocol.ByteCount(2)).Times(2)
+			frame1 := wire.StreamFrame{
+				Offset: 0,
+				Data:   []byte{0xDE, 0xAD},
+			}
+			frame2 := wire.StreamFrame{
+				Offset: 0,
+				Data:   []byte{0x13, 0x37},
+			}
+			frame3 := wire.StreamFrame{
+				Offset: 2,
+				Data:   []byte{0xBE, 0xEF},
+			}
+			err := str.HandleStreamFrame(&frame1)
+			Expect(err).ToNot(HaveOccurred())
+			err = str.HandleStreamFrame(&frame2)
+			Expect(err).ToNot(HaveOccurred())
+			err = str.HandleStreamFrame(&frame3)
+			Expect(err).ToNot(HaveOccurred())
+			b := make([]byte, 4)
+			n, err := strWithTimeout.Read(b)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(n).To(Equal(4))
+			Expect(b).To(Equal([]byte{0xDE, 0xAD, 0xBE, 0xEF}))
+		})
+
+		It("doesn't rejects a STREAM frames with an overlapping data range", func() {
+			mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(4), false)
+			mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(6), false)
+			mockFC.EXPECT().AddBytesRead(protocol.ByteCount(2))
+			mockFC.EXPECT().AddBytesRead(protocol.ByteCount(4))
+			frame1 := wire.StreamFrame{
+				Offset: 0,
+				Data:   []byte("foob"),
+			}
+			frame2 := wire.StreamFrame{
+				Offset: 2,
+				Data:   []byte("obar"),
+			}
+			err := str.HandleStreamFrame(&frame1)
+			Expect(err).ToNot(HaveOccurred())
+			err = str.HandleStreamFrame(&frame2)
+			Expect(err).ToNot(HaveOccurred())
+			b := make([]byte, 6)
+			n, err := strWithTimeout.Read(b)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(n).To(Equal(6))
+			Expect(b).To(Equal([]byte("foobar")))
+		})
+
+		It("calls onData", func() {
+			mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(4), false)
+			mockFC.EXPECT().AddBytesRead(protocol.ByteCount(4))
+			frame := wire.StreamFrame{
+				Offset: 0,
+				Data:   []byte{0xDE, 0xAD, 0xBE, 0xEF},
+			}
+			err := str.HandleStreamFrame(&frame)
+			Expect(err).ToNot(HaveOccurred())
+			b := make([]byte, 4)
+			_, err = strWithTimeout.Read(b)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(onDataCalled).To(BeTrue())
+		})
+
+		It("passes on errors from the streamFrameSorter", func() {
+			mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(0), false)
+			err := str.HandleStreamFrame(&wire.StreamFrame{StreamID: streamID}) // STREAM frame without data
+			Expect(err).To(MatchError(errEmptyStreamData))
+		})
+
+		Context("deadlines", func() {
+			It("the deadline error has the right net.Error properties", func() {
+				Expect(errDeadline.Temporary()).To(BeTrue())
+				Expect(errDeadline.Timeout()).To(BeTrue())
+				Expect(errDeadline).To(MatchError("deadline exceeded"))
+			})
+
+			It("returns an error when Read is called after the deadline", func() {
+				mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(6), false).AnyTimes()
+				f := &wire.StreamFrame{Data: []byte("foobar")}
+				err := str.HandleStreamFrame(f)
+				Expect(err).ToNot(HaveOccurred())
+				str.SetReadDeadline(time.Now().Add(-time.Second))
+				b := make([]byte, 6)
+				n, err := strWithTimeout.Read(b)
+				Expect(err).To(MatchError(errDeadline))
+				Expect(n).To(BeZero())
+			})
+
+			It("unblocks after the deadline", func() {
+				deadline := time.Now().Add(scaleDuration(50 * time.Millisecond))
+				str.SetReadDeadline(deadline)
+				b := make([]byte, 6)
+				n, err := strWithTimeout.Read(b)
+				Expect(err).To(MatchError(errDeadline))
+				Expect(n).To(BeZero())
+				Expect(time.Now()).To(BeTemporally("~", deadline, scaleDuration(10*time.Millisecond)))
+			})
+
+			It("doesn't unblock if the deadline is changed before the first one expires", func() {
+				deadline1 := time.Now().Add(scaleDuration(50 * time.Millisecond))
+				deadline2 := time.Now().Add(scaleDuration(100 * time.Millisecond))
+				str.SetReadDeadline(deadline1)
+				go func() {
+					defer GinkgoRecover()
+					time.Sleep(scaleDuration(20 * time.Millisecond))
+					str.SetReadDeadline(deadline2)
+					// make sure that this was actually execute before the deadline expires
+					Expect(time.Now()).To(BeTemporally("<", deadline1))
+				}()
+				runtime.Gosched()
+				b := make([]byte, 10)
+				n, err := strWithTimeout.Read(b)
+				Expect(err).To(MatchError(errDeadline))
+				Expect(n).To(BeZero())
+				Expect(time.Now()).To(BeTemporally("~", deadline2, scaleDuration(20*time.Millisecond)))
+			})
+
+			It("unblocks earlier, when a new deadline is set", func() {
+				deadline1 := time.Now().Add(scaleDuration(200 * time.Millisecond))
+				deadline2 := time.Now().Add(scaleDuration(50 * time.Millisecond))
+				go func() {
+					defer GinkgoRecover()
+					time.Sleep(scaleDuration(10 * time.Millisecond))
+					str.SetReadDeadline(deadline2)
+					// make sure that this was actually execute before the deadline expires
+					Expect(time.Now()).To(BeTemporally("<", deadline2))
+				}()
+				str.SetReadDeadline(deadline1)
+				runtime.Gosched()
+				b := make([]byte, 10)
+				_, err := strWithTimeout.Read(b)
+				Expect(err).To(MatchError(errDeadline))
+				Expect(time.Now()).To(BeTemporally("~", deadline2, scaleDuration(25*time.Millisecond)))
+			})
+		})
+
+		Context("closing", func() {
+			Context("with FIN bit", func() {
+				It("returns EOFs", func() {
+					mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(4), true)
+					mockFC.EXPECT().AddBytesRead(protocol.ByteCount(4))
+					frame := wire.StreamFrame{
+						Offset: 0,
+						Data:   []byte{0xDE, 0xAD, 0xBE, 0xEF},
+						FinBit: true,
+					}
+					str.HandleStreamFrame(&frame)
+					b := make([]byte, 4)
+					n, err := strWithTimeout.Read(b)
+					Expect(err).To(MatchError(io.EOF))
+					Expect(n).To(Equal(4))
+					Expect(b).To(Equal([]byte{0xDE, 0xAD, 0xBE, 0xEF}))
+					n, err = strWithTimeout.Read(b)
+					Expect(n).To(BeZero())
+					Expect(err).To(MatchError(io.EOF))
+				})
+
+				It("handles out-of-order frames", func() {
+					mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(2), false)
+					mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(4), true)
+					mockFC.EXPECT().AddBytesRead(protocol.ByteCount(2)).Times(2)
+					frame1 := wire.StreamFrame{
+						Offset: 2,
+						Data:   []byte{0xBE, 0xEF},
+						FinBit: true,
+					}
+					frame2 := wire.StreamFrame{
+						Offset: 0,
+						Data:   []byte{0xDE, 0xAD},
+					}
+					err := str.HandleStreamFrame(&frame1)
+					Expect(err).ToNot(HaveOccurred())
+					err = str.HandleStreamFrame(&frame2)
+					Expect(err).ToNot(HaveOccurred())
+					b := make([]byte, 4)
+					n, err := strWithTimeout.Read(b)
+					Expect(err).To(MatchError(io.EOF))
+					Expect(n).To(Equal(4))
+					Expect(b).To(Equal([]byte{0xDE, 0xAD, 0xBE, 0xEF}))
+					n, err = strWithTimeout.Read(b)
+					Expect(n).To(BeZero())
+					Expect(err).To(MatchError(io.EOF))
+				})
+
+				It("returns EOFs with partial read", func() {
+					mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(2), true)
+					mockFC.EXPECT().AddBytesRead(protocol.ByteCount(2))
+					frame := wire.StreamFrame{
+						Offset: 0,
+						Data:   []byte{0xDE, 0xAD},
+						FinBit: true,
+					}
+					err := str.HandleStreamFrame(&frame)
+					Expect(err).ToNot(HaveOccurred())
+					b := make([]byte, 4)
+					n, err := strWithTimeout.Read(b)
+					Expect(err).To(MatchError(io.EOF))
+					Expect(n).To(Equal(2))
+					Expect(b[:n]).To(Equal([]byte{0xDE, 0xAD}))
+				})
+
+				It("handles immediate FINs", func() {
+					mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(0), true)
+					mockFC.EXPECT().AddBytesRead(protocol.ByteCount(0))
+					frame := wire.StreamFrame{
+						Offset: 0,
+						Data:   []byte{},
+						FinBit: true,
+					}
+					err := str.HandleStreamFrame(&frame)
+					Expect(err).ToNot(HaveOccurred())
+					b := make([]byte, 4)
+					n, err := strWithTimeout.Read(b)
+					Expect(n).To(BeZero())
+					Expect(err).To(MatchError(io.EOF))
+				})
+			})
+
+			It("closes when CloseRemote is called", func() {
+				mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(0), true)
+				mockFC.EXPECT().AddBytesRead(protocol.ByteCount(0))
+				str.CloseRemote(0)
+				b := make([]byte, 8)
+				n, err := strWithTimeout.Read(b)
+				Expect(n).To(BeZero())
+				Expect(err).To(MatchError(io.EOF))
+			})
+		})
+
+		Context("closing for shutdown", func() {
+			testErr := errors.New("test error")
+
+			It("immediately returns all reads", func() {
+				done := make(chan struct{})
+				b := make([]byte, 4)
+				go func() {
+					defer GinkgoRecover()
+					n, err := strWithTimeout.Read(b)
+					Expect(n).To(BeZero())
+					Expect(err).To(MatchError(testErr))
+					close(done)
+				}()
+				Consistently(done).ShouldNot(BeClosed())
+				str.CloseForShutdown(testErr)
+				Eventually(done).Should(BeClosed())
+			})
+
+			It("errors for all following reads", func() {
+				str.CloseForShutdown(testErr)
+				b := make([]byte, 1)
+				n, err := strWithTimeout.Read(b)
+				Expect(n).To(BeZero())
+				Expect(err).To(MatchError(testErr))
+			})
+		})
+	})
+
+	Context("stream cancelations", func() {
+		Context("canceling read", func() {
+			It("unblocks Read", func() {
+				done := make(chan struct{})
+				go func() {
+					defer GinkgoRecover()
+					_, err := strWithTimeout.Read([]byte{0})
+					Expect(err).To(MatchError("Read on stream 1337 canceled with error code 1234"))
+					close(done)
+				}()
+				Consistently(done).ShouldNot(BeClosed())
+				err := str.CancelRead(1234)
+				Expect(err).ToNot(HaveOccurred())
+				Eventually(done).Should(BeClosed())
+			})
+
+			It("doesn't allow further calls to Read", func() {
+				err := str.CancelRead(1234)
+				Expect(err).ToNot(HaveOccurred())
+				_, err = strWithTimeout.Read([]byte{0})
+				Expect(err).To(MatchError("Read on stream 1337 canceled with error code 1234"))
+			})
+
+			It("does nothing when CancelRead is called twice", func() {
+				err := str.CancelRead(1234)
+				Expect(err).ToNot(HaveOccurred())
+				err = str.CancelRead(2345)
+				Expect(err).ToNot(HaveOccurred())
+				_, err = strWithTimeout.Read([]byte{0})
+				Expect(err).To(MatchError("Read on stream 1337 canceled with error code 1234"))
+			})
+
+			It("doesn't send a RST_STREAM frame, if the FIN was already read", func() {
+				mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(6), true)
+				mockFC.EXPECT().AddBytesRead(protocol.ByteCount(6))
+				err := str.HandleStreamFrame(&wire.StreamFrame{
+					StreamID: streamID,
+					Data:     []byte("foobar"),
+					FinBit:   true,
+				})
+				Expect(err).ToNot(HaveOccurred())
+				_, err = strWithTimeout.Read(make([]byte, 100))
+				Expect(err).To(MatchError(io.EOF))
+				err = str.CancelRead(1234)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(queuedControlFrames).To(BeEmpty()) // no RST_STREAM frame queued yet
+			})
+
+			Context("for IETF QUIC", func() {
+				It("queues a STOP_SENDING frame", func() {
+					err := str.CancelRead(1234)
+					Expect(err).ToNot(HaveOccurred())
+					Expect(queuedControlFrames).To(Equal([]wire.Frame{
+						&wire.StopSendingFrame{
+							StreamID:  streamID,
+							ErrorCode: 1234,
+						},
+					}))
+				})
+			})
+		})
+
+		Context("receiving RST_STREAM frames", func() {
+			rst := &wire.RstStreamFrame{
+				StreamID:   streamID,
+				ByteOffset: 42,
+				ErrorCode:  1234,
+			}
+
+			It("unblocks Read", func() {
+				mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(42), true)
+				done := make(chan struct{})
+				go func() {
+					defer GinkgoRecover()
+					_, err := strWithTimeout.Read([]byte{0})
+					Expect(err).To(MatchError("Stream 1337 was reset with error code 1234"))
+					Expect(err).To(BeAssignableToTypeOf(streamCanceledError{}))
+					Expect(err.(streamCanceledError).Canceled()).To(BeTrue())
+					Expect(err.(streamCanceledError).ErrorCode()).To(Equal(protocol.ApplicationErrorCode(1234)))
+					close(done)
+				}()
+				Consistently(done).ShouldNot(BeClosed())
+				str.HandleRstStreamFrame(rst)
+				Eventually(done).Should(BeClosed())
+			})
+
+			It("doesn't allow further calls to Read", func() {
+				mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(42), true)
+				err := str.HandleRstStreamFrame(rst)
+				Expect(err).ToNot(HaveOccurred())
+				_, err = strWithTimeout.Read([]byte{0})
+				Expect(err).To(MatchError("Stream 1337 was reset with error code 1234"))
+				Expect(err).To(BeAssignableToTypeOf(streamCanceledError{}))
+				Expect(err.(streamCanceledError).Canceled()).To(BeTrue())
+				Expect(err.(streamCanceledError).ErrorCode()).To(Equal(protocol.ApplicationErrorCode(1234)))
+			})
+
+			It("errors when receiving a RST_STREAM with an inconsistent offset", func() {
+				testErr := errors.New("already received a different final offset before")
+				mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(42), true).Return(testErr)
+				err := str.HandleRstStreamFrame(rst)
+				Expect(err).To(MatchError(testErr))
+			})
+
+			It("ignores duplicate RST_STREAM frames", func() {
+				mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(42), true).Times(2)
+				err := str.HandleRstStreamFrame(rst)
+				Expect(err).ToNot(HaveOccurred())
+				err = str.HandleRstStreamFrame(rst)
+				Expect(err).ToNot(HaveOccurred())
+			})
+
+			It("doesn't do anyting when it was closed for shutdown", func() {
+				str.CloseForShutdown(nil)
+				err := str.HandleRstStreamFrame(rst)
+				Expect(err).ToNot(HaveOccurred())
+			})
+
+			Context("for gQUIC", func() {
+				BeforeEach(func() {
+					str.version = versionGQUICFrames
+				})
+
+				It("unblocks Read when receiving a RST_STREAM frame with non-zero error code", func() {
+					mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(42), true)
+					readReturned := make(chan struct{})
+					go func() {
+						defer GinkgoRecover()
+						_, err := strWithTimeout.Read([]byte{0})
+						Expect(err).To(MatchError("Stream 1337 was reset with error code 1234"))
+						Expect(err).To(BeAssignableToTypeOf(streamCanceledError{}))
+						Expect(err.(streamCanceledError).Canceled()).To(BeTrue())
+						Expect(err.(streamCanceledError).ErrorCode()).To(Equal(protocol.ApplicationErrorCode(1234)))
+						close(readReturned)
+					}()
+					Consistently(readReturned).ShouldNot(BeClosed())
+					err := str.HandleRstStreamFrame(rst)
+					Expect(err).ToNot(HaveOccurred())
+					Eventually(readReturned).Should(BeClosed())
+				})
+
+				It("sends a RST_STREAM and continues reading until the end when receiving a RST_STREAM frame with error code 0", func() {
+					mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(6), true).Times(2)
+					mockFC.EXPECT().AddBytesRead(protocol.ByteCount(4))
+					mockFC.EXPECT().AddBytesRead(protocol.ByteCount(2))
+					readReturned := make(chan struct{})
+					go func() {
+						defer GinkgoRecover()
+						n, err := strWithTimeout.Read(make([]byte, 4))
+						Expect(err).ToNot(HaveOccurred())
+						Expect(n).To(Equal(4))
+						n, err = strWithTimeout.Read(make([]byte, 4))
+						Expect(err).To(MatchError(io.EOF))
+						Expect(n).To(Equal(2))
+						close(readReturned)
+					}()
+					Consistently(readReturned).ShouldNot(BeClosed())
+					err := str.HandleStreamFrame(&wire.StreamFrame{
+						StreamID: streamID,
+						Data:     []byte("foobar"),
+						FinBit:   true,
+					})
+					Expect(err).ToNot(HaveOccurred())
+					err = str.HandleRstStreamFrame(&wire.RstStreamFrame{
+						StreamID:   streamID,
+						ByteOffset: 6,
+						ErrorCode:  0,
+					})
+					Expect(err).ToNot(HaveOccurred())
+					Eventually(readReturned).Should(BeClosed())
+				})
+
+			})
+		})
+	})
+
+	Context("flow control", func() {
+		It("errors when a STREAM frame causes a flow control violation", func() {
+			testErr := errors.New("flow control violation")
+			mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(8), false).Return(testErr)
+			frame := wire.StreamFrame{
+				Offset: 2,
+				Data:   []byte("foobar"),
+			}
+			err := str.HandleStreamFrame(&frame)
+			Expect(err).To(MatchError(testErr))
+		})
+
+		It("gets a window update", func() {
+			mockFC.EXPECT().GetWindowUpdate().Return(protocol.ByteCount(0x100))
+			Expect(str.GetWindowUpdate()).To(Equal(protocol.ByteCount(0x100)))
+		})
+	})
+
+	Context("saying if it is finished", func() {
+		finishReading := func() {
+			mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(0), true)
+			err := str.HandleStreamFrame(&wire.StreamFrame{FinBit: true})
+			Expect(err).ToNot(HaveOccurred())
+			b := make([]byte, 100)
+			_, err = strWithTimeout.Read(b)
+			ExpectWithOffset(0, err).To(MatchError(io.EOF))
+		}
+
+		It("is finished after it is closed for shutdown", func() {
+			str.CloseForShutdown(errors.New("testErr"))
+			Expect(str.Finished()).To(BeTrue())
+		})
+
+		It("is finished if it is only closed for reading", func() {
+			mockFC.EXPECT().AddBytesRead(protocol.ByteCount(0))
+			finishReading()
+			Expect(str.Finished()).To(BeTrue())
+		})
+
+		// the stream still needs to stay alive until we receive the final offset
+		// (either by receiving a STREAM frame with FIN, or a RST_STREAM)
+		It("is not finished after CancelRead", func() {
+			err := str.CancelRead(123)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(str.Finished()).To(BeFalse())
+		})
+
+		It("is finished after receiving a RST_STREAM frame", func() {
+			mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(3), true)
+			err := str.HandleRstStreamFrame(&wire.RstStreamFrame{ByteOffset: 3})
+			Expect(err).ToNot(HaveOccurred())
+			Expect(str.Finished()).To(BeTrue())
+		})
+	})
+})

send_stream.go

@@ -0,0 +1,293 @@
+package quic
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/lucas-clemente/quic-go/internal/flowcontrol"
+	"github.com/lucas-clemente/quic-go/internal/protocol"
+	"github.com/lucas-clemente/quic-go/internal/utils"
+	"github.com/lucas-clemente/quic-go/internal/wire"
+)
+
+type sendStream struct {
+	mutex sync.Mutex
+
+	ctx       context.Context
+	ctxCancel context.CancelFunc
+
+	streamID protocol.StreamID
+	// onData tells the session that there's stuff to pack into a new packet
+	onData func()
+	// queueControlFrame queues a new control frame for sending
+	// it does not call onData
+	queueControlFrame func(wire.Frame)
+
+	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 b
+
+	dataForWriting []byte
+	writeChan      chan struct{}
+	writeDeadline  time.Time
+
+	flowController flowcontrol.StreamFlowController
+	version        protocol.VersionNumber
+}
+
+var _ SendStream = &sendStream{}
+
+func newSendStream(
+	streamID protocol.StreamID,
+	onData func(),
+	queueControlFrame func(wire.Frame),
+	flowController flowcontrol.StreamFlowController,
+	version protocol.VersionNumber,
+) *sendStream {
+	s := &sendStream{
+		streamID:          streamID,
+		onData:            onData,
+		queueControlFrame: queueControlFrame,
+		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 len(p) == 0 {
+		return 0, nil
+	}
+
+	s.dataForWriting = make([]byte, len(p))
+	copy(s.dataForWriting, p)
+	s.onData()
+
+	var err error
+	for {
+		deadline := s.writeDeadline
+		if !deadline.IsZero() && !time.Now().Before(deadline) {
+			err = errDeadline
+			break
+		}
+		if s.dataForWriting == nil || s.canceledWrite || s.closedForShutdown {
+			break
+		}
+
+		s.mutex.Unlock()
+		if deadline.IsZero() {
+			<-s.writeChan
+		} else {
+			select {
+			case <-s.writeChan:
+			case <-time.After(deadline.Sub(time.Now())):
+			}
+		}
+		s.mutex.Lock()
+	}
+
+	if s.closeForShutdownErr != nil {
+		err = s.closeForShutdownErr
+	} else if s.cancelWriteErr != nil {
+		err = s.cancelWriteErr
+	}
+	return len(p) - len(s.dataForWriting), err
+}
+
+// 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) *wire.StreamFrame {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	if s.closeForShutdownErr != nil {
+		return nil
+	}
+
+	frame := &wire.StreamFrame{
+		StreamID:       s.streamID,
+		Offset:         s.writeOffset,
+		DataLenPresent: true,
+	}
+	frameLen := frame.MinLength(s.version)
+	if frameLen >= maxBytes { // a STREAM frame must have at least one byte of data
+		return nil
+	}
+	frame.Data, frame.FinBit = s.getDataForWriting(maxBytes - frameLen)
+	if len(frame.Data) == 0 && !frame.FinBit {
+		return nil
+	}
+	if frame.FinBit {
+		s.finSent = true
+	}
+	return frame
+}
+
+func (s *sendStream) getDataForWriting(maxBytes protocol.ByteCount) ([]byte, bool /* should send FIN */) {
+	if s.dataForWriting == nil {
+		return nil, s.finishedWriting && !s.finSent
+	}
+
+	// TODO(#657): Flow control for the crypto stream
+	if s.streamID != s.version.CryptoStreamID() {
+		maxBytes = utils.MinByteCount(maxBytes, s.flowController.SendWindowSize())
+	}
+	if maxBytes == 0 {
+		return nil, false
+	}
+
+	var ret []byte
+	if protocol.ByteCount(len(s.dataForWriting)) > maxBytes {
+		ret = s.dataForWriting[:maxBytes]
+		s.dataForWriting = s.dataForWriting[maxBytes:]
+	} else {
+		ret = s.dataForWriting
+		s.dataForWriting = nil
+		s.signalWrite()
+	}
+	s.writeOffset += protocol.ByteCount(len(ret))
+	s.flowController.AddBytesSent(protocol.ByteCount(len(ret)))
+	return ret, s.finishedWriting && s.dataForWriting == nil && !s.finSent
+}
+
+func (s *sendStream) Close() error {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	if s.canceledWrite {
+		return fmt.Errorf("Close called for canceled stream %d", s.streamID)
+	}
+	s.finishedWriting = true
+	s.ctxCancel()
+	s.onData()
+	return nil
+}
+
+func (s *sendStream) CancelWrite(errorCode protocol.ApplicationErrorCode) error {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	return 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 protocol.ApplicationErrorCode, writeErr error) error {
+	if s.canceledWrite {
+		return nil
+	}
+	if s.finishedWriting {
+		return fmt.Errorf("CancelWrite for closed stream %d", s.streamID)
+	}
+	s.canceledWrite = true
+	s.cancelWriteErr = writeErr
+	s.signalWrite()
+	s.queueControlFrame(&wire.RstStreamFrame{
+		StreamID:   s.streamID,
+		ByteOffset: s.writeOffset,
+		ErrorCode:  errorCode,
+	})
+	// TODO(#991): cancel retransmissions for this stream
+	s.onData()
+	s.ctxCancel()
+	return nil
+}
+
+func (s *sendStream) HandleStopSendingFrame(frame *wire.StopSendingFrame) {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+	s.handleStopSendingFrameImpl(frame)
+}
+
+func (s *sendStream) HandleMaxStreamDataFrame(frame *wire.MaxStreamDataFrame) {
+	s.flowController.UpdateSendWindow(frame.ByteOffset)
+}
+
+func (s *sendStream) IsFlowControlBlocked() (bool, protocol.ByteCount) {
+	return s.flowController.IsBlocked()
+}
+
+// must be called after locking the mutex
+func (s *sendStream) handleStopSendingFrameImpl(frame *wire.StopSendingFrame) {
+	writeErr := streamCanceledError{
+		errorCode: frame.ErrorCode,
+		error:     fmt.Errorf("Stream %d was reset with error code %d", s.streamID, frame.ErrorCode),
+	}
+	errorCode := errorCodeStopping
+	if !s.version.UsesIETFFrameFormat() {
+		errorCode = errorCodeStoppingGQUIC
+	}
+	s.cancelWriteImpl(errorCode, writeErr)
+}
+
+func (s *sendStream) Context() context.Context {
+	return s.ctx
+}
+
+func (s *sendStream) SetWriteDeadline(t time.Time) error {
+	s.mutex.Lock()
+	oldDeadline := s.writeDeadline
+	s.writeDeadline = t
+	s.mutex.Unlock()
+	if t.Before(oldDeadline) {
+		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.closedForShutdown = true
+	s.closeForShutdownErr = err
+	s.mutex.Unlock()
+	s.signalWrite()
+	s.ctxCancel()
+}
+
+func (s *sendStream) Finished() bool {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	return s.closedForShutdown || // if the stream was abruptly closed for shutting down
+		s.finSent || s.canceledWrite
+}
+
+func (s *sendStream) getWriteOffset() protocol.ByteCount {
+	return s.writeOffset
+}
+
+// signalWrite performs a non-blocking send on the writeChan
+func (s *sendStream) signalWrite() {
+	select {
+	case s.writeChan <- struct{}{}:
+	default:
+	}
+}

send_stream_test.go

@@ -0,0 +1,468 @@
+package quic
+
+import (
+	"bytes"
+	"errors"
+	"io"
+	"runtime"
+	"time"
+
+	"github.com/golang/mock/gomock"
+	"github.com/lucas-clemente/quic-go/internal/mocks"
+	"github.com/lucas-clemente/quic-go/internal/protocol"
+	"github.com/lucas-clemente/quic-go/internal/wire"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	"github.com/onsi/gomega/gbytes"
+)
+
+var _ = Describe("Send Stream", func() {
+	const streamID protocol.StreamID = 1337
+
+	var (
+		str                 *sendStream
+		strWithTimeout      io.Writer // str wrapped with gbytes.TimeoutWriter
+		onDataCalled        bool
+		queuedControlFrames []wire.Frame
+		mockFC              *mocks.MockStreamFlowController
+	)
+
+	onData := func() { onDataCalled = true }
+	queueControlFrame := func(f wire.Frame) { queuedControlFrames = append(queuedControlFrames, f) }
+
+	BeforeEach(func() {
+		queuedControlFrames = queuedControlFrames[:0]
+		onDataCalled = false
+		mockFC = mocks.NewMockStreamFlowController(mockCtrl)
+		str = newSendStream(streamID, onData, queueControlFrame, mockFC, protocol.VersionWhatever)
+
+		timeout := scaleDuration(250 * time.Millisecond)
+		strWithTimeout = gbytes.TimeoutWriter(str, timeout)
+	})
+
+	It("gets stream id", func() {
+		Expect(str.StreamID()).To(Equal(protocol.StreamID(1337)))
+	})
+
+	Context("writing", func() {
+		It("writes and gets all data at once", func() {
+			mockFC.EXPECT().SendWindowSize().Return(protocol.ByteCount(9999))
+			mockFC.EXPECT().AddBytesSent(protocol.ByteCount(6))
+			done := make(chan struct{})
+			go func() {
+				defer GinkgoRecover()
+				n, err := strWithTimeout.Write([]byte("foobar"))
+				Expect(err).ToNot(HaveOccurred())
+				Expect(n).To(Equal(6))
+				close(done)
+			}()
+			Eventually(func() []byte {
+				str.mutex.Lock()
+				defer str.mutex.Unlock()
+				return str.dataForWriting
+			}).Should(Equal([]byte("foobar")))
+			Consistently(done).ShouldNot(BeClosed())
+			Expect(onDataCalled).To(BeTrue())
+			f := str.PopStreamFrame(1000)
+			Expect(f.Data).To(Equal([]byte("foobar")))
+			Expect(f.FinBit).To(BeFalse())
+			Expect(f.Offset).To(BeZero())
+			Expect(f.DataLenPresent).To(BeTrue())
+			Expect(str.writeOffset).To(Equal(protocol.ByteCount(6)))
+			Expect(str.dataForWriting).To(BeNil())
+			Eventually(done).Should(BeClosed())
+		})
+
+		It("writes and gets data in two turns", func() {
+			frameHeaderLen := protocol.ByteCount(4)
+			mockFC.EXPECT().SendWindowSize().Return(protocol.ByteCount(9999)).Times(2)
+			mockFC.EXPECT().AddBytesSent(gomock.Any() /* protocol.ByteCount(3)*/).Times(2)
+			done := make(chan struct{})
+			go func() {
+				defer GinkgoRecover()
+				n, err := strWithTimeout.Write([]byte("foobar"))
+				Expect(err).ToNot(HaveOccurred())
+				Expect(n).To(Equal(6))
+				close(done)
+			}()
+			Eventually(func() []byte {
+				str.mutex.Lock()
+				defer str.mutex.Unlock()
+				return str.dataForWriting
+			}).Should(Equal([]byte("foobar")))
+			Consistently(done).ShouldNot(BeClosed())
+			f := str.PopStreamFrame(3 + frameHeaderLen)
+			Expect(f.Data).To(Equal([]byte("foo")))
+			Expect(f.FinBit).To(BeFalse())
+			Expect(f.Offset).To(BeZero())
+			Expect(f.DataLenPresent).To(BeTrue())
+			f = str.PopStreamFrame(100)
+			Expect(f.Data).To(Equal([]byte("bar")))
+			Expect(f.FinBit).To(BeFalse())
+			Expect(f.Offset).To(Equal(protocol.ByteCount(3)))
+			Expect(f.DataLenPresent).To(BeTrue())
+			Expect(str.PopStreamFrame(1000)).To(BeNil())
+			Eventually(done).Should(BeClosed())
+		})
+
+		It("PopStreamFrame returns nil if no data is available", func() {
+			Expect(str.PopStreamFrame(1000)).To(BeNil())
+		})
+
+		It("copies the slice while writing", func() {
+			frameHeaderSize := protocol.ByteCount(4)
+			mockFC.EXPECT().SendWindowSize().Return(protocol.ByteCount(9999)).Times(2)
+			mockFC.EXPECT().AddBytesSent(protocol.ByteCount(1))
+			mockFC.EXPECT().AddBytesSent(protocol.ByteCount(2))
+			s := []byte("foo")
+			go func() {
+				defer GinkgoRecover()
+				n, err := strWithTimeout.Write(s)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(n).To(Equal(3))
+			}()
+			var frame *wire.StreamFrame
+			Eventually(func() *wire.StreamFrame { frame = str.PopStreamFrame(frameHeaderSize + 1); return frame }).ShouldNot(BeNil())
+			Expect(frame.Data).To(Equal([]byte("f")))
+			s[1] = 'e'
+			f := str.PopStreamFrame(100)
+			Expect(f).ToNot(BeNil())
+			Expect(f.Data).To(Equal([]byte("oo")))
+		})
+
+		It("returns when given a nil input", func() {
+			n, err := strWithTimeout.Write(nil)
+			Expect(n).To(BeZero())
+			Expect(err).ToNot(HaveOccurred())
+		})
+
+		It("returns when given an empty slice", func() {
+			n, err := strWithTimeout.Write([]byte(""))
+			Expect(n).To(BeZero())
+			Expect(err).ToNot(HaveOccurred())
+		})
+
+		It("cancels the context when Close is called", func() {
+			Expect(str.Context().Done()).ToNot(BeClosed())
+			str.Close()
+			Expect(str.Context().Done()).To(BeClosed())
+		})
+
+		Context("deadlines", func() {
+			It("returns an error when Write is called after the deadline", func() {
+				str.SetWriteDeadline(time.Now().Add(-time.Second))
+				n, err := strWithTimeout.Write([]byte("foobar"))
+				Expect(err).To(MatchError(errDeadline))
+				Expect(n).To(BeZero())
+			})
+
+			It("unblocks after the deadline", func() {
+				deadline := time.Now().Add(scaleDuration(50 * time.Millisecond))
+				str.SetWriteDeadline(deadline)
+				n, err := strWithTimeout.Write([]byte("foobar"))
+				Expect(err).To(MatchError(errDeadline))
+				Expect(n).To(BeZero())
+				Expect(time.Now()).To(BeTemporally("~", deadline, scaleDuration(20*time.Millisecond)))
+			})
+
+			It("returns the number of bytes written, when the deadline expires", func() {
+				mockFC.EXPECT().SendWindowSize().Return(protocol.ByteCount(10000)).AnyTimes()
+				mockFC.EXPECT().AddBytesSent(gomock.Any())
+				deadline := time.Now().Add(scaleDuration(50 * time.Millisecond))
+				str.SetWriteDeadline(deadline)
+				var n int
+				writeReturned := make(chan struct{})
+				go func() {
+					defer GinkgoRecover()
+					var err error
+					n, err = strWithTimeout.Write(bytes.Repeat([]byte{0}, 100))
+					Expect(err).To(MatchError(errDeadline))
+					Expect(time.Now()).To(BeTemporally("~", deadline, scaleDuration(20*time.Millisecond)))
+					close(writeReturned)
+				}()
+				var frame *wire.StreamFrame
+				Eventually(func() *wire.StreamFrame {
+					frame = str.PopStreamFrame(50)
+					return frame
+				}).ShouldNot(BeNil())
+				Eventually(writeReturned, scaleDuration(80*time.Millisecond)).Should(BeClosed())
+				Expect(n).To(BeEquivalentTo(frame.DataLen()))
+			})
+
+			It("doesn't unblock if the deadline is changed before the first one expires", func() {
+				deadline1 := time.Now().Add(scaleDuration(50 * time.Millisecond))
+				deadline2 := time.Now().Add(scaleDuration(100 * time.Millisecond))
+				str.SetWriteDeadline(deadline1)
+				go func() {
+					defer GinkgoRecover()
+					time.Sleep(scaleDuration(20 * time.Millisecond))
+					str.SetWriteDeadline(deadline2)
+					// make sure that this was actually execute before the deadline expires
+					Expect(time.Now()).To(BeTemporally("<", deadline1))
+				}()
+				runtime.Gosched()
+				n, err := strWithTimeout.Write([]byte("foobar"))
+				Expect(err).To(MatchError(errDeadline))
+				Expect(n).To(BeZero())
+				Expect(time.Now()).To(BeTemporally("~", deadline2, scaleDuration(20*time.Millisecond)))
+			})
+
+			It("unblocks earlier, when a new deadline is set", func() {
+				deadline1 := time.Now().Add(scaleDuration(200 * time.Millisecond))
+				deadline2 := time.Now().Add(scaleDuration(50 * time.Millisecond))
+				go func() {
+					defer GinkgoRecover()
+					time.Sleep(scaleDuration(10 * time.Millisecond))
+					str.SetWriteDeadline(deadline2)
+					// make sure that this was actually execute before the deadline expires
+					Expect(time.Now()).To(BeTemporally("<", deadline2))
+				}()
+				str.SetWriteDeadline(deadline1)
+				runtime.Gosched()
+				_, err := strWithTimeout.Write([]byte("foobar"))
+				Expect(err).To(MatchError(errDeadline))
+				Expect(time.Now()).To(BeTemporally("~", deadline2, scaleDuration(20*time.Millisecond)))
+			})
+		})
+
+		Context("closing", func() {
+			It("doesn't allow writes after it has been closed", func() {
+				str.Close()
+				_, err := strWithTimeout.Write([]byte("foobar"))
+				Expect(err).To(MatchError("write on closed stream 1337"))
+			})
+
+			It("allows FIN", func() {
+				str.Close()
+				f := str.PopStreamFrame(1000)
+				Expect(f).ToNot(BeNil())
+				Expect(f.Data).To(BeEmpty())
+				Expect(f.FinBit).To(BeTrue())
+			})
+
+			It("doesn't allow FIN when there's still data", func() {
+				frameHeaderLen := protocol.ByteCount(4)
+				mockFC.EXPECT().SendWindowSize().Return(protocol.ByteCount(9999)).Times(2)
+				mockFC.EXPECT().AddBytesSent(gomock.Any()).Times(2)
+				str.dataForWriting = []byte("foobar")
+				str.Close()
+				f := str.PopStreamFrame(3 + frameHeaderLen)
+				Expect(f).ToNot(BeNil())
+				Expect(f.Data).To(Equal([]byte("foo")))
+				Expect(f.FinBit).To(BeFalse())
+				f = str.PopStreamFrame(100)
+				Expect(f.Data).To(Equal([]byte("bar")))
+				Expect(f.FinBit).To(BeTrue())
+			})
+
+			It("doesn't allow FIN after an error", func() {
+				str.CloseForShutdown(errors.New("test"))
+				f := str.PopStreamFrame(1000)
+				Expect(f).To(BeNil())
+			})
+
+			It("doesn't allow FIN twice", func() {
+				str.Close()
+				f := str.PopStreamFrame(1000)
+				Expect(f).ToNot(BeNil())
+				Expect(f.Data).To(BeEmpty())
+				Expect(f.FinBit).To(BeTrue())
+				Expect(str.PopStreamFrame(1000)).To(BeNil())
+			})
+		})
+
+		Context("closing for shutdown", func() {
+			testErr := errors.New("test")
+
+			It("returns errors when the stream is cancelled", func() {
+				str.CloseForShutdown(testErr)
+				n, err := strWithTimeout.Write([]byte("foo"))
+				Expect(n).To(BeZero())
+				Expect(err).To(MatchError(testErr))
+			})
+
+			It("doesn't get data for writing if an error occurred", func() {
+				mockFC.EXPECT().SendWindowSize().Return(protocol.ByteCount(9999))
+				mockFC.EXPECT().AddBytesSent(gomock.Any())
+				done := make(chan struct{})
+				go func() {
+					defer GinkgoRecover()
+					_, err := strWithTimeout.Write(bytes.Repeat([]byte{0}, 500))
+					Expect(err).To(MatchError(testErr))
+					close(done)
+				}()
+				Eventually(func() *wire.StreamFrame { return str.PopStreamFrame(50) }).ShouldNot(BeNil()) // get a STREAM frame containing some data, but not all
+				str.CloseForShutdown(testErr)
+				Expect(str.PopStreamFrame(1000)).To(BeNil())
+				Eventually(done).Should(BeClosed())
+			})
+
+			It("cancels the context", func() {
+				Expect(str.Context().Done()).ToNot(BeClosed())
+				str.CloseForShutdown(testErr)
+				Expect(str.Context().Done()).To(BeClosed())
+			})
+		})
+	})
+
+	Context("stream cancelations", func() {
+		Context("canceling writing", func() {
+			It("queues a RST_STREAM frame", func() {
+				str.writeOffset = 1234
+				err := str.CancelWrite(9876)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(queuedControlFrames).To(Equal([]wire.Frame{
+					&wire.RstStreamFrame{
+						StreamID:   streamID,
+						ByteOffset: 1234,
+						ErrorCode:  9876,
+					},
+				}))
+			})
+
+			It("unblocks Write", func() {
+				mockFC.EXPECT().SendWindowSize().Return(protocol.MaxByteCount)
+				mockFC.EXPECT().AddBytesSent(gomock.Any())
+				writeReturned := make(chan struct{})
+				var n int
+				go func() {
+					defer GinkgoRecover()
+					var err error
+					n, err = strWithTimeout.Write(bytes.Repeat([]byte{0}, 100))
+					Expect(err).To(MatchError("Write on stream 1337 canceled with error code 1234"))
+					close(writeReturned)
+				}()
+				var frame *wire.StreamFrame
+				Eventually(func() *wire.StreamFrame {
+					frame = str.PopStreamFrame(50)
+					return frame
+				}).ShouldNot(BeNil())
+				err := str.CancelWrite(1234)
+				Expect(err).ToNot(HaveOccurred())
+				Eventually(writeReturned).Should(BeClosed())
+				Expect(n).To(BeEquivalentTo(frame.DataLen()))
+			})
+
+			It("cancels the context", func() {
+				Expect(str.Context().Done()).ToNot(BeClosed())
+				str.CancelWrite(1234)
+				Expect(str.Context().Done()).To(BeClosed())
+			})
+
+			It("doesn't allow further calls to Write", func() {
+				err := str.CancelWrite(1234)
+				Expect(err).ToNot(HaveOccurred())
+				_, err = strWithTimeout.Write([]byte("foobar"))
+				Expect(err).To(MatchError("Write on stream 1337 canceled with error code 1234"))
+			})
+
+			It("only cancels once", func() {
+				err := str.CancelWrite(1234)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(queuedControlFrames).To(HaveLen(1))
+				err = str.CancelWrite(4321)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(queuedControlFrames).To(HaveLen(1))
+			})
+
+			It("doesn't cancel when the stream was already closed", func() {
+				err := str.Close()
+				Expect(err).ToNot(HaveOccurred())
+				err = str.CancelWrite(123)
+				Expect(err).To(MatchError("CancelWrite for closed stream 1337"))
+			})
+		})
+
+		Context("receiving STOP_SENDING frames", func() {
+			It("queues a RST_STREAM frames with error code Stopping", func() {
+				str.HandleStopSendingFrame(&wire.StopSendingFrame{
+					StreamID:  streamID,
+					ErrorCode: 101,
+				})
+				Expect(queuedControlFrames).To(Equal([]wire.Frame{
+					&wire.RstStreamFrame{
+						StreamID:  streamID,
+						ErrorCode: errorCodeStopping,
+					},
+				}))
+			})
+
+			It("unblocks Write", func() {
+				done := make(chan struct{})
+				go func() {
+					defer GinkgoRecover()
+					_, err := str.Write([]byte("foobar"))
+					Expect(err).To(MatchError("Stream 1337 was reset with error code 123"))
+					Expect(err).To(BeAssignableToTypeOf(streamCanceledError{}))
+					Expect(err.(streamCanceledError).Canceled()).To(BeTrue())
+					Expect(err.(streamCanceledError).ErrorCode()).To(Equal(protocol.ApplicationErrorCode(123)))
+					close(done)
+				}()
+				Consistently(done).ShouldNot(BeClosed())
+				str.HandleStopSendingFrame(&wire.StopSendingFrame{
+					StreamID:  streamID,
+					ErrorCode: 123,
+				})
+				Eventually(done).Should(BeClosed())
+			})
+
+			It("doesn't allow further calls to Write", func() {
+				str.HandleStopSendingFrame(&wire.StopSendingFrame{
+					StreamID:  streamID,
+					ErrorCode: 123,
+				})
+				_, err := str.Write([]byte("foobar"))
+				Expect(err).To(MatchError("Stream 1337 was reset with error code 123"))
+				Expect(err).To(BeAssignableToTypeOf(streamCanceledError{}))
+				Expect(err.(streamCanceledError).Canceled()).To(BeTrue())
+				Expect(err.(streamCanceledError).ErrorCode()).To(Equal(protocol.ApplicationErrorCode(123)))
+			})
+		})
+	})
+
+	Context("flow control", func() {
+		It("says when it's flow control blocked", func() {
+			mockFC.EXPECT().IsBlocked().Return(false, protocol.ByteCount(0))
+			blocked, _ := str.IsFlowControlBlocked()
+			Expect(blocked).To(BeFalse())
+			mockFC.EXPECT().IsBlocked().Return(true, protocol.ByteCount(0x1337))
+			blocked, offset := str.IsFlowControlBlocked()
+			Expect(blocked).To(BeTrue())
+			Expect(offset).To(Equal(protocol.ByteCount(0x1337)))
+		})
+
+		It("updates the flow control window", func() {
+			mockFC.EXPECT().UpdateSendWindow(protocol.ByteCount(0x42))
+			str.HandleMaxStreamDataFrame(&wire.MaxStreamDataFrame{
+				StreamID:   streamID,
+				ByteOffset: 0x42,
+			})
+		})
+	})
+
+	Context("saying if it is finished", func() {
+		It("is finished after it is closed for shutdown", func() {
+			str.CloseForShutdown(errors.New("testErr"))
+			Expect(str.Finished()).To(BeTrue())
+		})
+
+		It("is finished after Close()", func() {
+			str.Close()
+			f := str.PopStreamFrame(1000)
+			Expect(f.FinBit).To(BeTrue())
+			Expect(str.Finished()).To(BeTrue())
+		})
+
+		It("is finished after CancelWrite", func() {
+			err := str.CancelWrite(123)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(str.Finished()).To(BeTrue())
+		})
+
+		It("is finished after receiving a STOP_SENDING (and sending a RST_STREAM)", func() {
+			str.HandleStopSendingFrame(&wire.StopSendingFrame{StreamID: streamID})
+			Expect(str.Finished()).To(BeTrue())
+		})
+	})
+})

server_test.go

@@ -59,7 +59,7 @@ func (s *mockSession) closeRemote(e error) {
 	close(s.stopRunLoop)
 }
 func (s *mockSession) OpenStream() (Stream, error) {
-	return &stream{streamID: 1337}, nil
+	return &stream{}, nil
 }
 func (s *mockSession) AcceptStream() (Stream, error)          { panic("not implemented") }
 func (s *mockSession) OpenStreamSync() (Stream, error)        { panic("not implemented") }

session_test.go

@@ -362,7 +362,7 @@ var _ = Describe("Session", func() {
 				fc := mocks.NewMockStreamFlowController(mockCtrl)
 				offset := protocol.ByteCount(0x4321)
 				fc.EXPECT().UpdateSendWindow(offset)
-				sess.cryptoStream.(*cryptoStream).flowController = fc
+				sess.cryptoStream.(*cryptoStream).sendStream.flowController = fc
 				err := sess.handleMaxStreamDataFrame(&wire.MaxStreamDataFrame{
 					StreamID:   sess.version.CryptoStreamID(),
 					ByteOffset: offset,
@@ -1118,9 +1118,21 @@ var _ = Describe("Session", func() {
 				s2, err := sess.GetOrOpenStream(7)
 				Expect(err).NotTo(HaveOccurred())
 
-				// Put data directly into the streams
-				s1.(*stream).dataForWriting = []byte("foobar1")
-				s2.(*stream).dataForWriting = []byte("foobar2")
+				done1 := make(chan struct{})
+				done2 := make(chan struct{})
+				go func() {
+					defer GinkgoRecover()
+					_, err := s1.Write([]byte("foobar1"))
+					Expect(err).ToNot(HaveOccurred())
+					close(done1)
+				}()
+				go func() {
+					defer GinkgoRecover()
+					s2.Write([]byte("foobar2"))
+					Expect(err).ToNot(HaveOccurred())
+					close(done2)
+				}()
+				time.Sleep(100 * time.Millisecond) // make sure the both writes are active
 
 				sess.scheduleSending()
 				go sess.run()
@@ -1130,6 +1142,8 @@ var _ = Describe("Session", func() {
 				packet := <-mconn.written
 				Expect(packet).To(ContainSubstring("foobar1"))
 				Expect(packet).To(ContainSubstring("foobar2"))
+				Eventually(done1).Should(BeClosed())
+				Eventually(done2).Should(BeClosed())
 			})
 
 			It("sends out two big frames in two packets", func() {

stream.go

@@ -1,30 +1,14 @@
 package quic
 
 import (
-	"context"
-	"fmt"
-	"io"
 	"net"
-	"sync"
 	"time"
 
 	"github.com/lucas-clemente/quic-go/internal/flowcontrol"
 	"github.com/lucas-clemente/quic-go/internal/protocol"
-	"github.com/lucas-clemente/quic-go/internal/utils"
 	"github.com/lucas-clemente/quic-go/internal/wire"
 )
 
-type streamCanceledError struct {
-	error
-	errorCode protocol.ApplicationErrorCode
-}
-
-func (streamCanceledError) Canceled() bool                             { return true }
-func (e streamCanceledError) ErrorCode() protocol.ApplicationErrorCode { return e.errorCode }
-
-var _ StreamError = &streamCanceledError{}
-var _ error = &streamCanceledError{}
-
 const (
 	errorCodeStopping      protocol.ApplicationErrorCode = 0
 	errorCodeStoppingGQUIC protocol.ApplicationErrorCode = 7
@@ -49,45 +33,10 @@ type streamI interface {
 //
 // Read() and Write() may be called concurrently, but multiple calls to Read() or Write() individually must be synchronized manually.
 type stream struct {
-	mutex sync.Mutex
+	receiveStream
+	sendStream
 
-	ctx       context.Context
-	ctxCancel context.CancelFunc
-
-	streamID protocol.StreamID
-	// onData tells the session that there's stuff to pack into a new packet
-	onData func()
-	// queueControlFrame queues a new control frame for sending
-	// it does not call onData
-	queueControlFrame func(wire.Frame)
-
-	readPosInFrame int
-	writeOffset    protocol.ByteCount
-	readOffset     protocol.ByteCount
-
-	closeForShutdownErr error
-	cancelWriteErr      error
-	cancelReadErr       error
-	resetRemotelyErr    StreamError
-
-	closedForShutdown bool // set when CloseForShutdown() is called
-	finRead           bool // set once we read a frame with a FinBit
-	finishedWriting   bool // set once Close() is called
-	canceledWrite     bool // set when CancelWrite() is called, or a STOP_SENDING frame is received
-	canceledRead      bool // set when CancelRead() is called
-	finSent           bool // set when a STREAM_FRAME with FIN bit has b
-	resetRemotely     bool // set when HandleRstStreamFrame() is called
-
-	frameQueue   *streamFrameSorter
-	readChan     chan struct{}
-	readDeadline time.Time
-
-	dataForWriting []byte
-	writeChan      chan struct{}
-	writeDeadline  time.Time
-
-	flowController flowcontrol.StreamFlowController
-	version        protocol.VersionNumber
+	version protocol.VersionNumber
 }
 
 var _ Stream = &stream{}
@@ -101,299 +50,41 @@ func (deadlineError) Timeout() bool   { return true }
 
 var errDeadline net.Error = &deadlineError{}
 
+type streamCanceledError struct {
+	error
+	errorCode protocol.ApplicationErrorCode
+}
+
+func (streamCanceledError) Canceled() bool                             { return true }
+func (e streamCanceledError) ErrorCode() protocol.ApplicationErrorCode { return e.errorCode }
+
+var _ StreamError = &streamCanceledError{}
+
 // newStream creates a new Stream
-func newStream(StreamID protocol.StreamID,
+func newStream(streamID protocol.StreamID,
 	onData func(),
 	queueControlFrame func(wire.Frame),
 	flowController flowcontrol.StreamFlowController,
 	version protocol.VersionNumber,
 ) *stream {
-	s := &stream{
-		onData:            onData,
-		queueControlFrame: queueControlFrame,
-		streamID:          StreamID,
-		flowController:    flowController,
-		frameQueue:        newStreamFrameSorter(),
-		readChan:          make(chan struct{}, 1),
-		writeChan:         make(chan struct{}, 1),
-		version:           version,
+	return &stream{
+		sendStream:    *newSendStream(streamID, onData, queueControlFrame, flowController, version),
+		receiveStream: *newReceiveStream(streamID, onData, queueControlFrame, flowController),
 	}
-	s.ctx, s.ctxCancel = context.WithCancel(context.Background())
-	return s
 }
 
-// Read implements io.Reader. It is not thread safe!
-func (s *stream) Read(p []byte) (int, error) {
-	s.mutex.Lock()
-	defer s.mutex.Unlock()
-
-	if s.finRead {
-		return 0, io.EOF
-	}
-	if s.canceledRead {
-		return 0, s.cancelReadErr
-	}
-	if s.resetRemotely {
-		return 0, s.resetRemotelyErr
-	}
-	if s.closedForShutdown {
-		return 0, s.closeForShutdownErr
-	}
-
-	bytesRead := 0
-	for bytesRead < len(p) {
-		frame := s.frameQueue.Head()
-		if frame == nil && bytesRead > 0 {
-			return bytesRead, s.closeForShutdownErr
-		}
-
-		for {
-			// Stop waiting on errors
-			if s.closedForShutdown {
-				return bytesRead, s.closeForShutdownErr
-			}
-			if s.canceledRead {
-				return bytesRead, s.cancelReadErr
-			}
-			if s.resetRemotely {
-				return bytesRead, s.resetRemotelyErr
-			}
-
-			deadline := s.readDeadline
-			if !deadline.IsZero() && !time.Now().Before(deadline) {
-				return bytesRead, errDeadline
-			}
-
-			if frame != nil {
-				s.readPosInFrame = int(s.readOffset - frame.Offset)
-				break
-			}
-
-			s.mutex.Unlock()
-			if deadline.IsZero() {
-				<-s.readChan
-			} else {
-				select {
-				case <-s.readChan:
-				case <-time.After(deadline.Sub(time.Now())):
-				}
-			}
-			s.mutex.Lock()
-			frame = s.frameQueue.Head()
-		}
-
-		if bytesRead > len(p) {
-			return bytesRead, fmt.Errorf("BUG: bytesRead (%d) > len(p) (%d) in stream.Read", bytesRead, len(p))
-		}
-		if s.readPosInFrame > int(frame.DataLen()) {
-			return bytesRead, fmt.Errorf("BUG: readPosInFrame (%d) > frame.DataLen (%d) in stream.Read", s.readPosInFrame, frame.DataLen())
-		}
-
-		s.mutex.Unlock()
-
-		copy(p[bytesRead:], frame.Data[s.readPosInFrame:])
-		m := utils.Min(len(p)-bytesRead, int(frame.DataLen())-s.readPosInFrame)
-		s.readPosInFrame += m
-		bytesRead += m
-		s.readOffset += protocol.ByteCount(m)
-
-		s.mutex.Lock()
-		// when a RST_STREAM was received, the was already informed about the final byteOffset for this stream
-		if !s.resetRemotely {
-			s.flowController.AddBytesRead(protocol.ByteCount(m))
-		}
-		s.onData() // so that a possible WINDOW_UPDATE is sent
-
-		if s.readPosInFrame >= int(frame.DataLen()) {
-			s.frameQueue.Pop()
-			s.finRead = frame.FinBit
-			if frame.FinBit {
-				return bytesRead, io.EOF
-			}
-		}
-	}
-	return bytesRead, nil
-}
-
-func (s *stream) 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 len(p) == 0 {
-		return 0, nil
-	}
-
-	s.dataForWriting = make([]byte, len(p))
-	copy(s.dataForWriting, p)
-	s.onData()
-
-	var err error
-	for {
-		deadline := s.writeDeadline
-		if !deadline.IsZero() && !time.Now().Before(deadline) {
-			err = errDeadline
-			break
-		}
-		if s.dataForWriting == nil || s.canceledWrite || s.closedForShutdown {
-			break
-		}
-
-		s.mutex.Unlock()
-		if deadline.IsZero() {
-			<-s.writeChan
-		} else {
-			select {
-			case <-s.writeChan:
-			case <-time.After(deadline.Sub(time.Now())):
-			}
-		}
-		s.mutex.Lock()
-	}
-
-	if s.closeForShutdownErr != nil {
-		err = s.closeForShutdownErr
-	} else if s.cancelWriteErr != nil {
-		err = s.cancelWriteErr
-	}
-	return len(p) - len(s.dataForWriting), err
-}
-
-// 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 *stream) PopStreamFrame(maxBytes protocol.ByteCount) *wire.StreamFrame {
-	s.mutex.Lock()
-	defer s.mutex.Unlock()
-
-	if s.closeForShutdownErr != nil {
-		return nil
-	}
-
-	frame := &wire.StreamFrame{
-		StreamID:       s.streamID,
-		Offset:         s.writeOffset,
-		DataLenPresent: true,
-	}
-	frameLen := frame.MinLength(s.version)
-	if frameLen >= maxBytes { // a STREAM frame must have at least one byte of data
-		return nil
-	}
-	frame.Data, frame.FinBit = s.getDataForWriting(maxBytes - frameLen)
-	if len(frame.Data) == 0 && !frame.FinBit {
-		return nil
-	}
-	if frame.FinBit {
-		s.finSent = true
-	}
-	return frame
-}
-
-func (s *stream) getDataForWriting(maxBytes protocol.ByteCount) ([]byte, bool /* should send FIN */) {
-	if s.dataForWriting == nil {
-		return nil, s.finishedWriting && !s.finSent
-	}
-
-	// TODO(#657): Flow control for the crypto stream
-	if s.streamID != s.version.CryptoStreamID() {
-		maxBytes = utils.MinByteCount(maxBytes, s.flowController.SendWindowSize())
-	}
-	if maxBytes == 0 {
-		return nil, false
-	}
-
-	var ret []byte
-	if protocol.ByteCount(len(s.dataForWriting)) > maxBytes {
-		ret = s.dataForWriting[:maxBytes]
-		s.dataForWriting = s.dataForWriting[maxBytes:]
-	} else {
-		ret = s.dataForWriting
-		s.dataForWriting = nil
-		s.signalWrite()
-	}
-	s.writeOffset += protocol.ByteCount(len(ret))
-	s.flowController.AddBytesSent(protocol.ByteCount(len(ret)))
-	return ret, s.finishedWriting && s.dataForWriting == nil && !s.finSent
+// need to define StreamID() here, since both receiveStream and readStream have a StreamID()
+func (s *stream) StreamID() protocol.StreamID {
+	// the result is same for receiveStream and sendStream
+	return s.sendStream.StreamID()
 }
 
 func (s *stream) Close() error {
-	s.mutex.Lock()
-	defer s.mutex.Unlock()
-
-	if s.canceledWrite {
-		return fmt.Errorf("Close called for canceled stream %d", s.streamID)
-	}
-	if s.canceledRead && !s.version.UsesIETFFrameFormat() {
-		s.queueControlFrame(&wire.RstStreamFrame{
-			StreamID:   s.streamID,
-			ByteOffset: s.writeOffset,
-			ErrorCode:  0,
-		})
-	}
-	s.finishedWriting = true
-	s.ctxCancel()
-	s.onData()
-	return nil
-}
-
-func (s *stream) HandleStreamFrame(frame *wire.StreamFrame) error {
-	maxOffset := frame.Offset + frame.DataLen()
-	if err := s.flowController.UpdateHighestReceived(maxOffset, frame.FinBit); err != nil {
+	if err := s.sendStream.Close(); err != nil {
 		return err
 	}
-
-	s.mutex.Lock()
-	defer s.mutex.Unlock()
-	if err := s.frameQueue.Push(frame); err != nil && err != errDuplicateStreamData {
-		return err
-	}
-	s.signalRead()
-	return nil
-}
-
-// signalRead performs a non-blocking send on the readChan
-func (s *stream) signalRead() {
-	select {
-	case s.readChan <- struct{}{}:
-	default:
-	}
-}
-
-// signalWrite performs a non-blocking send on the writeChan
-func (s *stream) signalWrite() {
-	select {
-	case s.writeChan <- struct{}{}:
-	default:
-	}
-}
-
-func (s *stream) SetReadDeadline(t time.Time) error {
-	s.mutex.Lock()
-	oldDeadline := s.readDeadline
-	s.readDeadline = t
-	s.mutex.Unlock()
-	// if the new deadline is before the currently set deadline, wake up Read()
-	if t.Before(oldDeadline) {
-		s.signalRead()
-	}
-	return nil
-}
-
-func (s *stream) SetWriteDeadline(t time.Time) error {
-	s.mutex.Lock()
-	oldDeadline := s.writeDeadline
-	s.writeDeadline = t
-	s.mutex.Unlock()
-	if t.Before(oldDeadline) {
-		s.signalWrite()
-	}
+	// in gQUIC, we need to send a RST_STREAM with the final offset if CancelRead() was called
+	s.receiveStream.onClose(s.sendStream.getWriteOffset())
 	return nil
 }
 
@@ -403,157 +94,27 @@ func (s *stream) SetDeadline(t time.Time) error {
 	return nil
 }
 
-// CloseRemote makes the stream receive a "virtual" FIN stream frame at a given offset
-func (s *stream) CloseRemote(offset protocol.ByteCount) {
-	s.HandleStreamFrame(&wire.StreamFrame{FinBit: true, Offset: offset})
-}
-
 // CloseForShutdown closes a stream abruptly.
 // It makes Read and 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 *stream) CloseForShutdown(err error) {
-	s.mutex.Lock()
-	s.closedForShutdown = true
-	s.closeForShutdownErr = err
-	s.mutex.Unlock()
-	s.signalRead()
-	s.signalWrite()
-	s.ctxCancel()
-}
-
-func (s *stream) CancelWrite(errorCode protocol.ApplicationErrorCode) error {
-	s.mutex.Lock()
-	defer s.mutex.Unlock()
-
-	return 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 *stream) cancelWriteImpl(errorCode protocol.ApplicationErrorCode, writeErr error) error {
-	if s.canceledWrite {
-		return nil
-	}
-	if s.finishedWriting {
-		return fmt.Errorf("CancelWrite for closed stream %d", s.streamID)
-	}
-	s.canceledWrite = true
-	s.cancelWriteErr = writeErr
-	s.signalWrite()
-	s.queueControlFrame(&wire.RstStreamFrame{
-		StreamID:   s.streamID,
-		ByteOffset: s.writeOffset,
-		ErrorCode:  errorCode,
-	})
-	// TODO(#991): cancel retransmissions for this stream
-	s.onData()
-	s.ctxCancel()
-	return nil
-}
-
-func (s *stream) CancelRead(errorCode protocol.ApplicationErrorCode) error {
-	s.mutex.Lock()
-	defer s.mutex.Unlock()
-
-	if s.finRead {
-		return nil
-	}
-	if s.canceledRead {
-		return nil
-	}
-	s.canceledRead = true
-	s.cancelReadErr = fmt.Errorf("Read on stream %d canceled with error code %d", s.streamID, errorCode)
-	s.signalRead()
-	if s.version.UsesIETFFrameFormat() {
-		s.queueControlFrame(&wire.StopSendingFrame{
-			StreamID:  s.streamID,
-			ErrorCode: errorCode,
-		})
-	}
-	return nil
+	s.sendStream.CloseForShutdown(err)
+	s.receiveStream.CloseForShutdown(err)
 }
 
 func (s *stream) HandleRstStreamFrame(frame *wire.RstStreamFrame) error {
-	s.mutex.Lock()
-	defer s.mutex.Unlock()
-
-	if s.closedForShutdown {
-		return nil
-	}
-	if err := s.flowController.UpdateHighestReceived(frame.ByteOffset, true); err != nil {
+	if err := s.receiveStream.HandleRstStreamFrame(frame); err != nil {
 		return err
 	}
 	if !s.version.UsesIETFFrameFormat() {
-		s.handleStopSendingFrameImpl(&wire.StopSendingFrame{
-			StreamID:  s.streamID,
+		s.HandleStopSendingFrame(&wire.StopSendingFrame{
+			StreamID:  s.StreamID(),
 			ErrorCode: frame.ErrorCode,
 		})
-		// In gQUIC, error code 0 has a special meaning.
-		// The peer will reliably continue transmitting, but is not interested in reading from the stream.
-		// We should therefore just continue reading from the stream, until we encounter the FIN bit.
-		if frame.ErrorCode == 0 {
-			return nil
-		}
 	}
-
-	// ignore duplicate RST_STREAM frames for this stream (after checking their final offset)
-	if s.resetRemotely {
-		return nil
-	}
-	s.resetRemotely = true
-	s.resetRemotelyErr = streamCanceledError{
-		errorCode: frame.ErrorCode,
-		error:     fmt.Errorf("Stream %d was reset with error code %d", s.streamID, frame.ErrorCode),
-	}
-	s.signalRead()
 	return nil
 }
 
-func (s *stream) HandleStopSendingFrame(frame *wire.StopSendingFrame) {
-	s.mutex.Lock()
-	defer s.mutex.Unlock()
-	s.handleStopSendingFrameImpl(frame)
-}
-
-// must be called after locking the mutex
-func (s *stream) handleStopSendingFrameImpl(frame *wire.StopSendingFrame) {
-	writeErr := streamCanceledError{
-		errorCode: frame.ErrorCode,
-		error:     fmt.Errorf("Stream %d was reset with error code %d", s.streamID, frame.ErrorCode),
-	}
-	errorCode := errorCodeStopping
-	if !s.version.UsesIETFFrameFormat() {
-		errorCode = errorCodeStoppingGQUIC
-	}
-	s.cancelWriteImpl(errorCode, writeErr)
-}
-
 func (s *stream) Finished() bool {
-	s.mutex.Lock()
-	defer s.mutex.Unlock()
-
-	sendSideClosed := s.finSent || s.canceledWrite
-	receiveSideClosed := s.finRead || s.resetRemotely
-
-	return s.closedForShutdown || // if the stream was abruptly closed for shutting down
-		sendSideClosed && receiveSideClosed
-}
-
-func (s *stream) Context() context.Context {
-	return s.ctx
-}
-
-func (s *stream) StreamID() protocol.StreamID {
-	return s.streamID
-}
-
-func (s *stream) HandleMaxStreamDataFrame(frame *wire.MaxStreamDataFrame) {
-	s.flowController.UpdateSendWindow(frame.ByteOffset)
-}
-
-func (s *stream) IsFlowControlBlocked() (bool, protocol.ByteCount) {
-	return s.flowController.IsBlocked()
-}
-
-func (s *stream) GetWindowUpdate() protocol.ByteCount {
-	return s.flowController.GetWindowUpdate()
+	return s.sendStream.Finished() && s.receiveStream.Finished()
 }

streams_map_test.go

@@ -678,7 +678,9 @@ var _ = Describe("Streams Map", func() {
 			BeforeEach(func() {
 				callbackCalledForStream = callbackCalledForStream[:0]
 				for i := 4; i <= 8; i++ {
-					err := m.putStream(&stream{streamID: protocol.StreamID(i)})
+					str := mocks.NewMockStreamI(mockCtrl)
+					str.EXPECT().StreamID().Return(protocol.StreamID(i)).AnyTimes()
+					err := m.putStream(str)
 					Expect(err).NotTo(HaveOccurred())
 				}
 			})