Fix splices to FDs that call usermem.IO.CopyIn/CopyInTo more than once.

Fixes #9932.

When Go is able to detect `io.Copy()` from a TCP socket or `AF_UNIX` stream
socket to a TCP socket, it attempts to implement the copy as a `splice(2)` from
the source to a pipe, followed by a `splice(2)` from the pipe to the
destination [1] (since `splice(2)` requires that one of the endpoints be a
pipe); the size of the pipe is set to 1 MB [2] (from a default of 64 KB [3]) to
reduce the number of splice syscalls required. In gVisor, a bug causes each
splice syscall from pipe to TCP socket to repeatedly read the *first* 64 KB [4]
of the pipe's data (when it contains more than 64 KB of data) rather than
*successive* chunks of 64 KB.

To fix this, advance pipe state by calling `Pipe.consumeLocked()` immediately
after `Pipe.peekLocked()`. Also defensively check that such FDs call
`Pipe.(usermem.IO)` methods on sequential addresses, and change
`fuse.deviceFD.Write()` to have this property.

[1] Go: `net/tcpsock_posix.go:TCPConn.readFrom()` =>
`net/splice_linux.go:splice()` => `internal/poll/splice_linux.go:Splice()`

[2] Go: `internal/poll/splice_linux.go:newPipe()` => `maxSpliceSize`

[3] `pkg/kernel/pipe/pipe.go:DefaultPipeSize`

[4] `pkg/tcpip/transport/tcp/endpoint.go:endpoint.Write()` =>
`endpoint.queueSegment()` => `endpoint.readFromPayloader()` =>
`pkg/buffer/buffer.go:Buffer.WriteFromReader()` =>
`pkg/buffer/chunk.go:MaxChunkSize`

PiperOrigin-RevId: 603151951

pkg/sentry/fsimpl/fuse/dev.go

@@ -225,7 +225,8 @@ func (fd *DeviceFD) Write(ctx context.Context, src usermem.IOSequence, opts vfs.
 		return 0, linuxerr.EPERM
 	}
 
-	if _, err := src.CopyIn(ctx, fd.writeBuf[:]); err != nil {
+	n, err := src.CopyIn(ctx, fd.writeBuf[:])
+	if err != nil {
 		return 0, err
 	}
 	var hdr linux.FUSEHeaderOut
@@ -243,9 +244,14 @@ func (fd *DeviceFD) Write(ctx context.Context, src usermem.IOSequence, opts vfs.
 	// will be copied over to the FR's data in the next iteration.
 	fut.hdr = &hdr
 	fut.data = make([]byte, fut.hdr.Len)
-	n, err := src.CopyIn(ctx, fut.data)
-	if err != nil {
-		return 0, err
+	copy(fut.data, fd.writeBuf[:])
+	if fut.hdr.Len > uint32(len(fd.writeBuf)) {
+		src = src.DropFirst(len(fd.writeBuf))
+		n2, err := src.CopyIn(ctx, fut.data[len(fd.writeBuf):])
+		if err != nil {
+			return 0, err
+		}
+		n += n2
 	}
 	if err := fd.sendResponse(ctx, fut); err != nil {
 		return 0, err

pkg/sentry/kernel/pipe/BUILD

@@ -46,6 +46,7 @@ go_library(
         "//pkg/context",
         "//pkg/errors/linuxerr",
         "//pkg/hostarch",
+        "//pkg/log",
         "//pkg/marshal/primitive",
         "//pkg/safemem",
         "//pkg/sentry/arch",

pkg/sentry/kernel/pipe/vfs.go

@@ -19,6 +19,7 @@ import (
 	"gvisor.dev/gvisor/pkg/context"
 	"gvisor.dev/gvisor/pkg/errors/linuxerr"
 	"gvisor.dev/gvisor/pkg/hostarch"
+	"gvisor.dev/gvisor/pkg/log"
 	"gvisor.dev/gvisor/pkg/safemem"
 	"gvisor.dev/gvisor/pkg/sentry/arch"
 	"gvisor.dev/gvisor/pkg/sentry/vfs"
@@ -165,6 +166,10 @@ type VFSPipeFD struct {
 	vfs.LockFD
 
 	pipe *Pipe
+
+	// lastAddr is the last hostarch.Addr at which a call to a
+	// VFSPipeFD.(usermem.IO) method ended. lastAddr is protected by pipe.mu.
+	lastAddr hostarch.Addr
 }
 
 // Release implements vfs.FileDescriptionImpl.Release.
@@ -268,14 +273,12 @@ func (fd *VFSPipeFD) SpliceToNonPipe(ctx context.Context, out *vfs.FileDescripti
 		n   int64
 		err error
 	)
+	fd.lastAddr = 0
 	if off == -1 {
 		n, err = out.Write(ctx, src, vfs.WriteOptions{})
 	} else {
 		n, err = out.PWrite(ctx, src, off, vfs.WriteOptions{})
 	}
-	if n > 0 {
-		fd.pipe.consumeLocked(n)
-	}
 
 	// Implementations of out.[P]Write() that ignore written data (e.g.
 	// /dev/null) may skip calling src.CopyIn[To]() and therefore miss getting
@@ -304,6 +307,7 @@ func (fd *VFSPipeFD) SpliceFromNonPipe(ctx context.Context, in *vfs.FileDescript
 		err error
 	)
 	fd.pipe.mu.Lock()
+	fd.lastAddr = 0
 	if off == -1 {
 		n, err = in.Read(ctx, dst, vfs.ReadOptions{})
 	} else {
@@ -318,14 +322,20 @@ func (fd *VFSPipeFD) SpliceFromNonPipe(ctx context.Context, in *vfs.FileDescript
 }
 
 // CopyIn implements usermem.IO.CopyIn. Note that it is the caller's
-// responsibility to call fd.pipe.consumeLocked() and
-// fd.pipe.Notify(waiter.WritableEvents) after the read is completed.
+// responsibility to call fd.pipe.Notify(waiter.WritableEvents) after the read
+// is completed.
 //
 // Preconditions: fd.pipe.mu must be locked.
 func (fd *VFSPipeFD) CopyIn(ctx context.Context, addr hostarch.Addr, dst []byte, opts usermem.IOOpts) (int, error) {
+	if addr != fd.lastAddr {
+		log.Traceback("Non-sequential VFSPipeFD.CopyIn: lastAddr=%#x addr=%#x", fd.lastAddr, addr)
+		return 0, linuxerr.EINVAL
+	}
 	n, err := fd.pipe.peekLocked(int64(len(dst)), func(srcs safemem.BlockSeq) (uint64, error) {
 		return safemem.CopySeq(safemem.BlockSeqOf(safemem.BlockFromSafeSlice(dst)), srcs)
 	})
+	fd.pipe.consumeLocked(n)
+	fd.lastAddr = addr + hostarch.Addr(n)
 	return int(n), err
 }
 
@@ -335,9 +345,14 @@ func (fd *VFSPipeFD) CopyIn(ctx context.Context, addr hostarch.Addr, dst []byte,
 //
 // Preconditions: fd.pipe.mu must be locked.
 func (fd *VFSPipeFD) CopyOut(ctx context.Context, addr hostarch.Addr, src []byte, opts usermem.IOOpts) (int, error) {
+	if addr != fd.lastAddr {
+		log.Traceback("Non-sequential VFSPipeFD.CopyOut: lastAddr=%#x addr=%#x", fd.lastAddr, addr)
+		return 0, linuxerr.EINVAL
+	}
 	n, err := fd.pipe.writeLocked(int64(len(src)), func(dsts safemem.BlockSeq) (uint64, error) {
 		return safemem.CopySeq(dsts, safemem.BlockSeqOf(safemem.BlockFromSafeSlice(src)))
 	})
+	fd.lastAddr = addr + hostarch.Addr(n)
 	return int(n), err
 }
 
@@ -345,9 +360,14 @@ func (fd *VFSPipeFD) CopyOut(ctx context.Context, addr hostarch.Addr, src []byte
 //
 // Preconditions: fd.pipe.mu must be locked.
 func (fd *VFSPipeFD) ZeroOut(ctx context.Context, addr hostarch.Addr, toZero int64, opts usermem.IOOpts) (int64, error) {
+	if addr != fd.lastAddr {
+		log.Traceback("Non-sequential VFSPipeFD.ZeroOut: lastAddr=%#x addr=%#x", fd.lastAddr, addr)
+		return 0, linuxerr.EINVAL
+	}
 	n, err := fd.pipe.writeLocked(toZero, func(dsts safemem.BlockSeq) (uint64, error) {
 		return safemem.ZeroSeq(dsts)
 	})
+	fd.lastAddr = addr + hostarch.Addr(n)
 	return n, err
 }
 
@@ -357,9 +377,25 @@ func (fd *VFSPipeFD) ZeroOut(ctx context.Context, addr hostarch.Addr, toZero int
 //
 // Preconditions: fd.pipe.mu must be locked.
 func (fd *VFSPipeFD) CopyInTo(ctx context.Context, ars hostarch.AddrRangeSeq, dst safemem.Writer, opts usermem.IOOpts) (int64, error) {
-	return fd.pipe.peekLocked(ars.NumBytes(), func(srcs safemem.BlockSeq) (uint64, error) {
-		return dst.WriteFromBlocks(srcs)
-	})
+	total := int64(0)
+	for !ars.IsEmpty() {
+		ar := ars.Head()
+		if ar.Start != fd.lastAddr {
+			log.Traceback("Non-sequential VFSPipeFD.CopyInTo: lastAddr=%#x addr=%#x", fd.lastAddr, ar.Start)
+			return total, linuxerr.EINVAL
+		}
+		n, err := fd.pipe.peekLocked(int64(ar.Length()), func(srcs safemem.BlockSeq) (uint64, error) {
+			return dst.WriteFromBlocks(srcs)
+		})
+		fd.pipe.consumeLocked(n)
+		fd.lastAddr = ar.Start + hostarch.Addr(n)
+		total += n
+		if err != nil {
+			return total, err
+		}
+		ars = ars.Tail()
+	}
+	return total, nil
 }
 
 // CopyOutFrom implements usermem.IO.CopyOutFrom. Note that it is the caller's
@@ -368,9 +404,24 @@ func (fd *VFSPipeFD) CopyInTo(ctx context.Context, ars hostarch.AddrRangeSeq, ds
 //
 // Preconditions: fd.pipe.mu must be locked.
 func (fd *VFSPipeFD) CopyOutFrom(ctx context.Context, ars hostarch.AddrRangeSeq, src safemem.Reader, opts usermem.IOOpts) (int64, error) {
-	return fd.pipe.writeLocked(ars.NumBytes(), func(dsts safemem.BlockSeq) (uint64, error) {
-		return src.ReadToBlocks(dsts)
-	})
+	total := int64(0)
+	for !ars.IsEmpty() {
+		ar := ars.Head()
+		if ar.Start != fd.lastAddr {
+			log.Traceback("Non-sequential VFSPipeFD.CopyOutFrom: lastAddr=%#x addr=%#x", fd.lastAddr, ar.Start)
+			return total, linuxerr.EINVAL
+		}
+		n, err := fd.pipe.writeLocked(int64(ar.Length()), func(dsts safemem.BlockSeq) (uint64, error) {
+			return src.ReadToBlocks(dsts)
+		})
+		fd.lastAddr = ar.Start + hostarch.Addr(n)
+		total += n
+		if err != nil {
+			return total, err
+		}
+		ars = ars.Tail()
+	}
+	return total, nil
 }
 
 // SwapUint32 implements usermem.IO.SwapUint32.

test/syscalls/linux/socket_ip_tcp_generic.cc

@@ -1119,6 +1119,52 @@ TEST_P(TCPSocketPairTest, SpliceToPipe) {
   EXPECT_EQ(memcmp(rbuf.data(), buf.data(), buf.size()), 0);
 }
 
+// Regression test for #9932.
+TEST_P(TCPSocketPairTest, LargeSpliceFromPipe) {
+  auto sockets = ASSERT_NO_ERRNO_AND_VALUE(NewSocketPair());
+
+  // Create a pipe, increase its size from the default 64K, and fill it with
+  // data.
+  int pipe_fds[2];
+  ASSERT_THAT(pipe(pipe_fds), SyscallSucceeds());
+  const FileDescriptor pipe_rfd(pipe_fds[0]);
+  const FileDescriptor pipe_wfd(pipe_fds[1]);
+  constexpr size_t kPipeSize = 1 << 20;
+  ASSERT_THAT(fcntl(pipe_wfd.get(), F_SETPIPE_SZ, kPipeSize),
+              SyscallSucceeds());
+  std::vector<char> orig_data(kPipeSize);
+  RandomizeBuffer(orig_data.data(), orig_data.size());
+  ASSERT_THAT(WriteFd(pipe_wfd.get(), orig_data.data(), orig_data.size()),
+              SyscallSucceedsWithValue(orig_data.size()));
+
+  // Splice all data from the pipe to one end of the TCP socket pair in a
+  // separate thread, while draining the other end from this thread.
+  std::vector<char> read_data(orig_data.size());
+  ScopedThread reader_thread([&] {
+    size_t spliced_bytes = 0;
+    ssize_t n;
+    while (spliced_bytes < orig_data.size()) {
+      ASSERT_THAT(
+          n = RetryEINTR(splice)(pipe_rfd.get(), nullptr, sockets->first_fd(),
+                                 nullptr, orig_data.size() - spliced_bytes, 0),
+          SyscallSucceeds());
+      spliced_bytes += n;
+    }
+  });
+  size_t read_bytes = 0;
+  while (read_bytes < read_data.size()) {
+    ssize_t n;
+    ASSERT_THAT(n = RetryEINTR(read)(sockets->second_fd(),
+                                     read_data.data() + read_bytes,
+                                     read_data.size() - read_bytes),
+                SyscallSucceeds());
+    read_bytes += n;
+  }
+
+  // Check that correct data was spliced and read.
+  EXPECT_EQ(0, memcmp(orig_data.data(), read_data.data(), orig_data.size()));
+}
+
 #include <sys/sendfile.h>
 
 #include <memory>