conntest.go

     1  // Copyright 2016 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 nettest
     6  
     7  import (
     8  	"bytes"
     9  	"encoding/binary"
    10  	"io"
    11  	"math/rand"
    12  	"net"
    13  	"runtime"
    14  	"sync"
    15  	"testing"
    16  	"time"
    17  )
    18  
    19  // MakePipe creates a connection between two endpoints and returns the pair
    20  // as c1 and c2, such that anything written to c1 is read by c2 and vice-versa.
    21  // The stop function closes all resources, including c1, c2, and the underlying
    22  // net.Listener (if there is one), and should not be nil.
    23  type MakePipe func() (c1, c2 net.Conn, stop func(), err error)
    24  
    25  // TestConn tests that a net.Conn implementation properly satisfies the interface.
    26  // The tests should not produce any false positives, but may experience
    27  // false negatives. Thus, some issues may only be detected when the test is
    28  // run multiple times. For maximal effectiveness, run the tests under the
    29  // race detector.
    30  func TestConn(t *testing.T, mp MakePipe) {
    31  	t.Run("BasicIO", func(t *testing.T) { timeoutWrapper(t, mp, testBasicIO) })
    32  	t.Run("PingPong", func(t *testing.T) { timeoutWrapper(t, mp, testPingPong) })
    33  	t.Run("RacyRead", func(t *testing.T) { timeoutWrapper(t, mp, testRacyRead) })
    34  	t.Run("RacyWrite", func(t *testing.T) { timeoutWrapper(t, mp, testRacyWrite) })
    35  	t.Run("ReadTimeout", func(t *testing.T) { timeoutWrapper(t, mp, testReadTimeout) })
    36  	t.Run("WriteTimeout", func(t *testing.T) { timeoutWrapper(t, mp, testWriteTimeout) })
    37  	t.Run("PastTimeout", func(t *testing.T) { timeoutWrapper(t, mp, testPastTimeout) })
    38  	t.Run("PresentTimeout", func(t *testing.T) { timeoutWrapper(t, mp, testPresentTimeout) })
    39  	t.Run("FutureTimeout", func(t *testing.T) { timeoutWrapper(t, mp, testFutureTimeout) })
    40  	t.Run("CloseTimeout", func(t *testing.T) { timeoutWrapper(t, mp, testCloseTimeout) })
    41  	t.Run("ConcurrentMethods", func(t *testing.T) { timeoutWrapper(t, mp, testConcurrentMethods) })
    42  }
    43  
    44  type connTester func(t *testing.T, c1, c2 net.Conn)
    45  
    46  func timeoutWrapper(t *testing.T, mp MakePipe, f connTester) {
    47  	t.Helper()
    48  	c1, c2, stop, err := mp()
    49  	if err != nil {
    50  		t.Fatalf("unable to make pipe: %v", err)
    51  	}
    52  	var once sync.Once
    53  	defer once.Do(func() { stop() })
    54  	timer := time.AfterFunc(time.Minute, func() {
    55  		once.Do(func() {
    56  			t.Error("test timed out; terminating pipe")
    57  			stop()
    58  		})
    59  	})
    60  	defer timer.Stop()
    61  	f(t, c1, c2)
    62  }
    63  
    64  // testBasicIO tests that the data sent on c1 is properly received on c2.
    65  func testBasicIO(t *testing.T, c1, c2 net.Conn) {
    66  	want := make([]byte, 1<<20)
    67  	rand.New(rand.NewSource(0)).Read(want)
    68  
    69  	dataCh := make(chan []byte)
    70  	go func() {
    71  		rd := bytes.NewReader(want)
    72  		if err := chunkedCopy(c1, rd); err != nil {
    73  			t.Errorf("unexpected c1.Write error: %v", err)
    74  		}
    75  		if err := c1.Close(); err != nil {
    76  			t.Errorf("unexpected c1.Close error: %v", err)
    77  		}
    78  	}()
    79  
    80  	go func() {
    81  		wr := new(bytes.Buffer)
    82  		if err := chunkedCopy(wr, c2); err != nil {
    83  			t.Errorf("unexpected c2.Read error: %v", err)
    84  		}
    85  		if err := c2.Close(); err != nil {
    86  			t.Errorf("unexpected c2.Close error: %v", err)
    87  		}
    88  		dataCh <- wr.Bytes()
    89  	}()
    90  
    91  	if got := <-dataCh; !bytes.Equal(got, want) {
    92  		t.Error("transmitted data differs")
    93  	}
    94  }
    95  
    96  // testPingPong tests that the two endpoints can synchronously send data to
    97  // each other in a typical request-response pattern.
    98  func testPingPong(t *testing.T, c1, c2 net.Conn) {
    99  	var wg sync.WaitGroup
   100  	defer wg.Wait()
   101  
   102  	pingPonger := func(c net.Conn) {
   103  		defer wg.Done()
   104  		buf := make([]byte, 8)
   105  		var prev uint64
   106  		for {
   107  			if _, err := io.ReadFull(c, buf); err != nil {
   108  				if err == io.EOF {
   109  					break
   110  				}
   111  				t.Errorf("unexpected Read error: %v", err)
   112  			}
   113  
   114  			v := binary.LittleEndian.Uint64(buf)
   115  			binary.LittleEndian.PutUint64(buf, v+1)
   116  			if prev != 0 && prev+2 != v {
   117  				t.Errorf("mismatching value: got %d, want %d", v, prev+2)
   118  			}
   119  			prev = v
   120  			if v == 1000 {
   121  				break
   122  			}
   123  
   124  			if _, err := c.Write(buf); err != nil {
   125  				t.Errorf("unexpected Write error: %v", err)
   126  				break
   127  			}
   128  		}
   129  		if err := c.Close(); err != nil {
   130  			t.Errorf("unexpected Close error: %v", err)
   131  		}
   132  	}
   133  
   134  	wg.Add(2)
   135  	go pingPonger(c1)
   136  	go pingPonger(c2)
   137  
   138  	// Start off the chain reaction.
   139  	if _, err := c1.Write(make([]byte, 8)); err != nil {
   140  		t.Errorf("unexpected c1.Write error: %v", err)
   141  	}
   142  }
   143  
   144  // testRacyRead tests that it is safe to mutate the input Read buffer
   145  // immediately after cancelation has occurred.
   146  func testRacyRead(t *testing.T, c1, c2 net.Conn) {
   147  	go chunkedCopy(c2, rand.New(rand.NewSource(0)))
   148  
   149  	var wg sync.WaitGroup
   150  	defer wg.Wait()
   151  
   152  	c1.SetReadDeadline(time.Now().Add(time.Millisecond))
   153  	for i := 0; i < 10; i++ {
   154  		wg.Add(1)
   155  		go func() {
   156  			defer wg.Done()
   157  
   158  			b1 := make([]byte, 1024)
   159  			b2 := make([]byte, 1024)
   160  			for j := 0; j < 100; j++ {
   161  				_, err := c1.Read(b1)
   162  				copy(b1, b2) // Mutate b1 to trigger potential race
   163  				if err != nil {
   164  					checkForTimeoutError(t, err)
   165  					c1.SetReadDeadline(time.Now().Add(time.Millisecond))
   166  				}
   167  			}
   168  		}()
   169  	}
   170  }
   171  
   172  // testRacyWrite tests that it is safe to mutate the input Write buffer
   173  // immediately after cancelation has occurred.
   174  func testRacyWrite(t *testing.T, c1, c2 net.Conn) {
   175  	go chunkedCopy(io.Discard, c2)
   176  
   177  	var wg sync.WaitGroup
   178  	defer wg.Wait()
   179  
   180  	c1.SetWriteDeadline(time.Now().Add(time.Millisecond))
   181  	for i := 0; i < 10; i++ {
   182  		wg.Add(1)
   183  		go func() {
   184  			defer wg.Done()
   185  
   186  			b1 := make([]byte, 1024)
   187  			b2 := make([]byte, 1024)
   188  			for j := 0; j < 100; j++ {
   189  				_, err := c1.Write(b1)
   190  				copy(b1, b2) // Mutate b1 to trigger potential race
   191  				if err != nil {
   192  					checkForTimeoutError(t, err)
   193  					c1.SetWriteDeadline(time.Now().Add(time.Millisecond))
   194  				}
   195  			}
   196  		}()
   197  	}
   198  }
   199  
   200  // testReadTimeout tests that Read timeouts do not affect Write.
   201  func testReadTimeout(t *testing.T, c1, c2 net.Conn) {
   202  	go chunkedCopy(io.Discard, c2)
   203  
   204  	c1.SetReadDeadline(aLongTimeAgo)
   205  	_, err := c1.Read(make([]byte, 1024))
   206  	checkForTimeoutError(t, err)
   207  	if _, err := c1.Write(make([]byte, 1024)); err != nil {
   208  		t.Errorf("unexpected Write error: %v", err)
   209  	}
   210  }
   211  
   212  // testWriteTimeout tests that Write timeouts do not affect Read.
   213  func testWriteTimeout(t *testing.T, c1, c2 net.Conn) {
   214  	go chunkedCopy(c2, rand.New(rand.NewSource(0)))
   215  
   216  	c1.SetWriteDeadline(aLongTimeAgo)
   217  	_, err := c1.Write(make([]byte, 1024))
   218  	checkForTimeoutError(t, err)
   219  	if _, err := c1.Read(make([]byte, 1024)); err != nil {
   220  		t.Errorf("unexpected Read error: %v", err)
   221  	}
   222  }
   223  
   224  // testPastTimeout tests that a deadline set in the past immediately times out
   225  // Read and Write requests.
   226  func testPastTimeout(t *testing.T, c1, c2 net.Conn) {
   227  	go chunkedCopy(c2, c2)
   228  
   229  	testRoundtrip(t, c1)
   230  
   231  	c1.SetDeadline(aLongTimeAgo)
   232  	n, err := c1.Write(make([]byte, 1024))
   233  	if n != 0 {
   234  		t.Errorf("unexpected Write count: got %d, want 0", n)
   235  	}
   236  	checkForTimeoutError(t, err)
   237  	n, err = c1.Read(make([]byte, 1024))
   238  	if n != 0 {
   239  		t.Errorf("unexpected Read count: got %d, want 0", n)
   240  	}
   241  	checkForTimeoutError(t, err)
   242  
   243  	testRoundtrip(t, c1)
   244  }
   245  
   246  // testPresentTimeout tests that a past deadline set while there are pending
   247  // Read and Write operations immediately times out those operations.
   248  func testPresentTimeout(t *testing.T, c1, c2 net.Conn) {
   249  	var wg sync.WaitGroup
   250  	defer wg.Wait()
   251  	wg.Add(3)
   252  
   253  	deadlineSet := make(chan bool, 1)
   254  	go func() {
   255  		defer wg.Done()
   256  		time.Sleep(100 * time.Millisecond)
   257  		deadlineSet <- true
   258  		c1.SetReadDeadline(aLongTimeAgo)
   259  		c1.SetWriteDeadline(aLongTimeAgo)
   260  	}()
   261  	go func() {
   262  		defer wg.Done()
   263  		n, err := c1.Read(make([]byte, 1024))
   264  		if n != 0 {
   265  			t.Errorf("unexpected Read count: got %d, want 0", n)
   266  		}
   267  		checkForTimeoutError(t, err)
   268  		if len(deadlineSet) == 0 {
   269  			t.Error("Read timed out before deadline is set")
   270  		}
   271  	}()
   272  	go func() {
   273  		defer wg.Done()
   274  		var err error
   275  		for err == nil {
   276  			_, err = c1.Write(make([]byte, 1024))
   277  		}
   278  		checkForTimeoutError(t, err)
   279  		if len(deadlineSet) == 0 {
   280  			t.Error("Write timed out before deadline is set")
   281  		}
   282  	}()
   283  }
   284  
   285  // testFutureTimeout tests that a future deadline will eventually time out
   286  // Read and Write operations.
   287  func testFutureTimeout(t *testing.T, c1, c2 net.Conn) {
   288  	var wg sync.WaitGroup
   289  	wg.Add(2)
   290  
   291  	c1.SetDeadline(time.Now().Add(100 * time.Millisecond))
   292  	go func() {
   293  		defer wg.Done()
   294  		_, err := c1.Read(make([]byte, 1024))
   295  		checkForTimeoutError(t, err)
   296  	}()
   297  	go func() {
   298  		defer wg.Done()
   299  		var err error
   300  		for err == nil {
   301  			_, err = c1.Write(make([]byte, 1024))
   302  		}
   303  		checkForTimeoutError(t, err)
   304  	}()
   305  	wg.Wait()
   306  
   307  	go chunkedCopy(c2, c2)
   308  	resyncConn(t, c1)
   309  	testRoundtrip(t, c1)
   310  }
   311  
   312  // testCloseTimeout tests that calling Close immediately times out pending
   313  // Read and Write operations.
   314  func testCloseTimeout(t *testing.T, c1, c2 net.Conn) {
   315  	go chunkedCopy(c2, c2)
   316  
   317  	var wg sync.WaitGroup
   318  	defer wg.Wait()
   319  	wg.Add(3)
   320  
   321  	// Test for cancelation upon connection closure.
   322  	c1.SetDeadline(neverTimeout)
   323  	go func() {
   324  		defer wg.Done()
   325  		time.Sleep(100 * time.Millisecond)
   326  		c1.Close()
   327  	}()
   328  	go func() {
   329  		defer wg.Done()
   330  		var err error
   331  		buf := make([]byte, 1024)
   332  		for err == nil {
   333  			_, err = c1.Read(buf)
   334  		}
   335  	}()
   336  	go func() {
   337  		defer wg.Done()
   338  		var err error
   339  		buf := make([]byte, 1024)
   340  		for err == nil {
   341  			_, err = c1.Write(buf)
   342  		}
   343  	}()
   344  }
   345  
   346  // testConcurrentMethods tests that the methods of net.Conn can safely
   347  // be called concurrently.
   348  func testConcurrentMethods(t *testing.T, c1, c2 net.Conn) {
   349  	if runtime.GOOS == "plan9" {
   350  		t.Skip("skipping on plan9; see https://golang.org/issue/20489")
   351  	}
   352  	go chunkedCopy(c2, c2)
   353  
   354  	// The results of the calls may be nonsensical, but this should
   355  	// not trigger a race detector warning.
   356  	var wg sync.WaitGroup
   357  	for i := 0; i < 100; i++ {
   358  		wg.Add(7)
   359  		go func() {
   360  			defer wg.Done()
   361  			c1.Read(make([]byte, 1024))
   362  		}()
   363  		go func() {
   364  			defer wg.Done()
   365  			c1.Write(make([]byte, 1024))
   366  		}()
   367  		go func() {
   368  			defer wg.Done()
   369  			c1.SetDeadline(time.Now().Add(10 * time.Millisecond))
   370  		}()
   371  		go func() {
   372  			defer wg.Done()
   373  			c1.SetReadDeadline(aLongTimeAgo)
   374  		}()
   375  		go func() {
   376  			defer wg.Done()
   377  			c1.SetWriteDeadline(aLongTimeAgo)
   378  		}()
   379  		go func() {
   380  			defer wg.Done()
   381  			c1.LocalAddr()
   382  		}()
   383  		go func() {
   384  			defer wg.Done()
   385  			c1.RemoteAddr()
   386  		}()
   387  	}
   388  	wg.Wait() // At worst, the deadline is set 10ms into the future
   389  
   390  	resyncConn(t, c1)
   391  	testRoundtrip(t, c1)
   392  }
   393  
   394  // checkForTimeoutError checks that the error satisfies the Error interface
   395  // and that Timeout returns true.
   396  func checkForTimeoutError(t *testing.T, err error) {
   397  	t.Helper()
   398  	if nerr, ok := err.(net.Error); ok {
   399  		if !nerr.Timeout() {
   400  			if runtime.GOOS == "windows" && runtime.GOARCH == "arm64" && t.Name() == "TestTestConn/TCP/RacyRead" {
   401  				t.Logf("ignoring known failure mode on windows/arm64; see https://go.dev/issue/52893")
   402  			} else {
   403  				t.Errorf("got error: %v, want err.Timeout() = true", nerr)
   404  			}
   405  		}
   406  	} else {
   407  		t.Errorf("got %T: %v, want net.Error", err, err)
   408  	}
   409  }
   410  
   411  // testRoundtrip writes something into c and reads it back.
   412  // It assumes that everything written into c is echoed back to itself.
   413  func testRoundtrip(t *testing.T, c net.Conn) {
   414  	t.Helper()
   415  	if err := c.SetDeadline(neverTimeout); err != nil {
   416  		t.Errorf("roundtrip SetDeadline error: %v", err)
   417  	}
   418  
   419  	const s = "Hello, world!"
   420  	buf := []byte(s)
   421  	if _, err := c.Write(buf); err != nil {
   422  		t.Errorf("roundtrip Write error: %v", err)
   423  	}
   424  	if _, err := io.ReadFull(c, buf); err != nil {
   425  		t.Errorf("roundtrip Read error: %v", err)
   426  	}
   427  	if string(buf) != s {
   428  		t.Errorf("roundtrip data mismatch: got %q, want %q", buf, s)
   429  	}
   430  }
   431  
   432  // resyncConn resynchronizes the connection into a sane state.
   433  // It assumes that everything written into c is echoed back to itself.
   434  // It assumes that 0xff is not currently on the wire or in the read buffer.
   435  func resyncConn(t *testing.T, c net.Conn) {
   436  	t.Helper()
   437  	c.SetDeadline(neverTimeout)
   438  	errCh := make(chan error)
   439  	go func() {
   440  		_, err := c.Write([]byte{0xff})
   441  		errCh <- err
   442  	}()
   443  	buf := make([]byte, 1024)
   444  	for {
   445  		n, err := c.Read(buf)
   446  		if n > 0 && bytes.IndexByte(buf[:n], 0xff) == n-1 {
   447  			break
   448  		}
   449  		if err != nil {
   450  			t.Errorf("unexpected Read error: %v", err)
   451  			break
   452  		}
   453  	}
   454  	if err := <-errCh; err != nil {
   455  		t.Errorf("unexpected Write error: %v", err)
   456  	}
   457  }
   458  
   459  // chunkedCopy copies from r to w in fixed-width chunks to avoid
   460  // causing a Write that exceeds the maximum packet size for packet-based
   461  // connections like "unixpacket".
   462  // We assume that the maximum packet size is at least 1024.
   463  func chunkedCopy(w io.Writer, r io.Reader) error {
   464  	b := make([]byte, 1024)
   465  	_, err := io.CopyBuffer(struct{ io.Writer }{w}, struct{ io.Reader }{r}, b)
   466  	return err
   467  }
   468
View as plain text