Skip to content

Use testing.synctest to ensure all goroutines exit before checking memory leaks in tcp e2e tests #12369

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
uzairnawaz wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Use testing.synctest to ensure all goroutines exit before checking memory leaks in tcp e2e tests #12369

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
622 changes: 337 additions & 285 deletions pkg/tcpip/transport/tcp/test/e2e/dual_stack_test.go
View file
Open in desktop

Large diffs are not rendered by default.

308 changes: 159 additions & 149 deletions pkg/tcpip/transport/tcp/test/e2e/forwarder_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ package forwarder_test
import (
"os"
"testing"
"testing/synctest"
"time"

"gvisor.dev/gvisor/pkg/atomicbitops"
Expand All @@ -31,43 +32,46 @@ import (
)

func TestForwarderSendMSSLessThanMTU(t *testing.T) {
const maxPayload = 100
const mtu = 1200
c := context.New(t, mtu)
defer c.Cleanup()

s := c.Stack()
ch := make(chan tcpip.Error, 1)
f := tcp.NewForwarder(s, 65536, 10, func(r *tcp.ForwarderRequest) {
var err tcpip.Error
c.EP, err = r.CreateEndpoint(&c.WQ)
ch <- err
close(ch)
r.Complete(false)
})
s.SetTransportProtocolHandler(tcp.ProtocolNumber, f.HandlePacket)
synctest.Test(t, func(t *testing.T) {
defer synctest.Wait()
const maxPayload = 100
const mtu = 1200
c := context.New(t, mtu)
defer c.Cleanup()

s := c.Stack()
ch := make(chan tcpip.Error, 1)
f := tcp.NewForwarder(s, 65536, 10, func(r *tcp.ForwarderRequest) {
var err tcpip.Error
c.EP, err = r.CreateEndpoint(&c.WQ)
ch <- err
close(ch)
r.Complete(false)
})
s.SetTransportProtocolHandler(tcp.ProtocolNumber, f.HandlePacket)

// Do 3-way handshake.
c.PassiveConnect(maxPayload, -1, header.TCPSynOptions{MSS: mtu - header.IPv4MinimumSize - header.TCPMinimumSize})
// Do 3-way handshake.
c.PassiveConnect(maxPayload, -1, header.TCPSynOptions{MSS: mtu - header.IPv4MinimumSize - header.TCPMinimumSize})

// Wait for connection to be available.
select {
case err := <-ch:
if err != nil {
t.Fatalf("Error creating endpoint: %s", err)
// Wait for connection to be available.
select {
case err := <-ch:
if err != nil {
t.Fatalf("Error creating endpoint: %s", err)
}
case <-time.After(2 * time.Second):
t.Fatalf("Timed out waiting for connection")
}
case <-time.After(2 * time.Second):
t.Fatalf("Timed out waiting for connection")
}

// Check that data gets properly segmented.
e2e.CheckBrokenUpWrite(t, c, maxPayload)
// Check that data gets properly segmented.
e2e.CheckBrokenUpWrite(t, c, maxPayload)
})
}

func TestForwarderDoesNotRejectECNFlags(t *testing.T) {
testCases := []struct {
name string
flags header.TCPFlags
name string
flags header.TCPFlags
}{
{name: "non-setup ECN SYN w/ ECE", flags: header.TCPFlagEce},
{name: "non-setup ECN SYN w/ CWR", flags: header.TCPFlagCwr},
Expand All @@ -76,146 +80,152 @@ func TestForwarderDoesNotRejectECNFlags(t *testing.T) {

for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
const maxPayload = 100
const mtu = 1200
c := context.New(t, mtu)
defer c.Cleanup()

s := c.Stack()
ch := make(chan tcpip.Error, 1)
f := tcp.NewForwarder(s, 65536, 10, func(r *tcp.ForwarderRequest) {
var err tcpip.Error
c.EP, err = r.CreateEndpoint(&c.WQ)
ch <- err
close(ch)
r.Complete(false)
})
s.SetTransportProtocolHandler(tcp.ProtocolNumber, f.HandlePacket)

// Do 3-way handshake.
c.PassiveConnect(maxPayload, -1, header.TCPSynOptions{MSS: mtu - header.IPv4MinimumSize - header.TCPMinimumSize, Flags: tc.flags})

// Wait for connection to be available.
select {
case err := <-ch:
if err != nil {
t.Fatalf("Error creating endpoint: %s", err)
synctest.Test(t, func(t *testing.T) {
defer synctest.Wait()
const maxPayload = 100
const mtu = 1200
c := context.New(t, mtu)
defer c.Cleanup()

s := c.Stack()
ch := make(chan tcpip.Error, 1)
f := tcp.NewForwarder(s, 65536, 10, func(r *tcp.ForwarderRequest) {
var err tcpip.Error
c.EP, err = r.CreateEndpoint(&c.WQ)
ch <- err
close(ch)
r.Complete(false)
})
s.SetTransportProtocolHandler(tcp.ProtocolNumber, f.HandlePacket)

// Do 3-way handshake.
c.PassiveConnect(maxPayload, -1, header.TCPSynOptions{MSS: mtu - header.IPv4MinimumSize - header.TCPMinimumSize, Flags: tc.flags})

// Wait for connection to be available.
select {
case err := <-ch:
if err != nil {
t.Fatalf("Error creating endpoint: %s", err)
}
case <-time.After(2 * time.Second):
t.Fatalf("Timed out waiting for connection")
}
case <-time.After(2 * time.Second):
t.Fatalf("Timed out waiting for connection")
}
})
})
}
}

func TestForwarderFailedConnect(t *testing.T) {
const mtu = 1200
c := context.New(t, mtu)
defer c.Cleanup()

s := c.Stack()
ch := make(chan tcpip.Error, 1)
f := tcp.NewForwarder(s, 65536, 10, func(r *tcp.ForwarderRequest) {
var err tcpip.Error
c.EP, err = r.CreateEndpoint(&c.WQ)
ch <- err
close(ch)
r.Complete(false)
})
s.SetTransportProtocolHandler(tcp.ProtocolNumber, f.HandlePacket)

// Initiate a connection that will be forwarded by the Forwarder.
// Send a SYN request.
iss := seqnum.Value(context.TestInitialSequenceNumber)
c.SendPacket(nil, &context.Headers{
SrcPort: context.TestPort,
DstPort: context.StackPort,
Flags: header.TCPFlagSyn,
SeqNum: iss,
RcvWnd: 30000,
})
synctest.Test(t, func(t *testing.T) {
defer synctest.Wait()
const mtu = 1200
c := context.New(t, mtu)
defer c.Cleanup()

s := c.Stack()
ch := make(chan tcpip.Error, 1)
f := tcp.NewForwarder(s, 65536, 10, func(r *tcp.ForwarderRequest) {
var err tcpip.Error
c.EP, err = r.CreateEndpoint(&c.WQ)
ch <- err
close(ch)
r.Complete(false)
})
s.SetTransportProtocolHandler(tcp.ProtocolNumber, f.HandlePacket)

// Receive the SYN-ACK reply. Make sure MSS and other expected options
// are present.
v := c.GetPacket()
defer v.Release()
tcp := header.TCP(header.IPv4(v.AsSlice()).Payload())
c.IRS = seqnum.Value(tcp.SequenceNumber())

tcpCheckers := []checker.TransportChecker{
checker.SrcPort(context.StackPort),
checker.DstPort(context.TestPort),
checker.TCPFlags(header.TCPFlagAck | header.TCPFlagSyn),
checker.TCPAckNum(uint32(iss) + 1),
}
checker.IPv4(t, v, checker.TCP(tcpCheckers...))

// Now send an active RST to abort the handshake.
c.SendPacket(nil, &context.Headers{
SrcPort: context.TestPort,
DstPort: context.StackPort,
Flags: header.TCPFlagRst,
SeqNum: iss + 1,
RcvWnd: 0,
})
// Initiate a connection that will be forwarded by the Forwarder.
// Send a SYN request.
iss := seqnum.Value(context.TestInitialSequenceNumber)
c.SendPacket(nil, &context.Headers{
SrcPort: context.TestPort,
DstPort: context.StackPort,
Flags: header.TCPFlagSyn,
SeqNum: iss,
RcvWnd: 30000,
})

// Wait for connect to fail.
select {
case err := <-ch:
if err == nil {
t.Fatalf("endpoint creation should have failed")
// Receive the SYN-ACK reply. Make sure MSS and other expected options
// are present.
v := c.GetPacket()
defer v.Release()
tcp := header.TCP(header.IPv4(v.AsSlice()).Payload())
c.IRS = seqnum.Value(tcp.SequenceNumber())

tcpCheckers := []checker.TransportChecker{
checker.SrcPort(context.StackPort),
checker.DstPort(context.TestPort),
checker.TCPFlags(header.TCPFlagAck | header.TCPFlagSyn),
checker.TCPAckNum(uint32(iss) + 1),
}
case <-time.After(2 * time.Second):
t.Fatalf("Timed out waiting for connection to fail")
}
}
checker.IPv4(t, v, checker.TCP(tcpCheckers...))

func TestForwarderDroppedStats(t *testing.T) {
const maxPayload = 100
const mtu = 1200
c := context.New(t, mtu)
defer c.Cleanup()

const maxInFlight = 2
iters := atomicbitops.FromInt64(maxInFlight)
s := c.Stack()
checkedStats := make(chan struct{})
done := make(chan struct{})
f := tcp.NewForwarder(s, 65536, maxInFlight, func(r *tcp.ForwarderRequest) {
<-checkedStats
// Complete all requests without doing anything
r.Complete(false)
if iter := iters.Add(-1); iter == 0 {
close(done)
// Now send an active RST to abort the handshake.
c.SendPacket(nil, &context.Headers{
SrcPort: context.TestPort,
DstPort: context.StackPort,
Flags: header.TCPFlagRst,
SeqNum: iss + 1,
RcvWnd: 0,
})

// Wait for connect to fail.
select {
case err := <-ch:
if err == nil {
t.Fatalf("endpoint creation should have failed")
}
case <-time.After(2 * time.Second):
t.Fatalf("Timed out waiting for connection to fail")
}
})
s.SetTransportProtocolHandler(tcp.ProtocolNumber, f.HandlePacket)
}

for i := 0; i < maxInFlight+1; i++ {
iss := seqnum.Value(context.TestInitialSequenceNumber + i)
c.SendPacket(nil, &context.Headers{
SrcPort: uint16(context.TestPort + i),
DstPort: context.StackPort,
Flags: header.TCPFlagSyn,
SeqNum: iss,
RcvWnd: 30000,
func TestForwarderDroppedStats(t *testing.T) {
synctest.Test(t, func(t *testing.T) {
defer synctest.Wait()
const maxPayload = 100
const mtu = 1200
c := context.New(t, mtu)
defer c.Cleanup()

const maxInFlight = 2
iters := atomicbitops.FromInt64(maxInFlight)
s := c.Stack()
checkedStats := make(chan struct{})
done := make(chan struct{})
f := tcp.NewForwarder(s, 65536, maxInFlight, func(r *tcp.ForwarderRequest) {
<-checkedStats
// Complete all requests without doing anything
r.Complete(false)
if iter := iters.Add(-1); iter == 0 {
close(done)
}
})
}
s.SetTransportProtocolHandler(tcp.ProtocolNumber, f.HandlePacket)

for i := 0; i < maxInFlight+1; i++ {
iss := seqnum.Value(context.TestInitialSequenceNumber + i)
c.SendPacket(nil, &context.Headers{
SrcPort: uint16(context.TestPort + i),
DstPort: context.StackPort,
Flags: header.TCPFlagSyn,
SeqNum: iss,
RcvWnd: 30000,
})
}

// Verify that we got one ignored packet.
if curr := s.Stats().TCP.ForwardMaxInFlightDrop.Value(); curr != 1 {
t.Errorf("Expected one dropped connection, but got %d", curr)
}
close(checkedStats)
<-done
// Verify that we got one ignored packet.
if curr := s.Stats().TCP.ForwardMaxInFlightDrop.Value(); curr != 1 {
t.Errorf("Expected one dropped connection, but got %d", curr)
}
close(checkedStats)
<-done
})
}

func TestMain(m *testing.M) {
refs.SetLeakMode(refs.LeaksPanic)
code := m.Run()
// Allow TCP async work to complete to avoid false reports of leaks.
// TODO(gvisor.dev/issue/5940): Use fake clock in tests.
time.Sleep(1 * time.Second)
refs.DoLeakCheck()
os.Exit(code)
}
Loading