Simplify the pipe implementation.

- Remove the pipe package's dependence on the buffer package, which becomes
  unused as a result. The buffer package is currently intended to serve two use
  cases, pipes and temporary buffers, and does neither optimally as a result;
  this change facilitates retooling the buffer package to better serve the
  latter.

- Pass callbacks taking safemem.BlockSeq to the internal pipe I/O methods,
  which makes most callbacks trivial.

- Fix VFS1's splice() and tee() to immediately return if a pipe returns a
  partial write.

PiperOrigin-RevId: 351911375

1 files changed, 106 insertions, 0 deletions

diff --git a/test/syscalls/linux/splice.cc b/test/syscalls/linux/splice.cc
index c2369db54..e5730a606 100644
--- a/test/syscalls/linux/splice.cc
+++ b/test/syscalls/linux/splice.cc
@@ -483,6 +483,112 @@ TEST(SpliceTest, TwoPipes) {
   EXPECT_EQ(memcmp(rbuf.data(), buf.data(), kPageSize), 0);
 }
 
+TEST(SpliceTest, TwoPipesPartialRead) {
+  // Create two pipes.
+  int fds[2];
+  ASSERT_THAT(pipe(fds), SyscallSucceeds());
+  const FileDescriptor first_rfd(fds[0]);
+  const FileDescriptor first_wfd(fds[1]);
+  ASSERT_THAT(pipe(fds), SyscallSucceeds());
+  const FileDescriptor second_rfd(fds[0]);
+  const FileDescriptor second_wfd(fds[1]);
+
+  // Write half a page of data to the first pipe.
+  std::vector<char> buf(kPageSize / 2);
+  RandomizeBuffer(buf.data(), buf.size());
+  ASSERT_THAT(write(first_wfd.get(), buf.data(), buf.size()),
+              SyscallSucceedsWithValue(kPageSize / 2));
+
+  // Attempt to splice one page from the first pipe to the second; it should
+  // immediately return after splicing the half-page previously written to the
+  // first pipe.
+  EXPECT_THAT(
+      splice(first_rfd.get(), nullptr, second_wfd.get(), nullptr, kPageSize, 0),
+      SyscallSucceedsWithValue(kPageSize / 2));
+}
+
+TEST(SpliceTest, TwoPipesPartialWrite) {
+  // Create two pipes.
+  int fds[2];
+  ASSERT_THAT(pipe(fds), SyscallSucceeds());
+  const FileDescriptor first_rfd(fds[0]);
+  const FileDescriptor first_wfd(fds[1]);
+  ASSERT_THAT(pipe(fds), SyscallSucceeds());
+  const FileDescriptor second_rfd(fds[0]);
+  const FileDescriptor second_wfd(fds[1]);
+
+  // Write two pages of data to the first pipe.
+  std::vector<char> buf(2 * kPageSize);
+  RandomizeBuffer(buf.data(), buf.size());
+  ASSERT_THAT(write(first_wfd.get(), buf.data(), buf.size()),
+              SyscallSucceedsWithValue(2 * kPageSize));
+
+  // Limit the second pipe to two pages, then write one page of data to it.
+  ASSERT_THAT(fcntl(second_wfd.get(), F_SETPIPE_SZ, 2 * kPageSize),
+              SyscallSucceeds());
+  ASSERT_THAT(write(second_wfd.get(), buf.data(), buf.size() / 2),
+              SyscallSucceedsWithValue(kPageSize));
+
+  // Attempt to splice two pages from the first pipe to the second; it should
+  // immediately return after splicing the first page previously written to the
+  // first pipe.
+  EXPECT_THAT(splice(first_rfd.get(), nullptr, second_wfd.get(), nullptr,
+                     2 * kPageSize, 0),
+              SyscallSucceedsWithValue(kPageSize));
+}
+
+TEST(TeeTest, TwoPipesPartialRead) {
+  // Create two pipes.
+  int fds[2];
+  ASSERT_THAT(pipe(fds), SyscallSucceeds());
+  const FileDescriptor first_rfd(fds[0]);
+  const FileDescriptor first_wfd(fds[1]);
+  ASSERT_THAT(pipe(fds), SyscallSucceeds());
+  const FileDescriptor second_rfd(fds[0]);
+  const FileDescriptor second_wfd(fds[1]);
+
+  // Write half a page of data to the first pipe.
+  std::vector<char> buf(kPageSize / 2);
+  RandomizeBuffer(buf.data(), buf.size());
+  ASSERT_THAT(write(first_wfd.get(), buf.data(), buf.size()),
+              SyscallSucceedsWithValue(kPageSize / 2));
+
+  // Attempt to tee one page from the first pipe to the second; it should
+  // immediately return after copying the half-page previously written to the
+  // first pipe.
+  EXPECT_THAT(tee(first_rfd.get(), second_wfd.get(), kPageSize, 0),
+              SyscallSucceedsWithValue(kPageSize / 2));
+}
+
+TEST(TeeTest, TwoPipesPartialWrite) {
+  // Create two pipes.
+  int fds[2];
+  ASSERT_THAT(pipe(fds), SyscallSucceeds());
+  const FileDescriptor first_rfd(fds[0]);
+  const FileDescriptor first_wfd(fds[1]);
+  ASSERT_THAT(pipe(fds), SyscallSucceeds());
+  const FileDescriptor second_rfd(fds[0]);
+  const FileDescriptor second_wfd(fds[1]);
+
+  // Write two pages of data to the first pipe.
+  std::vector<char> buf(2 * kPageSize);
+  RandomizeBuffer(buf.data(), buf.size());
+  ASSERT_THAT(write(first_wfd.get(), buf.data(), buf.size()),
+              SyscallSucceedsWithValue(2 * kPageSize));
+
+  // Limit the second pipe to two pages, then write one page of data to it.
+  ASSERT_THAT(fcntl(second_wfd.get(), F_SETPIPE_SZ, 2 * kPageSize),
+              SyscallSucceeds());
+  ASSERT_THAT(write(second_wfd.get(), buf.data(), buf.size() / 2),
+              SyscallSucceedsWithValue(kPageSize));
+
+  // Attempt to tee two pages from the first pipe to the second; it should
+  // immediately return after copying the first page previously written to the
+  // first pipe.
+  EXPECT_THAT(tee(first_rfd.get(), second_wfd.get(), 2 * kPageSize, 0),
+              SyscallSucceedsWithValue(kPageSize));
+}
+
 TEST(SpliceTest, TwoPipesCircular) {
   // This test deadlocks the sentry on VFS1 because VFS1 splice ordering is
   // based on fs.File.UniqueID, which does not prevent circular ordering between