diff --git a/internal/headers/cri/pipe.go b/internal/headers/cri/pipe.go index 203f8bd9..3fdc860e 100644 --- a/internal/headers/cri/pipe.go +++ b/internal/headers/cri/pipe.go @@ -159,6 +159,15 @@ func (c *pipeConn) Read(p []byte) (int, error) { } }) if err != nil { + // Control can reject a closed file before invoking the callback. + c.mu.Lock() + closed := c.closed + c.mu.Unlock() + + if closed { + err = os.ErrClosed + } + return 0, err } @@ -200,6 +209,8 @@ func (c *pipeConn) SetDeadline(t time.Time) error { return c.SetWriteDeadline(t) type readAheadConn struct { net.Conn + readWG sync.WaitGroup + mu sync.Mutex cond *sync.Cond buf []byte @@ -210,7 +221,7 @@ func newReadAheadConn(conn net.Conn) *readAheadConn { c := &readAheadConn{Conn: conn} c.cond = sync.NewCond(&c.mu) - go c.readLoop() + c.readWG.Go(c.readLoop) return c } @@ -255,6 +266,14 @@ func (c *readAheadConn) Read(p []byte) (int, error) { return n, nil } +// Close cancels the read and waits for the background reader to exit. +func (c *readAheadConn) Close() error { + err := c.Conn.Close() + c.readWG.Wait() + + return err +} + // SetDeadline only applies to writes. A read deadline would stop the background reader for good. func (c *readAheadConn) SetDeadline(t time.Time) error { return c.SetWriteDeadline(t) diff --git a/internal/headers/cri/pipe_internal_test.go b/internal/headers/cri/pipe_internal_test.go index d60ddd9d..bd47996a 100644 --- a/internal/headers/cri/pipe_internal_test.go +++ b/internal/headers/cri/pipe_internal_test.go @@ -26,6 +26,7 @@ import ( "sync" "sync/atomic" "testing" + "testing/synctest" "time" "github.com/stretchr/testify/require" @@ -280,6 +281,11 @@ func TestPipeConnClose(t *testing.T) { case <-time.After(5 * time.Second): t.Fatal("Close did not release the pipe handle") } + + // A read started after Close must return the same error as a canceled read. + n, err := server.Read(make([]byte, 1)) + require.Zero(t, n) + require.ErrorIs(t, err, os.ErrClosed) } // Both peers write first. On a pipe without buffer, this blocks both writes @@ -394,3 +400,64 @@ func TestReadAheadConnClose(t *testing.T) { t.Fatal("pending read was not unblocked by Close") } } + +// delayedReadConn keeps a canceled read active until the test releases it. +type delayedReadConn struct { + net.Conn + + started chan struct{} + canceled chan struct{} + finish <-chan struct{} +} + +func (c *delayedReadConn) Read(p []byte) (int, error) { + close(c.started) + + n, err := c.Conn.Read(p) + close(c.canceled) + <-c.finish + + return n, err +} + +func TestReadAheadConnCloseWaitsForReader(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + client, server := net.Pipe() + defer server.Close() + + finish := make(chan struct{}) + + release := sync.OnceFunc(func() { close(finish) }) + defer release() + + reader := &delayedReadConn{ + Conn: client, + started: make(chan struct{}), + canceled: make(chan struct{}), + finish: finish, + } + conn := newReadAheadConn(reader) + <-reader.started + + closed := make(chan error, 1) + + go func() { closed <- conn.Close() }() + + <-reader.canceled + synctest.Wait() + + select { + case <-closed: + t.Fatal("Close returned before the background read finished") + default: + } + + release() + require.NoError(t, <-closed) + + _, err := conn.Read(make([]byte, 1)) + require.ErrorIs(t, err, io.ErrClosedPipe) + }) +}