diff --git a/sctp_close_fuzz_test.go b/sctp_close_fuzz_test.go new file mode 100644 index 0000000..de2fda1 --- /dev/null +++ b/sctp_close_fuzz_test.go @@ -0,0 +1,275 @@ +//go:build linux && !386 +// +build linux,!386 + +// Copyright 2019 Wataru Ishida. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or +// implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package sctp + +import ( + "errors" + "sync" + "syscall" + "testing" + "time" +) + +// FuzzCloseWithTimeout varies the grace period, including the negative and +// zero values that select the immediate-abort path, and the sub-microsecond +// values where the timeval conversion could round down to "no timeout" and +// block forever. +// +// The invariant: CloseWithTimeout always returns, always releases the +// descriptor, and a second call always reports EBADF. +func FuzzCloseWithTimeout(f *testing.F) { + f.Add(int64(0)) + f.Add(int64(-1)) + f.Add(int64(1)) // 1ns: rounds up to 1us + f.Add(int64(999)) // sub-microsecond + f.Add(int64(time.Millisecond)) //nolint:gosec + f.Add(int64(3 * time.Second)) //nolint:gosec + f.Add(int64(time.Hour)) //nolint:gosec + + f.Fuzz(func(t *testing.T, ns int64) { + // Keep the wait bounded: this exercises the conversion arithmetic and + // the branch selection, not real multi-second waits. + if ns > int64(2*time.Second) { + ns = int64(2 * time.Second) + } + timeout := time.Duration(ns) + + // Rapid dial churn can hit EISCONN from SCTPConnect (see + // TestDialUnderChurnReportsEISCONN); that is a dial-side limitation, + // not a close-path failure, so skip rather than fail this iteration. + client, server, ok := eorPairMaybe(t) + if !ok { + t.Skip("association could not be established this iteration") + } + defer func() { _ = server.Close() }() + + fd := client.fd() + + done := make(chan error, 1) + go func() { done <- client.CloseWithTimeout(timeout) }() + + select { + case err := <-done: + if err != nil && !errors.Is(err, syscall.EBADF) { + t.Fatalf("timeout=%v: CloseWithTimeout = %v", timeout, err) + } + case <-time.After(30 * time.Second): + t.Fatalf("timeout=%v: CloseWithTimeout never returned", timeout) + } + + if fd >= 0 && fdIsOpen(fd) { + t.Fatalf("timeout=%v: descriptor %d still open after close", timeout, fd) + } + if err := client.Close(); !errors.Is(err, syscall.EBADF) { + t.Fatalf("timeout=%v: second Close = %v, want EBADF", timeout, err) + } + }) +} + +// TestCloseTimeoutZeroIsImmediate pins the documented equivalence between a +// non-positive timeout and Abort. +func TestCloseTimeoutZeroIsImmediate(t *testing.T) { + for _, timeout := range []time.Duration{0, -time.Second} { + client, server := eorPairNoCleanup(t) + + start := time.Now() + if err := client.CloseWithTimeout(timeout); err != nil { + t.Fatalf("timeout=%v: %v", timeout, err) + } + if d := time.Since(start); d > time.Second { + t.Errorf("timeout=%v: took %v, expected an immediate abort", timeout, d) + } + _ = server.Close() + } +} + +// TestCloseSubMicrosecondTimeout guards the timeval conversion: a timeout +// under one microsecond must not truncate to zero, which the kernel reads as +// "block forever". +func TestCloseSubMicrosecondTimeout(t *testing.T) { + client, server := eorPairNoCleanup(t) + defer func() { _ = server.Close() }() + + done := make(chan error, 1) + go func() { done <- client.CloseWithTimeout(1 * time.Nanosecond) }() + + select { + case err := <-done: + if err != nil { + t.Fatalf("CloseWithTimeout(1ns) = %v", err) + } + case <-time.After(15 * time.Second): + t.Fatal("CloseWithTimeout(1ns) blocked; the timeval likely truncated to zero") + } +} + +// TestCloseChurnUnderLoad is the industry case: a signalling node cycling many +// associations must not accumulate descriptors or wedge on teardown. +func TestCloseChurnUnderLoad(t *testing.T) { + addr, _ := ResolveSCTPAddr("sctp", "127.0.0.1:0") + ln, err := ListenSCTP("sctp", addr) + if err != nil { + t.Fatalf("listen: %v", err) + } + defer func() { _ = ln.Close() }() + + go func() { + for { + c, err := ln.AcceptSCTP() + if err != nil { + return + } + go func(c *SCTPConn) { + buf := make([]byte, 512) + for { + if _, _, err := c.SCTPRead(buf); err != nil { + // The client has already gone; abort rather than + // run a graceful shutdown that has no peer to + // answer it and would serialise the whole test. + _ = c.Abort() + return + } + } + }(c) + } + }() + + before := countOpenFds(t) + + const workers, perWorker = 8, 15 + var wg sync.WaitGroup + wg.Add(workers) + errs := make(chan error, workers*perWorker) + for w := 0; w < workers; w++ { + go func(w int) { + defer wg.Done() + for i := 0; i < perWorker; i++ { + conn, err := DialSCTP("sctp", nil, ln.Addr().(*SCTPAddr)) + if err != nil { + errs <- err + continue + } + // Send something so the association is real before teardown. + _, _ = conn.SCTPWrite([]byte("churn"), nil) + + // Alternate graceful and immediate teardown. The graceful + // path uses a short grace period deliberately: the server + // aborts its side as soon as its read fails, so a SHUTDOWN + // may find no peer to answer it, and the default period + // would make this loop wait out the timeout each time. + if (w+i)%2 == 0 { + err = conn.CloseWithTimeout(200 * time.Millisecond) + } else { + err = conn.Abort() + } + if err != nil { + errs <- err + } + } + }(w) + } + wg.Wait() + close(errs) + + var failures int + for err := range errs { + failures++ + if failures <= 3 { + t.Logf("teardown error: %v", err) + } + } + if failures > 0 { + t.Errorf("%d of %d cycles failed", failures, workers*perWorker) + } + + // Wait for the accept goroutines to reap their side. The descriptor count + // is process-global, so other tests running concurrently in the same + // binary can inflate it transiently; poll until it settles rather than + // sampling once and treating a neighbour's socket as a leak here. + const margin = 20 + var after int + for attempt := 0; attempt < 40; attempt++ { + time.Sleep(100 * time.Millisecond) + after = countOpenFds(t) + if after <= before+margin { + break + } + } + t.Logf("descriptors before=%d after=%d over %d cycles", + before, after, workers*perWorker) + if after > before+margin { + t.Errorf("descriptors grew from %d to %d over %d cycles and did not settle", + before, after, workers*perWorker) + } +} + +// TestCloseRacesWithReadAndWrite drives close against concurrent I/O, the +// shape that surfaces use-after-close on a recycled descriptor. +func TestCloseRacesWithReadAndWrite(t *testing.T) { + for round := 0; round < 20; round++ { + client, server := eorPairNoCleanup(t) + + var wg sync.WaitGroup + wg.Add(3) + + go func() { + defer wg.Done() + buf := make([]byte, 256) + for i := 0; i < 50; i++ { + if _, _, err := client.SCTPRead(buf); err != nil { + return + } + } + }() + go func() { + defer wg.Done() + payload := []byte("racing") + for i := 0; i < 50; i++ { + if _, err := client.SCTPWrite(payload, nil); err != nil { + return + } + } + }() + go func() { + defer wg.Done() + time.Sleep(time.Duration(round%5) * time.Millisecond) + _ = client.Close() + }() + + wg.Wait() + _ = server.Close() + } +} + +// TestAbortDoesNotWait checks Abort returns promptly even when the peer is +// gone, since it performs no handshake. +func TestAbortDoesNotWait(t *testing.T) { + client, server := eorPairNoCleanup(t) + if err := server.Abort(); err != nil { + t.Fatalf("server abort: %v", err) + } + + start := time.Now() + if err := client.Abort(); err != nil { + t.Fatalf("client abort: %v", err) + } + if d := time.Since(start); d > time.Second { + t.Errorf("Abort took %v against a dead peer; it must not wait", d) + } +} diff --git a/sctp_close_test.go b/sctp_close_test.go new file mode 100644 index 0000000..6d1fa6b --- /dev/null +++ b/sctp_close_test.go @@ -0,0 +1,624 @@ +//go:build linux && !386 +// +build linux,!386 + +// Copyright 2019 Wataru Ishida. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or +// implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package sctp + +import ( + "errors" + "fmt" + "net" + "os" + "os/exec" + "sync" + "syscall" + "testing" + "time" +) + +// testingTB is the subset of testing.TB these helpers need, so the same setup +// serves both tests and fuzz targets. +type testingTB interface { + Helper() + Fatalf(format string, args ...interface{}) + Cleanup(func()) +} + +// fdIsOpen reports whether fd refers to an open descriptor in this process. +func fdIsOpen(fd int) bool { + _, _, errno := syscall.Syscall(syscall.SYS_FCNTL, + uintptr(fd), uintptr(syscall.F_GETFD), 0) + return errno == 0 +} + +// runChild re-executes the test binary for a single test with env set, so a +// test may disturb process-wide state (such as closing stdin) in isolation. +// It returns the child's combined output. +func runChild(exe, testName, env string) (string, error) { + cmd := exec.Command(exe, "-test.run", "^"+testName+"$", "-test.v") + cmd.Env = append(os.Environ(), env) + out, err := cmd.CombinedOutput() + return string(out), err +} + +// closePair brings up a loopback association and closes both ends on test +// cleanup. Tests that tear the connections down themselves should use +// eorPairNoCleanup instead, to avoid a double close during cleanup. +func closePair(t testingTB) (client, server *SCTPConn) { + t.Helper() + client, server = eorPairNoCleanup(t) + t.Cleanup(func() { _ = client.Close() }) + t.Cleanup(func() { _ = server.Close() }) + return client, server +} + +// eorPairNoCleanup is closePair without the t.Cleanup close hooks, for tests +// that close the connections themselves and would otherwise trip a double +// close during cleanup. +func eorPairNoCleanup(t testingTB) (client, server *SCTPConn) { + t.Helper() + client, server, ok := eorPairMaybe(t) + if !ok { + t.Fatalf("could not establish association") + } + return client, server +} + +// eorPairMaybe is eorPairNoCleanup that reports failure instead of aborting +// the test, for callers that run many iterations and can tolerate the +// occasional dial failure under churn. +func eorPairMaybe(t testingTB) (client, server *SCTPConn, ok bool) { + t.Helper() + + addr, err := ResolveSCTPAddr("sctp", "127.0.0.1:0") + if err != nil { + t.Fatalf("resolve: %v", err) + } + ln, err := ListenSCTP("sctp", addr) + if err != nil { + t.Fatalf("listen: %v", err) + } + t.Cleanup(func() { _ = ln.Close() }) + + type accepted struct { + conn *SCTPConn + err error + } + ch := make(chan accepted, 1) + go func() { + c, err := ln.AcceptSCTP() + ch <- accepted{c, err} + }() + + client, err = DialSCTP("sctp", nil, ln.Addr().(*SCTPAddr)) + if err != nil { + return nil, nil, false + } + + a := <-ch + if a.err != nil { + _ = client.Close() + return nil, nil, false + } + return client, a.conn, true +} + +// TestDialUnderChurnReportsEISCONN documents a dial-side limitation found +// while fuzzing the close paths. +// +// Under rapid dial/teardown cycles, SCTPConnect can return EISCONN +// ("transport endpoint is already connected") on a freshly created socket. +// The socket is new, so the error reflects kernel association state that has +// not finished tearing down rather than anything the caller did. DialSCTP +// surfaces it unchanged, so callers doing rapid reconnects must retry. +// +// It was originally observed when accepted connections were left open, which +// suggests lingering peer-side associations are what provoke it; aborting +// each accepted connection promptly, as here, usually avoids it. The test +// therefore records what it observes rather than asserting a rate, so it +// documents the behaviour without becoming flaky either way. +// +// This is separate from the close path: connections being closed are released +// correctly, as TestCloseChurnUnderLoad shows. +func TestDialUnderChurnReportsEISCONN(t *testing.T) { + addr, _ := ResolveSCTPAddr("sctp", "127.0.0.1:0") + ln, err := ListenSCTP("sctp", addr) + if err != nil { + t.Fatalf("listen: %v", err) + } + defer func() { _ = ln.Close() }() + + // Abort each accepted connection rather than letting them accumulate: + // leaving hundreds of server-side associations open would make the + // listener's own teardown dominate the test's runtime. + go func() { + for { + c, err := ln.AcceptSCTP() + if err != nil { + return + } + _ = c.Abort() + } + }() + + const cycles = 300 + failures := map[string]int{} + for i := 0; i < cycles; i++ { + conn, err := DialSCTP("sctp", nil, ln.Addr().(*SCTPAddr)) + if err != nil { + failures[err.Error()]++ + continue + } + if err := conn.Abort(); err != nil { + t.Fatalf("abort %d: %v", i, err) + } + } + + if len(failures) == 0 { + t.Logf("%d rapid dial cycles, no failures", cycles) + return + } + for msg, n := range failures { + t.Logf("%d/%d dials failed with %q", n, cycles, msg) + } +} + +// TestCloseReleasesFdZero covers a descriptor leak that only appears when a +// connection lands on fd 0. +// +// Close and Abort guarded the descriptor with "fd > 0". Zero is a perfectly +// valid descriptor, so a connection on fd 0 had its _fd swapped to -1 and was +// then never closed: the socket leaked and Close reported EBADF as though +// nothing had been open. Daemons that close stdin routinely hand out fd 0. +// +// The test runs in a subprocess because it must close stdin to force the +// allocation, which would otherwise disturb the rest of the suite. +func TestCloseReleasesFdZero(t *testing.T) { + if os.Getenv("SCTP_FD_ZERO_CHILD") == "1" { + fdZeroChild() + return + } + + exe, err := os.Executable() + if err != nil { + t.Skipf("cannot locate test binary: %v", err) + } + out, err := runChild(exe, "TestCloseReleasesFdZero", "SCTP_FD_ZERO_CHILD=1") + if err != nil { + t.Fatalf("child failed: %v\n%s", err, out) + } + t.Logf("child reported:\n%s", out) +} + +// fdZeroChild closes stdin, opens an association that therefore lands on fd 0, +// and checks Close both reports success and actually releases the descriptor. +func fdZeroChild() { + report := func(format string, args ...interface{}) { + fmt.Fprintf(os.Stderr, format+"\n", args...) + } + + addr, err := ResolveSCTPAddr("sctp", "127.0.0.1:0") + if err != nil { + report("FAIL resolve: %v", err) + os.Exit(1) + } + ln, err := ListenSCTP("sctp", addr) + if err != nil { + report("FAIL listen: %v", err) + os.Exit(1) + } + go func() { + for { + c, err := ln.AcceptSCTP() + if err != nil { + return + } + _ = c + } + }() + + // Free fd 0 so the next socket lands there. + if err := syscall.Close(0); err != nil { + report("SKIP cannot close stdin: %v", err) + os.Exit(0) + } + + conn, err := DialSCTP("sctp", nil, ln.Addr().(*SCTPAddr)) + if err != nil { + report("SKIP dial: %v", err) + os.Exit(0) + } + fd := conn.fd() + if fd != 0 { + report("SKIP association landed on fd %d, not 0; nothing to prove", fd) + os.Exit(0) + } + + if err := conn.Close(); err != nil { + report("FAIL Close on fd 0 returned %v, want nil", err) + os.Exit(1) + } + if fdIsOpen(0) { + report("FAIL fd 0 is still open after Close; the descriptor leaked") + os.Exit(1) + } + report("ok: connection on fd 0 closed cleanly and the descriptor was released") + os.Exit(0) +} + +// TestAbortReleasesFdZero is the same guarantee for Abort. +func TestAbortReleasesFdZero(t *testing.T) { + if os.Getenv("SCTP_FD_ZERO_ABORT_CHILD") == "1" { + abortFdZeroChild() + return + } + + exe, err := os.Executable() + if err != nil { + t.Skipf("cannot locate test binary: %v", err) + } + out, err := runChild(exe, "TestAbortReleasesFdZero", "SCTP_FD_ZERO_ABORT_CHILD=1") + if err != nil { + t.Fatalf("child failed: %v\n%s", err, out) + } + t.Logf("child reported:\n%s", out) +} + +func abortFdZeroChild() { + report := func(format string, args ...interface{}) { + fmt.Fprintf(os.Stderr, format+"\n", args...) + } + + addr, _ := ResolveSCTPAddr("sctp", "127.0.0.1:0") + ln, err := ListenSCTP("sctp", addr) + if err != nil { + report("FAIL listen: %v", err) + os.Exit(1) + } + go func() { + for { + c, err := ln.AcceptSCTP() + if err != nil { + return + } + _ = c + } + }() + + if err := syscall.Close(0); err != nil { + report("SKIP cannot close stdin: %v", err) + os.Exit(0) + } + conn, err := DialSCTP("sctp", nil, ln.Addr().(*SCTPAddr)) + if err != nil { + report("SKIP dial: %v", err) + os.Exit(0) + } + if fd := conn.fd(); fd != 0 { + report("SKIP association landed on fd %d, not 0", fd) + os.Exit(0) + } + if err := conn.Abort(); err != nil { + report("FAIL Abort on fd 0 returned %v, want nil", err) + os.Exit(1) + } + if fdIsOpen(0) { + report("FAIL fd 0 is still open after Abort; the descriptor leaked") + os.Exit(1) + } + report("ok: Abort on fd 0 released the descriptor") + os.Exit(0) +} + +// TestCloseDoesNotLeakDescriptors is the general form: repeated dial/close +// cycles must not accumulate descriptors, whatever fd they land on. +func TestCloseDoesNotLeakDescriptors(t *testing.T) { + addr, _ := ResolveSCTPAddr("sctp", "127.0.0.1:0") + ln, err := ListenSCTP("sctp", addr) + if err != nil { + t.Fatalf("listen: %v", err) + } + defer func() { _ = ln.Close() }() + + go func() { + for { + c, err := ln.AcceptSCTP() + if err != nil { + return + } + go func(c *SCTPConn) { + buf := make([]byte, 256) + for { + if _, _, err := c.SCTPRead(buf); err != nil { + // See TestCloseChurnUnderLoad: the peer is gone, so + // a graceful shutdown would only add latency. + _ = c.Abort() + return + } + } + }(c) + } + }() + + before := countOpenFds(t) + for i := 0; i < 50; i++ { + conn, err := DialSCTP("sctp", nil, ln.Addr().(*SCTPAddr)) + if err != nil { + // Rapid reconnects can hit EISCONN from SCTPConnect; see + // TestDialUnderChurnReportsEISCONN. That is a dial-side + // limitation, not a descriptor leak, so skip the cycle. + continue + } + if err := conn.Close(); err != nil { + t.Fatalf("close %d: %v", i, err) + } + } + // The count is process-global, so poll until it settles rather than + // treating a concurrently running test's socket as a leak here. + const margin = 10 + var after int + for attempt := 0; attempt < 40; attempt++ { + after = countOpenFds(t) + if after <= before+margin { + break + } + time.Sleep(100 * time.Millisecond) + } + t.Logf("open descriptors before=%d after=%d", before, after) + if after > before+margin { + t.Errorf("descriptor count grew from %d to %d over 50 dial/close cycles", + before, after) + } +} + +func countOpenFds(t *testing.T) int { + t.Helper() + entries, err := os.ReadDir("/proc/self/fd") + if err != nil { + t.Skipf("cannot read /proc/self/fd: %v", err) + } + return len(entries) +} + +// TestCloseReleasesPortForRebind is the symptom the close rework exists to +// fix: after Close returns, the local address must be immediately reusable +// rather than yielding EADDRINUSE. +func TestCloseReleasesPortForRebind(t *testing.T) { + // Bind a listener on a fixed port, close it, and rebind the same port. + for attempt := 0; attempt < 5; attempt++ { + addr, err := ResolveSCTPAddr("sctp", "127.0.0.1:0") + if err != nil { + t.Fatalf("resolve: %v", err) + } + ln, err := ListenSCTP("sctp", addr) + if err != nil { + t.Fatalf("listen: %v", err) + } + bound := ln.Addr().(*SCTPAddr) + + // Hold the accepted connection so it can be closed deterministically: + // an association still open on the listener's address keeps that + // address bound, independently of the listener itself. + accepted := make(chan *SCTPConn, 1) + go func() { + c, err := ln.AcceptSCTP() + if err != nil { + accepted <- nil + return + } + accepted <- c + }() + + // Retry past the EISCONN that rapid reconnects can produce; see + // TestDialUnderChurnReportsEISCONN. + var conn *SCTPConn + for attempt := 0; attempt < 20; attempt++ { + conn, err = DialSCTP("sctp", nil, bound) + if err == nil { + break + } + time.Sleep(10 * time.Millisecond) + } + if err != nil { + _ = ln.Close() + t.Fatalf("dial: %v", err) + } + srv := <-accepted + if err := conn.Close(); err != nil { + t.Fatalf("conn close: %v", err) + } + if srv != nil { + if err := srv.Close(); err != nil { + t.Fatalf("accepted conn close: %v", err) + } + } + if err := ln.Close(); err != nil { + t.Fatalf("listener close: %v", err) + } + + // The exact address must now be bindable again. + ln2, err := ListenSCTP("sctp", bound) + if err != nil { + t.Fatalf("attempt %d: rebinding %s after close failed: %v", + attempt, bound, err) + } + _ = ln2.Close() + } +} + +// TestCloseWithUnreachablePeerReturnsWithinTimeout pins the fallback path. +// When the peer never acknowledges the shutdown, Close must still return +// promptly rather than blocking indefinitely. +func TestCloseWithUnreachablePeerReturnsWithinTimeout(t *testing.T) { + client, server := closePair(t) + + // Make the peer unresponsive by abandoning it without a graceful close: + // drop the server's descriptor so nothing answers the SHUTDOWN. + if err := server.Abort(); err != nil { + t.Fatalf("abort server: %v", err) + } + + done := make(chan error, 1) + start := time.Now() + go func() { done <- client.Close() }() + + select { + case err := <-done: + elapsed := time.Since(start) + t.Logf("Close returned after %v with err=%v", elapsed, err) + if elapsed > closeTimeout+2*time.Second { + t.Errorf("Close took %v, well past the %v timeout", elapsed, closeTimeout) + } + case <-time.After(closeTimeout + 5*time.Second): + t.Fatalf("Close did not return within %v", closeTimeout+5*time.Second) + } +} + +// TestDoubleCloseReturnsEBADF checks the second close is reported rather than +// silently succeeding or closing a descriptor the process has since reused. +func TestDoubleCloseReturnsEBADF(t *testing.T) { + client, _ := closePair(t) + + if err := client.Close(); err != nil { + t.Fatalf("first Close: %v", err) + } + if err := client.Close(); !errors.Is(err, syscall.EBADF) { + t.Errorf("second Close = %v, want EBADF", err) + } + if err := client.Abort(); !errors.Is(err, syscall.EBADF) { + t.Errorf("Abort after Close = %v, want EBADF", err) + } +} + +// TestCloseOnNilConn guards the nil receiver path. +func TestCloseOnNilConn(t *testing.T) { + var c *SCTPConn + if err := c.Close(); !errors.Is(err, syscall.EBADF) { + t.Errorf("nil Close = %v, want EBADF", err) + } + if err := c.Abort(); !errors.Is(err, syscall.EBADF) { + t.Errorf("nil Abort = %v, want EBADF", err) + } +} + +// TestConcurrentCloseAndAbort checks only one caller wins the descriptor and +// the rest get EBADF, with no double close of a reused fd. +func TestConcurrentCloseAndAbort(t *testing.T) { + for round := 0; round < 25; round++ { + client, _ := eorPairNoCleanup(t) + + const racers = 8 + var ( + wg sync.WaitGroup + mu sync.Mutex + okCount int + ) + wg.Add(racers) + for i := 0; i < racers; i++ { + go func(i int) { + defer wg.Done() + var err error + if i%2 == 0 { + err = client.Close() + } else { + err = client.Abort() + } + if err == nil { + mu.Lock() + okCount++ + mu.Unlock() + } + }(i) + } + wg.Wait() + + if okCount != 1 { + t.Fatalf("round %d: %d callers reported success, want exactly 1", + round, okCount) + } + } +} + +// TestCloseDuringBlockedRead covers a shutdown racing an in-flight read. +func TestCloseDuringBlockedRead(t *testing.T) { + client, server := closePair(t) + _ = client + + readDone := make(chan error, 1) + go func() { + buf := make([]byte, 1024) + _, _, err := server.SCTPRead(buf) + readDone <- err + }() + + // Give the read time to block in recvmsg. + time.Sleep(200 * time.Millisecond) + _ = server.Close() + + select { + case err := <-readDone: + t.Logf("blocked read returned %v after Close", err) + case <-time.After(10 * time.Second): + t.Fatal("read did not return after Close; the descriptor never unblocked") + } +} + +// TestCloseDuringWrite covers close racing a writer. +func TestCloseDuringWrite(t *testing.T) { + client, server := closePair(t) + + // Drain the peer. Without a reader the send queue fills and a blocking + // write never returns, so the test would be measuring send-buffer + // pressure rather than the close. + go func() { + buf := make([]byte, 4096) + for { + if _, _, err := server.SCTPRead(buf); err != nil { + return + } + } + }() + + stop := make(chan struct{}) + done := make(chan struct{}) + go func() { + defer close(done) + payload := make([]byte, 1024) + for { + select { + case <-stop: + return + default: + } + if _, err := client.SCTPWrite(payload, nil); err != nil { + return + } + } + }() + + time.Sleep(100 * time.Millisecond) + _ = client.Close() + close(stop) + + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("writer did not observe the close") + } +} + +var _ net.Conn = (*SCTPConn)(nil) diff --git a/sctp_linux.go b/sctp_linux.go index 778d3cf..66a680d 100644 --- a/sctp_linux.go +++ b/sctp_linux.go @@ -25,6 +25,7 @@ import ( "runtime" "sync/atomic" "syscall" + "time" "unsafe" ) @@ -152,19 +153,122 @@ func (c *SCTPConn) SCTPRead(b []byte) (int, *SndRcvInfo, error) { } } +// Close closes the SCTP connection gracefully with a timeout fallback. +// +// It initiates a graceful shutdown by sending a SHUTDOWN chunk to the peer +// and waits up to 3 seconds for the peer to acknowledge (SHUTDOWN-ACK). +// If the peer responds, the connection closes gracefully and resources are +// released immediately. If the peer does not respond within the timeout +// (e.g., network failure or unreachable peer), an ABORT chunk is sent to +// forcefully terminate the association and release resources. +// +// This ensures that Close always returns promptly and releases resources, +// avoiding the EADDRINUSE "Address already in use" error that can occur when the kernel +// still occupies the resource. +// +// For immediate termination without waiting, use Abort() instead. func (c *SCTPConn) Close() error { - if c != nil { - fd := atomic.SwapInt32(&c._fd, -1) - if fd > 0 { - info := &SndRcvInfo{ - Flags: SCTP_EOF, - } - c.SCTPWrite(nil, info) - syscall.Shutdown(int(fd), syscall.SHUT_RDWR) - return syscall.Close(int(fd)) + return c.CloseWithTimeout(closeTimeout) +} + +// closeTimeout is how long Close waits for the peer to acknowledge a +// shutdown before falling back to an ABORT. It is a compromise: long enough +// that a peer on a congested link can still complete the handshake, short +// enough that a server tearing down many associations is not held up by a +// peer that will never answer. Use CloseWithTimeout to choose another value. +const closeTimeout = 3 * time.Second + +// establishTimeout bounds the same wait on the error paths of dial and +// listen. Those sockets have no established association to shut down +// gracefully, so the wait exists only to let the kernel release the address. +const establishTimeout = 1 * time.Second + +// CloseWithTimeout is Close with a caller-chosen grace period. +// +// A zero or negative timeout skips the wait entirely and terminates the +// association immediately, which is equivalent to Abort. +func (c *SCTPConn) CloseWithTimeout(timeout time.Duration) error { + if c == nil { + return syscall.EBADF + } + fd := atomic.SwapInt32(&c._fd, -1) + // Zero is a valid descriptor: a process that has closed stdin can be + // handed fd 0 for a socket. Guarding with "fd > 0" leaked it. + if fd < 0 { + return syscall.EBADF + } + return closeSctpSocket(int(fd), timeout) +} + +func closeSctpSocket(fd int, timeout time.Duration) error { + if timeout <= 0 { + return abortSctpSocket(fd) + } + + // Send SHUTDOWN to initiate graceful shutdown. A failure here means no + // graceful shutdown is possible, so skip the wait and abort instead of + // blocking for a SHUTDOWN-ACK that cannot arrive. + if err := syscall.Shutdown(fd, syscall.SHUT_RDWR); err != nil { + return abortSctpSocket(fd) + } + + // Wait for graceful shutdown to complete. + // If peer responds, Read returns immediately with ENOTCONN. + // If peer is unreachable, Read times out after the configured duration. + // + // The timeout is what bounds this read, so if it cannot be programmed the + // read would block indefinitely. Abort rather than risk that. + tv := syscall.NsecToTimeval(timeout.Nanoseconds()) + if err := syscall.SetsockoptTimeval(fd, syscall.SOL_SOCKET, syscall.SO_RCVTIMEO, + &tv); err != nil { + return abortSctpSocket(fd) + } + var buf [1]byte + // The result is deliberately ignored: this read exists to wait for the + // peer's SHUTDOWN-ACK, and any outcome (data, ENOTCONN, timeout) means + // the wait is over. + _, _ = syscall.Read(fd, buf[:]) + + // Set linger=0 so close() sends ABORT if handshake didn't complete, + // or just releases resources if it did. + if err := syscall.SetsockoptLinger(fd, syscall.SOL_SOCKET, syscall.SO_LINGER, + &syscall.Linger{Onoff: 1, Linger: 0}); err != nil { + // Without linger the close may leave the association lingering, but + // the descriptor must still be released. + if cerr := syscall.Close(fd); cerr != nil { + return cerr } + return err + } + return syscall.Close(fd) +} + +// abortSctpSocket terminates the association immediately and releases fd. +func abortSctpSocket(fd int) error { + // Setting SO_LINGER with l_onoff=1 and l_linger=0 causes the kernel to + // send an ABORT chunk instead of SHUTDOWN when closing. + lerr := syscall.SetsockoptLinger(fd, syscall.SOL_SOCKET, syscall.SO_LINGER, + &syscall.Linger{Onoff: 1, Linger: 0}) + if err := syscall.Close(fd); err != nil { + return err } - return syscall.EBADF + return lerr +} + +// Abort terminates the SCTP association immediately by sending an ABORT chunk. +// Unlike Close(), this does not perform a graceful shutdown handshake. +// Use this when you need immediate resource release without waiting for +// the peer to acknowledge the shutdown (e.g., when the peer is unreachable). +func (c *SCTPConn) Abort() error { + if c == nil { + return syscall.EBADF + } + fd := atomic.SwapInt32(&c._fd, -1) + // See CloseWithTimeout: fd 0 is valid and must not be skipped. + if fd < 0 { + return syscall.EBADF + } + return abortSctpSocket(int(fd)) } func (c *SCTPConn) SetWriteBuffer(bytes int) error { @@ -208,7 +312,7 @@ func listenSCTPExtConfig(network string, laddr *SCTPAddr, options InitMsg, contr // close socket on error defer func() { if err != nil { - syscall.Close(sock) + _ = closeSctpSocket(sock, establishTimeout) } }() if err = setDefaultSockopts(sock, af, ipv6only); err != nil { @@ -325,7 +429,7 @@ func dialSCTPExtConfig(network string, laddr, raddr *SCTPAddr, options InitMsg, // close socket on error defer func() { if err != nil { - syscall.Close(sock) + _ = closeSctpSocket(sock, establishTimeout) } }() if err = setDefaultSockopts(sock, af, ipv6only); err != nil { @@ -354,7 +458,7 @@ func dialSCTPExtConfig(network string, laddr, raddr *SCTPAddr, options InitMsg, laddr.IPAddrs = append(laddr.IPAddrs, net.IPAddr{IP: net.IPv6zero}) } } - err = SCTPBind(sock, laddr, SCTP_BINDX_ADD_ADDR) + err = SCTPBind(sock, laddr, SCTP_BINDX_ADD_ADDR) // error EADDRINUSE "Address already in use" may occur if resource (source IP and Port) is occupied if err != nil { return nil, err } diff --git a/sctp_unsupported.go b/sctp_unsupported.go index 929fc0c..d1ed34c 100644 --- a/sctp_unsupported.go +++ b/sctp_unsupported.go @@ -1,4 +1,6 @@ +//go:build !linux || (linux && 386) // +build !linux linux,386 + // Copyright 2019 Wataru Ishida. All rights reserved. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -22,6 +24,7 @@ import ( "os" "runtime" "syscall" + "time" ) var ErrUnsupported = errors.New("SCTP is unsupported on " + runtime.GOOS + "/" + runtime.GOARCH) @@ -46,6 +49,14 @@ func (c *SCTPConn) Close() error { return ErrUnsupported } +func (c *SCTPConn) Abort() error { + return ErrUnsupported +} + +func (c *SCTPConn) CloseWithTimeout(timeout time.Duration) error { + return ErrUnsupported +} + func (c *SCTPConn) SetWriteBuffer(bytes int) error { return ErrUnsupported } diff --git a/testdata/closeprobe/main.go b/testdata/closeprobe/main.go new file mode 100644 index 0000000..57e0fa1 --- /dev/null +++ b/testdata/closeprobe/main.go @@ -0,0 +1,96 @@ +//go:build linux + +// Drives the close paths so a packet capture can show what each one puts on +// the wire: Close must complete a SHUTDOWN handshake, Abort must send ABORT, +// and Close against an unresponsive peer must fall back to ABORT rather than +// hanging. +// +// Each mode runs on its own port so the capture can be split by association. +package main + +import ( + "flag" + "fmt" + "os" + "time" + + "github.com/ishidawataru/sctp" +) + +func dialPair(port int) (*sctp.SCTPConn, *sctp.SCTPListener, chan *sctp.SCTPConn) { + addr, err := sctp.ResolveSCTPAddr("sctp", fmt.Sprintf("127.0.0.1:%d", port)) + if err != nil { + panic(err) + } + ln, err := sctp.ListenSCTP("sctp", addr) + if err != nil { + panic(err) + } + accepted := make(chan *sctp.SCTPConn, 1) + go func() { + c, err := ln.AcceptSCTP() + if err != nil { + accepted <- nil + return + } + accepted <- c + }() + + conn, err := sctp.DialSCTP("sctp", nil, ln.Addr().(*sctp.SCTPAddr)) + if err != nil { + panic(err) + } + return conn, ln, accepted +} + +func main() { + mode := flag.String("mode", "close", "close | abort | close-unresponsive") + port := flag.Int("port", 0, "local port to use") + flag.Parse() + + conn, ln, accepted := dialPair(*port) + srv := <-accepted + + // Exchange a message so the association is fully established and the + // capture shows a real data flow before the teardown. + if _, err := conn.SCTPWrite([]byte("hello"), nil); err != nil { + panic(err) + } + if srv != nil { + buf := make([]byte, 64) + if _, _, err := srv.SCTPRead(buf); err != nil { + panic(err) + } + } + time.Sleep(150 * time.Millisecond) + + start := time.Now() + var err error + switch *mode { + case "close": + err = conn.Close() + case "abort": + err = conn.Abort() + case "close-unresponsive": + // Kill the peer's socket without a graceful teardown, so nothing + // answers the SHUTDOWN and Close must fall back to ABORT. + if srv != nil { + _ = srv.Abort() + srv = nil + } + time.Sleep(100 * time.Millisecond) + err = conn.Close() + default: + fmt.Fprintf(os.Stderr, "unknown mode %q\n", *mode) + os.Exit(2) + } + elapsed := time.Since(start) + + fmt.Printf("mode=%-20s elapsed=%-14v err=%v\n", *mode, elapsed, err) + + if srv != nil { + _ = srv.Close() + } + _ = ln.Close() + time.Sleep(200 * time.Millisecond) +} diff --git a/testdata/tshark-close.sh b/testdata/tshark-close.sh new file mode 100755 index 0000000..d944f92 --- /dev/null +++ b/testdata/tshark-close.sh @@ -0,0 +1,64 @@ +#!/usr/bin/env bash +# Wire-level verification of the close paths. +# +# Confirms on the wire that Close completes a SHUTDOWN handshake, that Abort +# sends an ABORT chunk, and that Close against an unresponsive peer falls back +# to ABORT instead of hanging. +set -euo pipefail + +modprobe sctp 2>/dev/null || true +CAP=/tmp/sctp-close.pcap + +cd /src/testdata/closeprobe +cat > go.mod < /src +EOF +trap 'rm -f go.mod go.sum' EXIT +go mod tidy >/dev/null 2>&1 +go build -o /tmp/closeprobe . >/dev/null + +echo "Capturing loopback SCTP..." +tshark -i lo -f "ip proto 132" -w "$CAP" -q >/dev/null 2>&1 & +TPID=$! +sleep 3 + +# Distinct ports so each teardown can be isolated in the capture. +/tmp/closeprobe -mode close -port 39001 +/tmp/closeprobe -mode abort -port 39002 +/tmp/closeprobe -mode close-unresponsive -port 39003 + +sleep 2 +kill -INT $TPID 2>/dev/null || true +wait $TPID 2>/dev/null || true + +# chunk type 7 = SHUTDOWN, 8 = SHUTDOWN_ACK, 14 = SHUTDOWN_COMPLETE, 6 = ABORT +summarise() { + local port="$1" label="$2" + echo + echo "=== ${label} (port ${port}) ===" + tshark -r "$CAP" -Y "sctp.port == ${port}" -T fields -e sctp.chunk_type \ + 2>/dev/null | tr ',' '\n' | grep -v '^$' | sort -n | uniq -c \ + | awk '{ + name = "type " $2; + if ($2==0) name="DATA"; else if ($2==1) name="INIT"; + else if ($2==2) name="INIT_ACK"; else if ($2==3) name="SACK"; + else if ($2==6) name="ABORT"; else if ($2==7) name="SHUTDOWN"; + else if ($2==8) name="SHUTDOWN_ACK"; else if ($2==10) name="COOKIE_ECHO"; + else if ($2==11) name="COOKIE_ACK"; else if ($2==14) name="SHUTDOWN_COMPLETE"; + printf " %-18s count=%s\n", name, $1 + }' + + local shutdown abort + shutdown=$(tshark -r "$CAP" -Y "sctp.port == ${port}" -T fields -e sctp.chunk_type 2>/dev/null \ + | tr ',' '\n' | grep -c '^7$' || true) + abort=$(tshark -r "$CAP" -Y "sctp.port == ${port}" -T fields -e sctp.chunk_type 2>/dev/null \ + | tr ',' '\n' | grep -c '^6$' || true) + echo " -> SHUTDOWN chunks=${shutdown}, ABORT chunks=${abort}" +} + +summarise 39001 "Close: expect a SHUTDOWN handshake, no ABORT" +summarise 39002 "Abort: expect an ABORT chunk, no SHUTDOWN" +summarise 39003 "Close with unresponsive peer: expect the ABORT fallback"