Source file src/vendor/golang.org/x/net/quic/stream.go

     1  // Copyright 2023 The Go Authors. All rights reserved.
     2  // Use of this source code is governed by a BSD-style
     3  // license that can be found in the LICENSE file.
     4  
     5  package quic
     6  
     7  import (
     8  	"context"
     9  	"errors"
    10  	"fmt"
    11  	"io"
    12  	"math"
    13  	"sync"
    14  
    15  	"golang.org/x/net/internal/quic/quicwire"
    16  )
    17  
    18  // A Stream is an ordered byte stream.
    19  //
    20  // Streams may be bidirectional, read-only, or write-only.
    21  // Methods inappropriate for a stream's direction
    22  // (for example, [Write] to a read-only stream)
    23  // return errors.
    24  //
    25  // It is not safe to perform concurrent reads from or writes to a stream.
    26  // It is safe, however, to read and write at the same time.
    27  //
    28  // Reads and writes are buffered.
    29  // It is generally not necessary to wrap a stream in a [bufio.ReadWriter]
    30  // or otherwise apply additional buffering.
    31  //
    32  // To cancel reads or writes, use the [SetReadContext] and [SetWriteContext] methods.
    33  type Stream struct {
    34  	id   streamID
    35  	conn *Conn
    36  
    37  	// Contexts used for read/write operations.
    38  	// Intentionally not mutex-guarded, to allow the race detector to catch concurrent access.
    39  	inctx  context.Context
    40  	outctx context.Context
    41  
    42  	// ingate's lock guards receive-related state.
    43  	//
    44  	// The gate condition is set if a read from the stream will not block,
    45  	// either because the stream has available data or because the read will fail.
    46  	ingate      gate
    47  	in          pipe            // received data
    48  	inwin       int64           // last MAX_STREAM_DATA sent to the peer
    49  	insendmax   sentVal         // set when we should send MAX_STREAM_DATA to the peer
    50  	inmaxbuf    int64           // maximum amount of data we will buffer
    51  	insize      int64           // stream final size; -1 before this is known
    52  	inset       rangeset[int64] // received ranges
    53  	inclosed    sentVal         // set by CloseRead
    54  	inresetcode int64           // RESET_STREAM code received from the peer; -1 if not reset
    55  
    56  	// outgate's lock guards send-related state.
    57  	//
    58  	// The gate condition is set if a write to the stream will not block,
    59  	// either because the stream has available flow control or because
    60  	// the write will fail.
    61  	outgate      gate
    62  	out          pipe            // buffered data to send
    63  	outflushed   int64           // offset of last flush call
    64  	outwin       int64           // maximum MAX_STREAM_DATA received from the peer
    65  	outmaxsent   int64           // maximum data offset we've sent to the peer
    66  	outmaxbuf    int64           // maximum amount of data we will buffer
    67  	outunsent    rangeset[int64] // ranges buffered but not yet sent (only flushed data)
    68  	outacked     rangeset[int64] // ranges sent and acknowledged
    69  	outopened    sentVal         // set if we should open the stream
    70  	outclosed    sentVal         // set by CloseWrite
    71  	outblocked   sentVal         // set when a write to the stream is blocked by flow control
    72  	outreset     sentVal         // set by Reset
    73  	outresetcode uint64          // reset code to send in RESET_STREAM
    74  	outdone      chan struct{}   // closed when all data sent
    75  
    76  	// Buffers used for fast path; mutex-guarded, but uncontended in normal operations.
    77  	inbufmu  sync.Mutex
    78  	inbuf    []byte // received data
    79  	inbufoff int    // bytes of inbuf which have been consumed
    80  
    81  	outbufmu  sync.Mutex
    82  	outbuf    []byte // written data
    83  	outbufoff int    // bytes of outbuf which contain data to write
    84  
    85  	// Atomic stream state bits.
    86  	//
    87  	// These bits provide a fast way to coordinate between the
    88  	// send and receive sides of the stream, and the conn's loop.
    89  	//
    90  	// streamIn* bits must be set with ingate held.
    91  	// streamOut* bits must be set with outgate held.
    92  	// streamConn* bits are set by the conn's loop.
    93  	// streamQueue* bits must be set with streamsState.sendMu held.
    94  	state atomicBits[streamState]
    95  
    96  	prev, next *Stream // guarded by streamsState.sendMu
    97  }
    98  
    99  type streamState uint32
   100  
   101  const (
   102  	// streamInSendMeta is set when there are frames to send for the
   103  	// inbound side of the stream. For example, MAX_STREAM_DATA.
   104  	// Inbound frames are never flow-controlled.
   105  	streamInSendMeta = streamState(1 << iota)
   106  
   107  	// streamOutSendMeta is set when there are non-flow-controlled frames
   108  	// to send for the outbound side of the stream. For example, STREAM_DATA_BLOCKED.
   109  	// streamOutSendData is set when there are no non-flow-controlled outbound frames
   110  	// and the stream has data to send.
   111  	//
   112  	// At most one of streamOutSendMeta and streamOutSendData is set at any time.
   113  	streamOutSendMeta
   114  	streamOutSendData
   115  
   116  	// streamInDone and streamOutDone are set when the inbound or outbound
   117  	// sides of the stream are finished. When both are set, the stream
   118  	// can be removed from the Conn and forgotten.
   119  	streamInDone
   120  	streamOutDone
   121  
   122  	// streamConnRemoved is set when the stream has been removed from the conn.
   123  	streamConnRemoved
   124  
   125  	// streamQueueMeta and streamQueueData indicate which of the streamsState
   126  	// send queues the conn is currently on.
   127  	streamQueueMeta
   128  	streamQueueData
   129  )
   130  
   131  type streamQueue int
   132  
   133  const (
   134  	noQueue   = streamQueue(iota)
   135  	metaQueue // streamsState.queueMeta
   136  	dataQueue // streamsState.queueData
   137  )
   138  
   139  // streamResetByConnClose is assigned to Stream.inresetcode to indicate that a stream
   140  // was implicitly reset when the connection closed. It's out of the range of
   141  // possible reset codes the peer can send.
   142  const streamResetByConnClose = math.MaxInt64
   143  
   144  // wantQueue returns the send queue the stream should be on.
   145  func (s streamState) wantQueue() streamQueue {
   146  	switch {
   147  	case s&(streamInSendMeta|streamOutSendMeta) != 0:
   148  		return metaQueue
   149  	case s&(streamInDone|streamOutDone|streamConnRemoved) == streamInDone|streamOutDone:
   150  		return metaQueue
   151  	case s&streamOutSendData != 0:
   152  		// The stream has no non-flow-controlled frames to send,
   153  		// but does have data. Put it on the data queue, which is only
   154  		// processed when flow control is available.
   155  		return dataQueue
   156  	}
   157  	return noQueue
   158  }
   159  
   160  // inQueue returns the send queue the stream is currently on.
   161  func (s streamState) inQueue() streamQueue {
   162  	switch {
   163  	case s&streamQueueMeta != 0:
   164  		return metaQueue
   165  	case s&streamQueueData != 0:
   166  		return dataQueue
   167  	}
   168  	return noQueue
   169  }
   170  
   171  // newStream returns a new stream.
   172  //
   173  // The stream's ingate and outgate are locked.
   174  // (We create the stream with locked gates so after the caller
   175  // initializes the flow control window,
   176  // unlocking outgate will set the stream writability state.)
   177  func newStream(c *Conn, id streamID) *Stream {
   178  	s := &Stream{
   179  		conn:        c,
   180  		id:          id,
   181  		insize:      -1, // -1 indicates the stream size is unknown
   182  		inresetcode: -1, // -1 indicates no RESET_STREAM received
   183  		ingate:      newLockedGate(),
   184  		outgate:     newLockedGate(),
   185  		inctx:       context.Background(),
   186  		outctx:      context.Background(),
   187  	}
   188  	if !s.IsReadOnly() {
   189  		s.outdone = make(chan struct{})
   190  	}
   191  	return s
   192  }
   193  
   194  // ID returns the QUIC stream ID of s.
   195  //
   196  // As specified in RFC 9000, the two least significant bits of a stream ID
   197  // indicate the initiator and directionality of the stream. The upper bits are
   198  // the stream number.
   199  func (s *Stream) ID() int64 {
   200  	return int64(s.id)
   201  }
   202  
   203  // SetReadContext sets the context used for reads from the stream.
   204  //
   205  // It is not safe to call SetReadContext concurrently.
   206  func (s *Stream) SetReadContext(ctx context.Context) {
   207  	s.inctx = ctx
   208  }
   209  
   210  // SetWriteContext sets the context used for writes to the stream.
   211  // The write context is also used by Close when waiting for writes to be
   212  // received by the peer.
   213  //
   214  // It is not safe to call SetWriteContext concurrently.
   215  func (s *Stream) SetWriteContext(ctx context.Context) {
   216  	s.outctx = ctx
   217  }
   218  
   219  // IsReadOnly reports whether the stream is read-only
   220  // (a unidirectional stream created by the peer).
   221  func (s *Stream) IsReadOnly() bool {
   222  	return s.id.streamType() == uniStream && s.id.initiator() != s.conn.side
   223  }
   224  
   225  // IsWriteOnly reports whether the stream is write-only
   226  // (a unidirectional stream created locally).
   227  func (s *Stream) IsWriteOnly() bool {
   228  	return s.id.streamType() == uniStream && s.id.initiator() == s.conn.side
   229  }
   230  
   231  // Read reads data from the stream.
   232  //
   233  // Read returns as soon as at least one byte of data is available.
   234  //
   235  // If the peer closes the stream cleanly, Read returns io.EOF after
   236  // returning all data sent by the peer.
   237  // If the peer aborts reads on the stream, Read returns
   238  // an error wrapping StreamResetCode.
   239  //
   240  // It is not safe to call Read concurrently.
   241  func (s *Stream) Read(b []byte) (n int, err error) {
   242  	if s.IsWriteOnly() {
   243  		return 0, errors.New("read from write-only stream")
   244  	}
   245  
   246  	fastPath := false
   247  	s.inbufmu.Lock()
   248  	if len(s.inbuf) > s.inbufoff {
   249  		// Fast path: If s.inbuf contains unread bytes, return them
   250  		// immediately.
   251  		n = copy(b, s.inbuf[s.inbufoff:])
   252  		s.inbufoff += n
   253  		fastPath = true
   254  	}
   255  	s.inbufmu.Unlock()
   256  	if fastPath {
   257  		return n, nil
   258  	}
   259  
   260  	if err := s.ingate.waitAndLock(s.inctx); err != nil {
   261  		return 0, err
   262  	}
   263  
   264  	if s.inbufoff > 0 {
   265  		// Discard bytes consumed by the fast path above.
   266  		s.in.discardBefore(s.in.start + int64(s.inbufoff))
   267  		s.inbufmu.Lock()
   268  		s.inbufoff = 0
   269  		s.inbuf = nil
   270  		s.inbufmu.Unlock()
   271  	}
   272  
   273  	// bytesRead contains the number of bytes of connection-level flow control to return.
   274  	// We return flow control for bytes read by this Read call, as well as bytes moved
   275  	// to the fast-path read buffer (s.inbuf).
   276  	var bytesRead int64
   277  	defer func() {
   278  		s.inUnlock()
   279  		s.conn.handleStreamBytesReadOffLoop(bytesRead) // must be done with ingate unlocked
   280  	}()
   281  	if s.inresetcode != -1 {
   282  		if s.inresetcode == streamResetByConnClose {
   283  			if err := s.conn.finalError(); err != nil {
   284  				return 0, err
   285  			}
   286  		}
   287  		return 0, fmt.Errorf("stream reset by peer: %w", StreamErrorCode(s.inresetcode))
   288  	}
   289  	if s.inclosed.isSet() {
   290  		return 0, errors.New("read from closed stream")
   291  	}
   292  	if s.insize == s.in.start {
   293  		return 0, io.EOF
   294  	}
   295  	// Getting here indicates the stream contains data to be read.
   296  	if len(s.inset) < 1 || s.inset[0].start != 0 || s.inset[0].end <= s.in.start {
   297  		panic("BUG: inconsistent input stream state")
   298  	}
   299  	if size := int(s.inset[0].end - s.in.start); size < len(b) {
   300  		b = b[:size]
   301  	}
   302  	bytesRead = int64(len(b))
   303  	start := s.in.start
   304  	end := start + int64(len(b))
   305  	raceAcquire()
   306  	s.in.copy(start, b)
   307  	s.in.discardBefore(end)
   308  	if end == s.insize {
   309  		// We have read up to the end of the stream.
   310  		// No need to update stream flow control.
   311  		return len(b), io.EOF
   312  	}
   313  
   314  	if len(s.inset) > 0 && s.inset[0].start <= s.in.start && s.inset[0].end > s.in.start {
   315  		// If we have more readable bytes available, put the next chunk of data
   316  		// in s.inbuf for lock-free reads.
   317  		s.inbufmu.Lock()
   318  		s.inbuf = s.in.peek(s.inset[0].end - s.in.start)
   319  		s.inbufmu.Unlock()
   320  		bytesRead += int64(len(s.inbuf))
   321  	}
   322  	if s.insize == -1 || s.insize > s.inwin {
   323  		newWindow := s.in.start + int64(len(s.inbuf)) + s.inmaxbuf
   324  		addedWindow := newWindow - s.inwin
   325  		if shouldUpdateFlowControl(s.inmaxbuf, addedWindow) {
   326  			// Update stream flow control with a STREAM_MAX_DATA frame.
   327  			s.insendmax.setUnsent()
   328  		}
   329  	}
   330  
   331  	return len(b), nil
   332  }
   333  
   334  // ReadByte reads and returns a single byte from the stream.
   335  //
   336  // It is not safe to call ReadByte concurrently.
   337  func (s *Stream) ReadByte() (byte, error) {
   338  	fastPath := false
   339  	s.inbufmu.Lock()
   340  	var readByte byte
   341  	if len(s.inbuf) > s.inbufoff {
   342  		readByte = s.inbuf[s.inbufoff]
   343  		s.inbufoff++
   344  		fastPath = true
   345  	}
   346  	s.inbufmu.Unlock()
   347  	if fastPath {
   348  		return readByte, nil
   349  	}
   350  
   351  	var b [1]byte
   352  	n, err := s.Read(b[:])
   353  	if n > 0 {
   354  		return b[0], nil
   355  	}
   356  	return 0, err
   357  }
   358  
   359  // shouldUpdateFlowControl determines whether to send a flow control window update.
   360  //
   361  // We want to balance keeping the peer well-supplied with flow control with not sending
   362  // many small updates.
   363  func shouldUpdateFlowControl(maxWindow, addedWindow int64) bool {
   364  	return addedWindow >= maxWindow/8
   365  }
   366  
   367  // Write writes data to the stream.
   368  //
   369  // Write writes data to the stream write buffer.
   370  // Buffered data is only sent when the buffer is sufficiently full.
   371  // Call the Flush method to ensure buffered data is sent.
   372  func (s *Stream) Write(b []byte) (n int, err error) {
   373  	if s.IsReadOnly() {
   374  		return 0, errors.New("write to read-only stream")
   375  	}
   376  
   377  	fastPath := false
   378  	s.outbufmu.Lock()
   379  	if len(b) > 0 && len(s.outbuf)-s.outbufoff >= len(b) {
   380  		// Fast path: The data to write fits in s.outbuf.
   381  		copy(s.outbuf[s.outbufoff:], b)
   382  		s.outbufoff += len(b)
   383  		fastPath = true
   384  	}
   385  	s.outbufmu.Unlock()
   386  	if fastPath {
   387  		return len(b), nil
   388  	}
   389  
   390  	canWrite := s.outgate.lock()
   391  	s.flushFastOutputBuffer()
   392  	for {
   393  		// The first time through this loop, we may or may not be write blocked.
   394  		// We exit the loop after writing all data, so on subsequent passes through
   395  		// the loop we are always write blocked.
   396  		if len(b) > 0 && !canWrite {
   397  			// Our send buffer is full. Wait for the peer to ack some data.
   398  			s.outUnlock()
   399  			if err := s.outgate.waitAndLock(s.outctx); err != nil {
   400  				return n, err
   401  			}
   402  			// Successfully returning from waitAndLockGate means we are no longer
   403  			// write blocked. (Unlike traditional condition variables, gates do not
   404  			// have spurious wakeups.)
   405  		}
   406  		if err := s.writeErrorLocked(); err != nil {
   407  			s.outUnlock()
   408  			return n, err
   409  		}
   410  		if len(b) == 0 {
   411  			break
   412  		}
   413  		// Write limit is our send buffer limit.
   414  		// This is a stream offset.
   415  		lim := s.out.start + s.outmaxbuf
   416  		// Amount to write is min(the full buffer, data up to the write limit).
   417  		// This is a number of bytes.
   418  		nn := min(int64(len(b)), lim-s.out.end)
   419  		// Copy the data into the output buffer.
   420  		s.out.writeAt(b[:nn], s.out.end)
   421  		b = b[nn:]
   422  		n += int(nn)
   423  		// Possibly flush the output buffer.
   424  		// We automatically flush if:
   425  		//   - We have enough data to consume the send window.
   426  		//     Sending this data may cause the peer to extend the window.
   427  		//   - We have buffered as much data as we're willing to.
   428  		//     We need to send data to clear out buffer space.
   429  		//   - We have enough data to fill a 1-RTT packet using the smallest
   430  		//     possible maximum datagram size (1200 bytes, less header byte,
   431  		//     connection ID, packet number, and AEAD overhead).
   432  		const autoFlushSize = smallestMaxDatagramSize - 1 - connIDLen - 1 - aeadOverhead
   433  		shouldFlush := s.out.end >= s.outwin || // peer send window is full
   434  			s.out.end >= lim || // local send buffer is full
   435  			(s.out.end-s.outflushed) >= autoFlushSize // enough data buffered
   436  		if shouldFlush {
   437  			s.flushLocked()
   438  		}
   439  		if s.out.end > s.outwin {
   440  			// We're blocked by flow control.
   441  			// Send a STREAM_DATA_BLOCKED frame to let the peer know.
   442  			s.outblocked.set()
   443  		}
   444  		// If we have bytes left to send, we're blocked.
   445  		canWrite = false
   446  	}
   447  	if lim := s.out.start + s.outmaxbuf - s.out.end - 1; lim > 0 {
   448  		// If s.out has space allocated and available to be written into,
   449  		// then reference it in s.outbuf for fast-path writes.
   450  		//
   451  		// It's perhaps a bit pointless to limit s.outbuf to the send buffer limit.
   452  		// We've already allocated this buffer so we aren't saving any memory
   453  		// by not using it.
   454  		// For now, we limit it anyway to make it easier to reason about limits.
   455  		//
   456  		// We set the limit to one less than the send buffer limit (the -1 above)
   457  		// so that a write which completely fills the buffer will overflow
   458  		// s.outbuf and trigger a flush.
   459  		s.outbufmu.Lock()
   460  		s.outbuf = s.out.availableBuffer()
   461  		if int64(len(s.outbuf)) > lim {
   462  			s.outbuf = s.outbuf[:lim]
   463  		}
   464  		s.outbufmu.Unlock()
   465  	}
   466  	raceReleaseMerge()
   467  	s.outUnlock()
   468  	return n, nil
   469  }
   470  
   471  // WriteByte writes a single byte to the stream.
   472  func (s *Stream) WriteByte(c byte) error {
   473  	fastPath := false
   474  	s.outbufmu.Lock()
   475  	if s.outbufoff < len(s.outbuf) {
   476  		s.outbuf[s.outbufoff] = c
   477  		s.outbufoff++
   478  		fastPath = true
   479  	}
   480  	s.outbufmu.Unlock()
   481  	if fastPath {
   482  		return nil
   483  	}
   484  
   485  	b := [1]byte{c}
   486  	_, err := s.Write(b[:])
   487  	return err
   488  }
   489  
   490  func (s *Stream) flushFastOutputBuffer() {
   491  	s.outbufmu.Lock()
   492  	defer s.outbufmu.Unlock()
   493  	if s.outbuf == nil {
   494  		return
   495  	}
   496  	// Commit data previously written to s.outbuf.
   497  	// s.outbuf is a reference to a buffer in s.out, so we just need to record
   498  	// that the output buffer has been extended.
   499  	s.out.end += int64(s.outbufoff)
   500  	s.outbuf = nil
   501  	s.outbufoff = 0
   502  }
   503  
   504  // Flush flushes data written to the stream.
   505  // It does not wait for the peer to acknowledge receipt of the data.
   506  // Use Close to wait for the peer's acknowledgement.
   507  func (s *Stream) Flush() error {
   508  	if s.IsReadOnly() {
   509  		return errors.New("flush of read-only stream")
   510  	}
   511  	s.outgate.lock()
   512  	defer s.outUnlock()
   513  	if err := s.writeErrorLocked(); err != nil {
   514  		return err
   515  	}
   516  	s.flushLocked()
   517  	return nil
   518  }
   519  
   520  // writeErrorLocked returns the error (if any) which should be returned by write operations
   521  // due to the stream being reset or closed.
   522  func (s *Stream) writeErrorLocked() error {
   523  	if s.outreset.isSet() {
   524  		if s.outresetcode == streamResetByConnClose {
   525  			if err := s.conn.finalError(); err != nil {
   526  				return err
   527  			}
   528  		}
   529  		return errors.New("write to reset stream")
   530  	}
   531  	if s.outclosed.isSet() {
   532  		return errors.New("write to closed stream")
   533  	}
   534  	return nil
   535  }
   536  
   537  func (s *Stream) flushLocked() {
   538  	s.flushFastOutputBuffer()
   539  	s.outopened.set()
   540  	if s.outflushed < s.outwin {
   541  		s.outunsent.add(s.outflushed, min(s.outwin, s.out.end))
   542  	}
   543  	s.outflushed = s.out.end
   544  }
   545  
   546  // Close closes the stream.
   547  // Any blocked stream operations will be unblocked and return errors.
   548  //
   549  // Close flushes any data in the stream write buffer and waits for the peer to
   550  // acknowledge receipt of the data.
   551  // If the stream has been reset, it waits for the peer to acknowledge the reset.
   552  // If the context expires before the peer receives the stream's data,
   553  // Close discards the buffer and returns the context error.
   554  func (s *Stream) Close() error {
   555  	s.CloseRead()
   556  	if s.IsReadOnly() {
   557  		return nil
   558  	}
   559  	s.CloseWrite()
   560  	// TODO: Return code from peer's RESET_STREAM frame?
   561  	if err := s.conn.waitOnDone(s.outctx, s.outdone); err != nil {
   562  		return err
   563  	}
   564  	s.outgate.lock()
   565  	defer s.outUnlock()
   566  	if s.outclosed.isReceived() && s.outacked.isrange(0, s.out.end) {
   567  		return nil
   568  	}
   569  	return errors.New("stream reset")
   570  }
   571  
   572  // CloseRead aborts reads on the stream.
   573  // Any blocked reads will be unblocked and return errors.
   574  //
   575  // CloseRead notifies the peer that the stream has been closed for reading.
   576  // It does not wait for the peer to acknowledge the closure.
   577  // Use Close to wait for the peer's acknowledgement.
   578  func (s *Stream) CloseRead() {
   579  	if s.IsWriteOnly() {
   580  		return
   581  	}
   582  	s.ingate.lock()
   583  	if s.inset.isrange(0, s.insize) || s.inresetcode != -1 {
   584  		// We've already received all data from the peer,
   585  		// so there's no need to send STOP_SENDING.
   586  		// This is the same as saying we sent one and they got it.
   587  		s.inclosed.setReceived()
   588  	} else {
   589  		s.inclosed.set()
   590  	}
   591  	discarded := s.in.end - s.in.start
   592  	s.in.discardBefore(s.in.end)
   593  	s.inUnlock()
   594  	s.conn.handleStreamBytesReadOffLoop(discarded) // must be done with ingate unlocked
   595  }
   596  
   597  // CloseWrite aborts writes on the stream.
   598  // Any blocked writes will be unblocked and return errors.
   599  //
   600  // CloseWrite sends any data in the stream write buffer to the peer.
   601  // It does not wait for the peer to acknowledge receipt of the data.
   602  // Use Close to wait for the peer's acknowledgement.
   603  func (s *Stream) CloseWrite() {
   604  	if s.IsReadOnly() {
   605  		return
   606  	}
   607  	s.outgate.lock()
   608  	defer s.outUnlock()
   609  	s.outclosed.set()
   610  	s.flushLocked()
   611  }
   612  
   613  // Reset aborts writes on the stream and notifies the peer
   614  // that the stream was terminated abruptly.
   615  // Any blocked writes will be unblocked and return errors.
   616  //
   617  // Reset sends the application protocol error code, which must be
   618  // less than 2^62, to the peer.
   619  // It does not wait for the peer to acknowledge receipt of the error.
   620  // Use Close to wait for the peer's acknowledgement.
   621  //
   622  // Reset does not affect reads.
   623  // Use CloseRead to abort reads on the stream.
   624  func (s *Stream) Reset(code uint64) {
   625  	const userClosed = true
   626  	s.resetInternal(code, userClosed)
   627  }
   628  
   629  // resetInternal resets the send side of the stream.
   630  //
   631  // If userClosed is true, this is s.Reset.
   632  // If userClosed is false, this is a reaction to a STOP_SENDING frame.
   633  func (s *Stream) resetInternal(code uint64, userClosed bool) {
   634  	s.outgate.lock()
   635  	defer s.outUnlock()
   636  	if s.IsReadOnly() {
   637  		return
   638  	}
   639  	if userClosed {
   640  		// Mark that the user closed the stream.
   641  		s.outclosed.set()
   642  	}
   643  	if s.outreset.isSet() {
   644  		return
   645  	}
   646  	if code > quicwire.MaxVarint {
   647  		code = quicwire.MaxVarint
   648  	}
   649  	// We could check here to see if the stream is closed and the
   650  	// peer has acked all the data and the FIN, but sending an
   651  	// extra RESET_STREAM in this case is harmless.
   652  	s.outreset.set()
   653  	s.outresetcode = code
   654  	s.outbufmu.Lock()
   655  	s.outbuf = nil
   656  	s.outbufoff = 0
   657  	s.outbufmu.Unlock()
   658  	s.out.discardBefore(s.out.end)
   659  	s.outunsent = rangeset[int64]{}
   660  	s.outblocked.clear()
   661  }
   662  
   663  // connHasClosed indicates the stream's conn has closed.
   664  func (s *Stream) connHasClosed() {
   665  	// If we're in the closing state, the user closed the conn.
   666  	// Otherwise, we the peer initiated the close.
   667  	// This only matters for the error we're going to return from stream operations.
   668  	localClose := s.conn.lifetime.state == connStateClosing
   669  
   670  	s.ingate.lock()
   671  	if !s.inset.isrange(0, s.insize) && s.inresetcode == -1 {
   672  		if localClose {
   673  			s.inclosed.set()
   674  		} else {
   675  			s.inresetcode = streamResetByConnClose
   676  		}
   677  	}
   678  	s.inUnlock()
   679  
   680  	s.outgate.lock()
   681  	if localClose {
   682  		s.outclosed.set()
   683  		s.outreset.set()
   684  	} else {
   685  		s.outresetcode = streamResetByConnClose
   686  		s.outreset.setReceived()
   687  	}
   688  	s.outUnlock()
   689  }
   690  
   691  // inUnlock unlocks s.ingate.
   692  // It sets the gate condition if reads from s will not block.
   693  // If s has receive-related frames to write or if both directions
   694  // are done and the stream should be removed, it notifies the Conn.
   695  func (s *Stream) inUnlock() {
   696  	state := s.inUnlockNoQueue()
   697  	s.conn.maybeQueueStreamForSend(s, state)
   698  }
   699  
   700  // inUnlockNoQueue is inUnlock,
   701  // but reports whether s has frames to write rather than notifying the Conn.
   702  func (s *Stream) inUnlockNoQueue() streamState {
   703  	nextByte := s.in.start + int64(len(s.inbuf))
   704  	canRead := s.inset.contains(nextByte) || // data available to read
   705  		s.insize == s.in.start+int64(len(s.inbuf)) || // at EOF
   706  		s.inresetcode != -1 || // reset by peer
   707  		s.inclosed.isSet() // closed locally
   708  	defer s.ingate.unlock(canRead)
   709  	var state streamState
   710  	switch {
   711  	case s.IsWriteOnly():
   712  		state = streamInDone
   713  	case s.inresetcode != -1: // reset by peer
   714  		fallthrough
   715  	case s.in.start == s.insize: // all data received and read
   716  		// We don't increase MAX_STREAMS until the user calls ReadClose or Close,
   717  		// so the receive side is not finished until inclosed is set.
   718  		if s.inclosed.isSet() {
   719  			state = streamInDone
   720  		}
   721  	case s.insendmax.shouldSend(): // STREAM_MAX_DATA
   722  		state = streamInSendMeta
   723  	case s.inclosed.shouldSend(): // STOP_SENDING
   724  		state = streamInSendMeta
   725  	}
   726  	const mask = streamInDone | streamInSendMeta
   727  	return s.state.set(state, mask)
   728  }
   729  
   730  // outUnlock unlocks s.outgate.
   731  // It sets the gate condition if writes to s will not block.
   732  // If s has send-related frames to write or if both directions
   733  // are done and the stream should be removed, it notifies the Conn.
   734  func (s *Stream) outUnlock() {
   735  	state := s.outUnlockNoQueue()
   736  	s.conn.maybeQueueStreamForSend(s, state)
   737  }
   738  
   739  // outUnlockNoQueue is outUnlock,
   740  // but reports whether s has frames to write rather than notifying the Conn.
   741  func (s *Stream) outUnlockNoQueue() streamState {
   742  	isDone := s.outclosed.isReceived() && s.outacked.isrange(0, s.out.end) || // all data acked
   743  		s.outreset.isSet() // reset locally
   744  	if isDone {
   745  		select {
   746  		case <-s.outdone:
   747  		default:
   748  			if !s.IsReadOnly() {
   749  				close(s.outdone)
   750  			}
   751  		}
   752  	}
   753  	lim := s.out.start + s.outmaxbuf
   754  	canWrite := lim > s.out.end || // available send buffer
   755  		s.outclosed.isSet() || // closed locally
   756  		s.outreset.isSet() // reset locally
   757  	defer s.outgate.unlock(canWrite)
   758  	var state streamState
   759  	switch {
   760  	case s.IsReadOnly():
   761  		state = streamOutDone
   762  	case s.outclosed.isReceived() && s.outacked.isrange(0, s.out.end): // all data sent and acked
   763  		fallthrough
   764  	case s.outreset.isReceived(): // RESET_STREAM sent and acked
   765  		// We don't increase MAX_STREAMS until the user calls WriteClose or Close,
   766  		// so the send side is not finished until outclosed is set.
   767  		if s.outclosed.isSet() {
   768  			state = streamOutDone
   769  		}
   770  	case s.outreset.shouldSend(): // RESET_STREAM
   771  		state = streamOutSendMeta
   772  	case s.outreset.isSet(): // RESET_STREAM sent but not acknowledged
   773  	case s.outblocked.shouldSend(): // STREAM_DATA_BLOCKED
   774  		state = streamOutSendMeta
   775  	case len(s.outunsent) > 0: // STREAM frame with data
   776  		if s.outunsent.min() < s.outmaxsent {
   777  			state = streamOutSendMeta // resent data, will not consume flow control
   778  		} else {
   779  			state = streamOutSendData // new data, requires flow control
   780  		}
   781  	case s.outclosed.shouldSend() && s.out.end == s.outmaxsent: // empty STREAM frame with FIN bit
   782  		state = streamOutSendMeta
   783  	case s.outopened.shouldSend(): // STREAM frame with no data
   784  		state = streamOutSendMeta
   785  	}
   786  	const mask = streamOutDone | streamOutSendMeta | streamOutSendData
   787  	return s.state.set(state, mask)
   788  }
   789  
   790  // handleData handles data received in a STREAM frame.
   791  func (s *Stream) handleData(off int64, b []byte, fin bool) error {
   792  	s.ingate.lock()
   793  	defer s.inUnlock()
   794  	end := off + int64(len(b))
   795  	if err := s.checkStreamBounds(end, fin); err != nil {
   796  		return err
   797  	}
   798  	if s.inclosed.isSet() || s.inresetcode != -1 {
   799  		// The user read-closed the stream, or the peer reset it.
   800  		// Either way, we can discard this frame.
   801  		return nil
   802  	}
   803  	if s.insize == -1 && end > s.in.end {
   804  		added := end - s.in.end
   805  		if err := s.conn.handleStreamBytesReceived(added); err != nil {
   806  			return err
   807  		}
   808  	}
   809  	if len(s.inset) > 0 && s.inset[0].contains(off) {
   810  		// We've received at least some of this data,
   811  		// and potentially moved it into s.inbuf
   812  		// (since it's part of the first range of received data).
   813  		// Avoid rewriting this data into s.in, since doing so could race
   814  		// with a reader reading the same data.
   815  		//
   816  		// (Note: We could apply additional checks here, to detect the peer
   817  		// sending us different data than we received the first time.
   818  		// We currently don't bother.)
   819  		newOff := min(end, s.inset[0].end)
   820  		b = b[newOff-off:]
   821  		off = newOff
   822  	}
   823  	s.in.writeAt(b, off)
   824  	s.inset.add(off, end)
   825  	if fin {
   826  		s.insize = end
   827  		// The peer has enough flow control window to send the entire stream.
   828  		s.insendmax.clear()
   829  	}
   830  	return nil
   831  }
   832  
   833  // handleReset handles a RESET_STREAM frame.
   834  func (s *Stream) handleReset(code uint64, finalSize int64) error {
   835  	s.ingate.lock()
   836  	defer s.inUnlock()
   837  	const fin = true
   838  	if err := s.checkStreamBounds(finalSize, fin); err != nil {
   839  		return err
   840  	}
   841  	if s.inresetcode != -1 {
   842  		// The stream was already reset.
   843  		return nil
   844  	}
   845  	if s.insize == -1 {
   846  		added := finalSize - s.in.end
   847  		if err := s.conn.handleStreamBytesReceived(added); err != nil {
   848  			return err
   849  		}
   850  	}
   851  	s.conn.handleStreamBytesReadOnLoop(finalSize - s.in.start)
   852  	s.in.discardBefore(s.in.end)
   853  	s.inresetcode = int64(code)
   854  	s.insize = finalSize
   855  	return nil
   856  }
   857  
   858  // checkStreamBounds validates the stream offset in a STREAM or RESET_STREAM frame.
   859  func (s *Stream) checkStreamBounds(end int64, fin bool) error {
   860  	if end > s.inwin {
   861  		// The peer sent us data past the maximum flow control window we gave them.
   862  		return localTransportError{
   863  			code:   errFlowControl,
   864  			reason: "stream flow control window exceeded",
   865  		}
   866  	}
   867  	if s.insize != -1 && end > s.insize {
   868  		// The peer sent us data past the final size of the stream they previously gave us.
   869  		return localTransportError{
   870  			code:   errFinalSize,
   871  			reason: "data received past end of stream",
   872  		}
   873  	}
   874  	if fin && s.insize != -1 && end != s.insize {
   875  		// The peer changed the final size of the stream.
   876  		return localTransportError{
   877  			code:   errFinalSize,
   878  			reason: "final size of stream changed",
   879  		}
   880  	}
   881  	if fin && end < s.in.end {
   882  		// The peer has previously sent us data past the final size.
   883  		return localTransportError{
   884  			code:   errFinalSize,
   885  			reason: "end of stream occurs before prior data",
   886  		}
   887  	}
   888  	return nil
   889  }
   890  
   891  // handleStopSending handles a STOP_SENDING frame.
   892  func (s *Stream) handleStopSending(code uint64) error {
   893  	// Peer requests that we reset this stream.
   894  	// https://www.rfc-editor.org/rfc/rfc9000#section-3.5-4
   895  	const userReset = false
   896  	s.resetInternal(code, userReset)
   897  	return nil
   898  }
   899  
   900  // handleMaxStreamData handles an update received in a MAX_STREAM_DATA frame.
   901  func (s *Stream) handleMaxStreamData(maxStreamData int64) error {
   902  	s.outgate.lock()
   903  	defer s.outUnlock()
   904  	if maxStreamData <= s.outwin {
   905  		return nil
   906  	}
   907  	if s.outflushed > s.outwin {
   908  		s.outunsent.add(s.outwin, min(maxStreamData, s.outflushed))
   909  	}
   910  	s.outwin = maxStreamData
   911  	if s.out.end > s.outwin {
   912  		// We've still got more data than flow control window.
   913  		s.outblocked.setUnsent()
   914  	} else {
   915  		s.outblocked.clear()
   916  	}
   917  	return nil
   918  }
   919  
   920  // ackOrLoss handles the fate of stream frames other than STREAM.
   921  func (s *Stream) ackOrLoss(pnum packetNumber, ftype byte, fate packetFate) {
   922  	// Frames which carry new information each time they are sent
   923  	// (MAX_STREAM_DATA, STREAM_DATA_BLOCKED) must only be marked
   924  	// as received if the most recent packet carrying this frame is acked.
   925  	//
   926  	// Frames which are always the same (STOP_SENDING, RESET_STREAM)
   927  	// can be marked as received if any packet carrying this frame is acked.
   928  	switch ftype {
   929  	case frameTypeResetStream:
   930  		s.outgate.lock()
   931  		s.outreset.ackOrLoss(pnum, fate)
   932  		s.outUnlock()
   933  	case frameTypeStopSending:
   934  		s.ingate.lock()
   935  		s.inclosed.ackOrLoss(pnum, fate)
   936  		s.inUnlock()
   937  	case frameTypeMaxStreamData:
   938  		s.ingate.lock()
   939  		s.insendmax.ackLatestOrLoss(pnum, fate)
   940  		s.inUnlock()
   941  	case frameTypeStreamDataBlocked:
   942  		s.outgate.lock()
   943  		s.outblocked.ackLatestOrLoss(pnum, fate)
   944  		s.outUnlock()
   945  	default:
   946  		panic("unhandled frame type")
   947  	}
   948  }
   949  
   950  // ackOrLossData handles the fate of a STREAM frame.
   951  func (s *Stream) ackOrLossData(pnum packetNumber, start, end int64, fin bool, fate packetFate) {
   952  	s.outgate.lock()
   953  	defer s.outUnlock()
   954  	s.outopened.ackOrLoss(pnum, fate)
   955  	if fin {
   956  		s.outclosed.ackOrLoss(pnum, fate)
   957  	}
   958  	if s.outreset.isSet() {
   959  		// If the stream has been reset, we don't care any more.
   960  		return
   961  	}
   962  	switch fate {
   963  	case packetAcked:
   964  		s.outacked.add(start, end)
   965  		s.outunsent.sub(start, end)
   966  		// If this ack is for data at the start of the send buffer, we can now discard it.
   967  		if s.outacked.contains(s.out.start) {
   968  			s.out.discardBefore(s.outacked[0].end)
   969  		}
   970  	case packetLost:
   971  		// Mark everything lost, but not previously acked, as needing retransmission.
   972  		// We do this by adding all the lost bytes to outunsent, and then
   973  		// removing everything already acked.
   974  		s.outunsent.add(start, end)
   975  		for _, a := range s.outacked {
   976  			s.outunsent.sub(a.start, a.end)
   977  		}
   978  	}
   979  }
   980  
   981  // appendInFramesLocked appends STOP_SENDING and MAX_STREAM_DATA frames
   982  // to the current packet.
   983  //
   984  // It returns true if no more frames need appending,
   985  // false if not everything fit in the current packet.
   986  func (s *Stream) appendInFramesLocked(w *packetWriter, pnum packetNumber, pto bool) bool {
   987  	if s.inclosed.shouldSendPTO(pto) {
   988  		// We don't currently have an API for setting the error code.
   989  		// Just send zero.
   990  		code := uint64(0)
   991  		if !w.appendStopSendingFrame(s.id, code) {
   992  			return false
   993  		}
   994  		s.inclosed.setSent(pnum)
   995  	}
   996  	// TODO: STOP_SENDING
   997  	if s.insendmax.shouldSendPTO(pto) {
   998  		// MAX_STREAM_DATA
   999  		maxStreamData := s.in.start + s.inmaxbuf
  1000  		if !w.appendMaxStreamDataFrame(s.id, maxStreamData) {
  1001  			return false
  1002  		}
  1003  		s.inwin = maxStreamData
  1004  		s.insendmax.setSent(pnum)
  1005  	}
  1006  	return true
  1007  }
  1008  
  1009  // appendOutFramesLocked appends RESET_STREAM, STREAM_DATA_BLOCKED, and STREAM frames
  1010  // to the current packet.
  1011  //
  1012  // It returns true if no more frames need appending,
  1013  // false if not everything fit in the current packet.
  1014  func (s *Stream) appendOutFramesLocked(w *packetWriter, pnum packetNumber, pto bool) bool {
  1015  	if s.outreset.isSet() {
  1016  		// RESET_STREAM
  1017  		if s.outreset.shouldSendPTO(pto) {
  1018  			if !w.appendResetStreamFrame(s.id, s.outresetcode, s.outmaxsent) {
  1019  				return false
  1020  			}
  1021  			s.outreset.setSent(pnum)
  1022  			s.frameOpensStream(pnum)
  1023  		}
  1024  		return true
  1025  	}
  1026  	if s.outblocked.shouldSendPTO(pto) {
  1027  		// STREAM_DATA_BLOCKED
  1028  		if !w.appendStreamDataBlockedFrame(s.id, s.outwin) {
  1029  			return false
  1030  		}
  1031  		s.outblocked.setSent(pnum)
  1032  		s.frameOpensStream(pnum)
  1033  	}
  1034  	for {
  1035  		// STREAM
  1036  		off, size := dataToSend(min(s.out.start, s.outwin), min(s.outflushed, s.outwin), s.outunsent, s.outacked, pto)
  1037  		if end := off + size; end > s.outmaxsent {
  1038  			// This will require connection-level flow control to send.
  1039  			end = min(end, s.outmaxsent+s.conn.streams.outflow.avail())
  1040  			end = max(end, off)
  1041  			size = end - off
  1042  		}
  1043  		fin := s.outclosed.isSet() && off+size == s.out.end
  1044  		shouldSend := size > 0 || // have data to send
  1045  			s.outopened.shouldSendPTO(pto) || // should open the stream
  1046  			(fin && s.outclosed.shouldSendPTO(pto)) // should close the stream
  1047  		if !shouldSend {
  1048  			return true
  1049  		}
  1050  		b, added := w.appendStreamFrame(s.id, off, int(size), fin)
  1051  		if !added {
  1052  			return false
  1053  		}
  1054  		s.out.copy(off, b)
  1055  		end := off + int64(len(b))
  1056  		if end > s.outmaxsent {
  1057  			s.conn.streams.outflow.consume(end - s.outmaxsent)
  1058  			s.outmaxsent = end
  1059  		}
  1060  		s.outunsent.sub(off, end)
  1061  		s.frameOpensStream(pnum)
  1062  		if fin {
  1063  			s.outclosed.setSent(pnum)
  1064  		}
  1065  		if pto {
  1066  			return true
  1067  		}
  1068  		if int64(len(b)) < size {
  1069  			return false
  1070  		}
  1071  	}
  1072  }
  1073  
  1074  // frameOpensStream records that we're sending a frame that will open the stream.
  1075  //
  1076  // If we don't have an acknowledgement from the peer for a previous frame opening the stream,
  1077  // record this packet as being the latest one to open it.
  1078  func (s *Stream) frameOpensStream(pnum packetNumber) {
  1079  	if !s.outopened.isReceived() {
  1080  		s.outopened.setSent(pnum)
  1081  	}
  1082  }
  1083  
  1084  // dataToSend returns the next range of data to send in a STREAM or CRYPTO_STREAM.
  1085  func dataToSend(start, end int64, outunsent, outacked rangeset[int64], pto bool) (sendStart, size int64) {
  1086  	switch {
  1087  	case pto:
  1088  		// On PTO, resend unacked data that fits in the probe packet.
  1089  		// For simplicity, we send the range starting at s.out.start
  1090  		// (which is definitely unacked, or else we would have discarded it)
  1091  		// up to the next acked byte (if any).
  1092  		//
  1093  		// This may miss unacked data starting after that acked byte,
  1094  		// but avoids resending data the peer has acked.
  1095  		for _, r := range outacked {
  1096  			if r.start > start {
  1097  				return start, r.start - start
  1098  			}
  1099  		}
  1100  		return start, end - start
  1101  	case outunsent.numRanges() > 0:
  1102  		return outunsent.min(), outunsent[0].size()
  1103  	default:
  1104  		return end, 0
  1105  	}
  1106  }
  1107  

View as plain text