conn_flow_test.go

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