Source file src/golang.org/x/net/quic/conn_flow_test.go

Documentation: golang.org/x/net/quic

     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  //go:build go1.21
     6  
     7  package quic
     8  
     9  import (
    10  	"context"
    11  	"testing"
    12  )
    13  
    14  func TestConnInflowReturnOnRead(t *testing.T) {
    15  	tc, s := newTestConnAndRemoteStream(t, serverSide, uniStream, func(c *Config) {
    16  		c.MaxConnReadBufferSize = 64
    17  	})
    18  	tc.writeFrames(packetType1RTT, debugFrameStream{
    19  		id:   s.id,
    20  		data: make([]byte, 8),
    21  	})
    22  	if n, err := s.Read(make([]byte, 8)); n != 8 || err != nil {
    23  		t.Fatalf("s.Read() = %v, %v; want %v, nil", n, err, 8)
    24  	}
    25  	tc.wantFrame("available window increases, send a MAX_DATA",
    26  		packetType1RTT, debugFrameMaxData{
    27  			max: 64 + 8,
    28  		})
    29  	// Peer can write up to the new limit.
    30  	tc.writeFrames(packetType1RTT, debugFrameStream{
    31  		id:   s.id,
    32  		off:  8,
    33  		data: make([]byte, 64),
    34  	})
    35  	if n, err := s.Read(make([]byte, 64+1)); n != 64 {
    36  		t.Fatalf("s.Read() = %v, %v; want %v, anything", n, err, 64)
    37  	}
    38  	tc.wantFrame("available window increases, send a MAX_DATA",
    39  		packetType1RTT, debugFrameMaxData{
    40  			max: 64 + 8 + 64,
    41  		})
    42  	tc.wantIdle("connection is idle")
    43  }
    44  
    45  func TestConnInflowReturnOnRacingReads(t *testing.T) {
    46  	// Perform two reads at the same time,
    47  	// one for half of MaxConnReadBufferSize
    48  	// and one for one byte.
    49  	//
    50  	// We should observe a single MAX_DATA update.
    51  	// Depending on the ordering of events,
    52  	// this may include the credit from just the larger read
    53  	// or the credit from both.
    54  	ctx := canceledContext()
    55  	tc := newTestConn(t, serverSide, func(c *Config) {
    56  		c.MaxConnReadBufferSize = 64
    57  	})
    58  	tc.handshake()
    59  	tc.ignoreFrame(frameTypeAck)
    60  	tc.writeFrames(packetType1RTT, debugFrameStream{
    61  		id:   newStreamID(clientSide, uniStream, 0),
    62  		data: make([]byte, 16),
    63  	})
    64  	tc.writeFrames(packetType1RTT, debugFrameStream{
    65  		id:   newStreamID(clientSide, uniStream, 1),
    66  		data: make([]byte, 1),
    67  	})
    68  	s1, err := tc.conn.AcceptStream(ctx)
    69  	if err != nil {
    70  		t.Fatalf("conn.AcceptStream() = %v", err)
    71  	}
    72  	s2, err := tc.conn.AcceptStream(ctx)
    73  	if err != nil {
    74  		t.Fatalf("conn.AcceptStream() = %v", err)
    75  	}
    76  	read1 := runAsync(tc, func(ctx context.Context) (int, error) {
    77  		return s1.Read(make([]byte, 16))
    78  	})
    79  	read2 := runAsync(tc, func(ctx context.Context) (int, error) {
    80  		return s2.Read(make([]byte, 1))
    81  	})
    82  	// This MAX_DATA might extend the window by 16 or 17, depending on
    83  	// whether the second write occurs before the update happens.
    84  	tc.wantFrameType("MAX_DATA update is sent",
    85  		packetType1RTT, debugFrameMaxData{})
    86  	tc.wantIdle("redundant MAX_DATA is not sent")
    87  	if _, err := read1.result(); err != nil {
    88  		t.Errorf("Read #1 = %v", err)
    89  	}
    90  	if _, err := read2.result(); err != nil {
    91  		t.Errorf("Read #2 = %v", err)
    92  	}
    93  }
    94  
    95  func TestConnInflowReturnOnClose(t *testing.T) {
    96  	tc, s := newTestConnAndRemoteStream(t, serverSide, uniStream, func(c *Config) {
    97  		c.MaxConnReadBufferSize = 64
    98  	})
    99  	tc.ignoreFrame(frameTypeStopSending)
   100  	tc.writeFrames(packetType1RTT, debugFrameStream{
   101  		id:   s.id,
   102  		data: make([]byte, 64),
   103  	})
   104  	s.CloseRead()
   105  	tc.wantFrame("closing stream updates connection-level flow control",
   106  		packetType1RTT, debugFrameMaxData{
   107  			max: 128,
   108  		})
   109  }
   110  
   111  func TestConnInflowReturnOnReset(t *testing.T) {
   112  	tc, s := newTestConnAndRemoteStream(t, serverSide, uniStream, func(c *Config) {
   113  		c.MaxConnReadBufferSize = 64
   114  	})
   115  	tc.ignoreFrame(frameTypeStopSending)
   116  	tc.writeFrames(packetType1RTT, debugFrameStream{
   117  		id:   s.id,
   118  		data: make([]byte, 32),
   119  	})
   120  	tc.writeFrames(packetType1RTT, debugFrameResetStream{
   121  		id:        s.id,
   122  		finalSize: 64,
   123  	})
   124  	s.CloseRead()
   125  	tc.wantFrame("receiving stream reseet updates connection-level flow control",
   126  		packetType1RTT, debugFrameMaxData{
   127  			max: 128,
   128  		})
   129  }
   130  
   131  func TestConnInflowStreamViolation(t *testing.T) {
   132  	tc := newTestConn(t, serverSide, func(c *Config) {
   133  		c.MaxConnReadBufferSize = 100
   134  	})
   135  	tc.handshake()
   136  	tc.ignoreFrame(frameTypeAck)
   137  	// Total MAX_DATA consumed: 50
   138  	tc.writeFrames(packetType1RTT, debugFrameStream{
   139  		id:   newStreamID(clientSide, bidiStream, 0),
   140  		data: make([]byte, 50),
   141  	})
   142  	// Total MAX_DATA consumed: 80
   143  	tc.writeFrames(packetType1RTT, debugFrameStream{
   144  		id:   newStreamID(clientSide, uniStream, 0),
   145  		off:  20,
   146  		data: make([]byte, 10),
   147  	})
   148  	// Total MAX_DATA consumed: 100
   149  	tc.writeFrames(packetType1RTT, debugFrameStream{
   150  		id:  newStreamID(clientSide, bidiStream, 0),
   151  		off: 70,
   152  		fin: true,
   153  	})
   154  	// This stream has already consumed quota for these bytes.
   155  	// Total MAX_DATA consumed: 100
   156  	tc.writeFrames(packetType1RTT, debugFrameStream{
   157  		id:   newStreamID(clientSide, uniStream, 0),
   158  		data: make([]byte, 20),
   159  	})
   160  	tc.wantIdle("peer has consumed all MAX_DATA quota")
   161  
   162  	// Total MAX_DATA consumed: 101
   163  	tc.writeFrames(packetType1RTT, debugFrameStream{
   164  		id:   newStreamID(clientSide, bidiStream, 2),
   165  		data: make([]byte, 1),
   166  	})
   167  	tc.wantFrame("peer violates MAX_DATA limit",
   168  		packetType1RTT, debugFrameConnectionCloseTransport{
   169  			code: errFlowControl,
   170  		})
   171  }
   172  
   173  func TestConnInflowResetViolation(t *testing.T) {
   174  	tc := newTestConn(t, serverSide, func(c *Config) {
   175  		c.MaxConnReadBufferSize = 100
   176  	})
   177  	tc.handshake()
   178  	tc.ignoreFrame(frameTypeAck)
   179  	tc.writeFrames(packetType1RTT, debugFrameStream{
   180  		id:   newStreamID(clientSide, bidiStream, 0),
   181  		data: make([]byte, 100),
   182  	})
   183  	tc.wantIdle("peer has consumed all MAX_DATA quota")
   184  
   185  	tc.writeFrames(packetType1RTT, debugFrameResetStream{
   186  		id:        newStreamID(clientSide, uniStream, 0),
   187  		finalSize: 0,
   188  	})
   189  	tc.wantIdle("stream reset does not consume MAX_DATA quota, no error")
   190  
   191  	tc.writeFrames(packetType1RTT, debugFrameResetStream{
   192  		id:        newStreamID(clientSide, uniStream, 1),
   193  		finalSize: 1,
   194  	})
   195  	tc.wantFrame("RESET_STREAM final size violates MAX_DATA limit",
   196  		packetType1RTT, debugFrameConnectionCloseTransport{
   197  			code: errFlowControl,
   198  		})
   199  }
   200  
   201  func TestConnInflowMultipleStreams(t *testing.T) {
   202  	tc := newTestConn(t, serverSide, func(c *Config) {
   203  		c.MaxConnReadBufferSize = 128
   204  	})
   205  	tc.handshake()
   206  	tc.ignoreFrame(frameTypeAck)
   207  
   208  	var streams []*Stream
   209  	for _, id := range []streamID{
   210  		newStreamID(clientSide, uniStream, 0),
   211  		newStreamID(clientSide, uniStream, 1),
   212  		newStreamID(clientSide, bidiStream, 0),
   213  		newStreamID(clientSide, bidiStream, 1),
   214  	} {
   215  		tc.writeFrames(packetType1RTT, debugFrameStream{
   216  			id:   id,
   217  			data: make([]byte, 1),
   218  		})
   219  		s := tc.acceptStream()
   220  		streams = append(streams, s)
   221  		if n, err := s.Read(make([]byte, 1)); err != nil || n != 1 {
   222  			t.Fatalf("s.Read() = %v, %v; want 1, nil", n, err)
   223  		}
   224  	}
   225  	tc.wantIdle("streams have read data, but not enough to update MAX_DATA")
   226  
   227  	for _, s := range streams {
   228  		tc.writeFrames(packetType1RTT, debugFrameStream{
   229  			id:   s.id,
   230  			off:  1,
   231  			data: make([]byte, 31),
   232  		})
   233  	}
   234  
   235  	if n, err := streams[0].Read(make([]byte, 32)); n != 31 {
   236  		t.Fatalf("s.Read() = %v, %v; want 31, anything", n, err)
   237  	}
   238  	tc.wantFrame("read enough data to trigger a MAX_DATA update",
   239  		packetType1RTT, debugFrameMaxData{
   240  			max: 128 + 32 + 1 + 1 + 1,
   241  		})
   242  
   243  	tc.ignoreFrame(frameTypeStopSending)
   244  	streams[2].CloseRead()
   245  	tc.wantFrame("closed stream triggers another MAX_DATA update",
   246  		packetType1RTT, debugFrameMaxData{
   247  			max: 128 + 32 + 1 + 32 + 1,
   248  		})
   249  }
   250  
   251  func TestConnOutflowBlocked(t *testing.T) {
   252  	tc, s := newTestConnAndLocalStream(t, clientSide, uniStream,
   253  		permissiveTransportParameters,
   254  		func(p *transportParameters) {
   255  			p.initialMaxData = 10
   256  		})
   257  	tc.ignoreFrame(frameTypeAck)
   258  
   259  	data := makeTestData(32)
   260  	n, err := s.Write(data)
   261  	if n != len(data) || err != nil {
   262  		t.Fatalf("s.Write() = %v, %v; want %v, nil", n, err, len(data))
   263  	}
   264  	s.Flush()
   265  
   266  	tc.wantFrame("stream writes data up to MAX_DATA limit",
   267  		packetType1RTT, debugFrameStream{
   268  			id:   s.id,
   269  			data: data[:10],
   270  		})
   271  	tc.wantIdle("stream is blocked by MAX_DATA limit")
   272  
   273  	tc.writeFrames(packetType1RTT, debugFrameMaxData{
   274  		max: 20,
   275  	})
   276  	tc.wantFrame("stream writes data up to new MAX_DATA limit",
   277  		packetType1RTT, debugFrameStream{
   278  			id:   s.id,
   279  			off:  10,
   280  			data: data[10:20],
   281  		})
   282  	tc.wantIdle("stream is blocked by new MAX_DATA limit")
   283  
   284  	tc.writeFrames(packetType1RTT, debugFrameMaxData{
   285  		max: 100,
   286  	})
   287  	tc.wantFrame("stream writes remaining data",
   288  		packetType1RTT, debugFrameStream{
   289  			id:   s.id,
   290  			off:  20,
   291  			data: data[20:],
   292  		})
   293  }
   294  
   295  func TestConnOutflowMaxDataDecreases(t *testing.T) {
   296  	tc, s := newTestConnAndLocalStream(t, clientSide, uniStream,
   297  		permissiveTransportParameters,
   298  		func(p *transportParameters) {
   299  			p.initialMaxData = 10
   300  		})
   301  	tc.ignoreFrame(frameTypeAck)
   302  
   303  	// Decrease in MAX_DATA is ignored.
   304  	tc.writeFrames(packetType1RTT, debugFrameMaxData{
   305  		max: 5,
   306  	})
   307  
   308  	data := makeTestData(32)
   309  	n, err := s.Write(data)
   310  	if n != len(data) || err != nil {
   311  		t.Fatalf("s.Write() = %v, %v; want %v, nil", n, err, len(data))
   312  	}
   313  	s.Flush()
   314  
   315  	tc.wantFrame("stream writes data up to MAX_DATA limit",
   316  		packetType1RTT, debugFrameStream{
   317  			id:   s.id,
   318  			data: data[:10],
   319  		})
   320  }
   321  
   322  func TestConnOutflowMaxDataRoundRobin(t *testing.T) {
   323  	ctx := canceledContext()
   324  	tc := newTestConn(t, clientSide, permissiveTransportParameters,
   325  		func(p *transportParameters) {
   326  			p.initialMaxData = 0
   327  		})
   328  	tc.handshake()
   329  	tc.ignoreFrame(frameTypeAck)
   330  
   331  	s1, err := tc.conn.newLocalStream(ctx, uniStream)
   332  	if err != nil {
   333  		t.Fatalf("conn.newLocalStream(%v) = %v", uniStream, err)
   334  	}
   335  	s2, err := tc.conn.newLocalStream(ctx, uniStream)
   336  	if err != nil {
   337  		t.Fatalf("conn.newLocalStream(%v) = %v", uniStream, err)
   338  	}
   339  
   340  	s1.Write(make([]byte, 10))
   341  	s1.Flush()
   342  	s2.Write(make([]byte, 10))
   343  	s2.Flush()
   344  
   345  	tc.writeFrames(packetType1RTT, debugFrameMaxData{
   346  		max: 1,
   347  	})
   348  	tc.wantFrame("stream 1 writes data up to MAX_DATA limit",
   349  		packetType1RTT, debugFrameStream{
   350  			id:   s1.id,
   351  			data: []byte{0},
   352  		})
   353  
   354  	tc.writeFrames(packetType1RTT, debugFrameMaxData{
   355  		max: 2,
   356  	})
   357  	tc.wantFrame("stream 2 writes data up to MAX_DATA limit",
   358  		packetType1RTT, debugFrameStream{
   359  			id:   s2.id,
   360  			data: []byte{0},
   361  		})
   362  
   363  	tc.writeFrames(packetType1RTT, debugFrameMaxData{
   364  		max: 3,
   365  	})
   366  	tc.wantFrame("stream 1 writes data up to MAX_DATA limit",
   367  		packetType1RTT, debugFrameStream{
   368  			id:   s1.id,
   369  			off:  1,
   370  			data: []byte{0},
   371  		})
   372  }
   373  
   374  func TestConnOutflowMetaAndData(t *testing.T) {
   375  	tc, s := newTestConnAndLocalStream(t, clientSide, bidiStream,
   376  		permissiveTransportParameters,
   377  		func(p *transportParameters) {
   378  			p.initialMaxData = 0
   379  		})
   380  	tc.ignoreFrame(frameTypeAck)
   381  
   382  	data := makeTestData(32)
   383  	s.Write(data)
   384  	s.Flush()
   385  
   386  	s.CloseRead()
   387  	tc.wantFrame("CloseRead sends a STOP_SENDING, not flow controlled",
   388  		packetType1RTT, debugFrameStopSending{
   389  			id: s.id,
   390  		})
   391  
   392  	tc.writeFrames(packetType1RTT, debugFrameMaxData{
   393  		max: 100,
   394  	})
   395  	tc.wantFrame("unblocked MAX_DATA",
   396  		packetType1RTT, debugFrameStream{
   397  			id:   s.id,
   398  			data: data,
   399  		})
   400  }
   401  
   402  func TestConnOutflowResentData(t *testing.T) {
   403  	tc, s := newTestConnAndLocalStream(t, clientSide, bidiStream,
   404  		permissiveTransportParameters,
   405  		func(p *transportParameters) {
   406  			p.initialMaxData = 10
   407  		})
   408  	tc.ignoreFrame(frameTypeAck)
   409  
   410  	data := makeTestData(15)
   411  	s.Write(data[:8])
   412  	s.Flush()
   413  	tc.wantFrame("data is under MAX_DATA limit, all sent",
   414  		packetType1RTT, debugFrameStream{
   415  			id:   s.id,
   416  			data: data[:8],
   417  		})
   418  
   419  	// Lose the last STREAM packet.
   420  	const pto = false
   421  	tc.triggerLossOrPTO(packetType1RTT, false)
   422  	tc.wantFrame("lost STREAM data is retransmitted",
   423  		packetType1RTT, debugFrameStream{
   424  			id:   s.id,
   425  			data: data[:8],
   426  		})
   427  
   428  	s.Write(data[8:])
   429  	s.Flush()
   430  	tc.wantFrame("new data is sent up to the MAX_DATA limit",
   431  		packetType1RTT, debugFrameStream{
   432  			id:   s.id,
   433  			off:  8,
   434  			data: data[8:10],
   435  		})
   436  }
   437  

View as plain text