Use send()/recv() in IoSocketHandleImpl::writev()/readv() when there …

…is only one IO buffer to work with. (envoyproxy#25695)

* Use write()/read() in OsSysCallsImpl::writev()/readv() when there is
only one IO buffer to work with.

Internal testing on Redis filter shows that this change gives 7%
performance improvement.

Signed-off-by: Wei Si <[email protected]>

envoy/api/os_sys_calls.h

@@ -84,6 +84,11 @@ class OsSysCalls {
    */
   virtual SysCallSizeResult pread(os_fd_t fd, void* buffer, size_t length, off_t offset) const PURE;
 
+  /**
+   * @see send (man 2 send)
+   */
+  virtual SysCallSizeResult send(os_fd_t socket, void* buffer, size_t length, int flags) PURE;
+
   /**
    * @see recv (man 2 recv)
    */

source/common/api/posix/os_sys_calls_impl.cc

@@ -63,6 +63,11 @@ SysCallSizeResult OsSysCallsImpl::pread(os_fd_t fd, void* buffer, size_t length,
   return {rc, rc != -1 ? 0 : errno};
 }
 
+SysCallSizeResult OsSysCallsImpl::send(os_fd_t socket, void* buffer, size_t length, int flags) {
+  const ssize_t rc = ::send(socket, buffer, length, flags);
+  return {rc, rc != -1 ? 0 : errno};
+}
+
 SysCallSizeResult OsSysCallsImpl::recv(os_fd_t socket, void* buffer, size_t length, int flags) {
   const ssize_t rc = ::recv(socket, buffer, length, flags);
   return {rc, rc != -1 ? 0 : errno};

source/common/api/posix/os_sys_calls_impl.h

@@ -6,6 +6,8 @@
 
 #include "source/common/singleton/threadsafe_singleton.h"
 
+#define ENVOY_DEFAULT_PIPE_TYPE AF_UNIX
+
 namespace Envoy {
 namespace Api {
 
@@ -21,6 +23,7 @@ class OsSysCallsImpl : public OsSysCalls {
   SysCallSizeResult pwrite(os_fd_t fd, const void* buffer, size_t length,
                            off_t offset) const override;
   SysCallSizeResult pread(os_fd_t fd, void* buffer, size_t length, off_t offset) const override;
+  SysCallSizeResult send(os_fd_t socket, void* buffer, size_t length, int flags) override;
   SysCallSizeResult recv(os_fd_t socket, void* buffer, size_t length, int flags) override;
   SysCallSizeResult recvmsg(os_fd_t sockfd, msghdr* msg, int flags) override;
   SysCallIntResult recvmmsg(os_fd_t sockfd, struct mmsghdr* msgvec, unsigned int vlen, int flags,

source/common/api/win32/os_sys_calls_impl.cc

@@ -158,6 +158,11 @@ SysCallSizeResult OsSysCallsImpl::pread(os_fd_t fd, void* buffer, size_t length,
   PANIC("not implemented");
 }
 
+SysCallSizeResult OsSysCallsImpl::send(os_fd_t socket, void* buffer, size_t length, int flags) {
+  const ssize_t rc = ::send(socket, static_cast<char*>(buffer), length, flags);
+  return {rc, rc != -1 ? 0 : ::WSAGetLastError()};
+}
+
 SysCallSizeResult OsSysCallsImpl::recv(os_fd_t socket, void* buffer, size_t length, int flags) {
   const ssize_t rc = ::recv(socket, static_cast<char*>(buffer), length, flags);
   return {rc, rc != -1 ? 0 : ::WSAGetLastError()};

source/common/api/win32/os_sys_calls_impl.h

@@ -6,6 +6,8 @@
 
 #include "source/common/singleton/threadsafe_singleton.h"
 
+#define ENVOY_DEFAULT_PIPE_TYPE AF_INET
+
 namespace Envoy {
 namespace Api {
 
@@ -22,6 +24,7 @@ class OsSysCallsImpl : public OsSysCalls {
   SysCallSizeResult pwrite(os_fd_t fd, const void* buffer, size_t length,
                            off_t offset) const override;
   SysCallSizeResult pread(os_fd_t fd, void* buffer, size_t length, off_t offset) const override;
+  SysCallSizeResult send(os_fd_t socket, void* buffer, size_t length, int flags) override;
   SysCallSizeResult recv(os_fd_t socket, void* buffer, size_t length, int flags) override;
   SysCallSizeResult recvmsg(os_fd_t sockfd, msghdr* msg, int flags) override;
   SysCallIntResult recvmmsg(os_fd_t sockfd, struct mmsghdr* msgvec, unsigned int vlen, int flags,

source/common/network/io_socket_handle_impl.cc

@@ -95,6 +95,13 @@ Api::IoCallUint64Result IoSocketHandleImpl::readv(uint64_t max_length, Buffer::R
     num_bytes_to_read += slice_length;
   }
   ASSERT(num_bytes_to_read <= max_length);
+
+  if (num_slices_to_read == 1) {
+    // Avoid paying the VFS overhead when there is only one IO buffer to work with
+    return sysCallResultToIoCallResult(
+        Api::OsSysCallsSingleton::get().recv(fd_, iov[0].iov_base, iov[0].iov_len, 0));
+  }
+
   auto result = sysCallResultToIoCallResult(Api::OsSysCallsSingleton::get().readv(
       fd_, iov.begin(), static_cast<int>(num_slices_to_read)));
   return result;
@@ -129,6 +136,13 @@ Api::IoCallUint64Result IoSocketHandleImpl::writev(const Buffer::RawSlice* slice
   if (num_slices_to_write == 0) {
     return Api::ioCallUint64ResultNoError();
   }
+
+  if (num_slices_to_write == 1) {
+    // Avoid paying the VFS overhead when there is only one IO buffer to work with
+    return sysCallResultToIoCallResult(
+        Api::OsSysCallsSingleton::get().send(fd_, iov[0].iov_base, iov[0].iov_len, 0));
+  }
+
   auto result = sysCallResultToIoCallResult(
       Api::OsSysCallsSingleton::get().writev(fd_, iov.begin(), num_slices_to_write));
   return result;

test/common/buffer/buffer_fuzz.cc

@@ -4,6 +4,7 @@
 
 #include "envoy/common/platform.h"
 
+#include "source/common/api/os_sys_calls_impl.h"
 #include "source/common/buffer/buffer_impl.h"
 #include "source/common/common/assert.h"
 #include "source/common/common/logger.h"
@@ -391,25 +392,27 @@ uint32_t bufferAction(Context& ctxt, char insert_value, uint32_t max_alloc, Buff
     if (max_length == 0) {
       break;
     }
-    int pipe_fds[2] = {0, 0};
-    FUZZ_ASSERT(::pipe(pipe_fds) == 0);
-    Network::IoSocketHandleImpl io_handle(pipe_fds[0]);
-    FUZZ_ASSERT(::fcntl(pipe_fds[0], F_SETFL, O_NONBLOCK) == 0);
-    FUZZ_ASSERT(::fcntl(pipe_fds[1], F_SETFL, O_NONBLOCK) == 0);
+    int fds[2] = {0, 0};
+    auto& os_sys_calls = Api::OsSysCallsSingleton::get();
+    FUZZ_ASSERT(os_sys_calls.socketpair(AF_UNIX, SOCK_STREAM, 0, fds).return_value_ == 0);
+    Network::IoSocketHandleImpl io_handle(fds[0]);
+    FUZZ_ASSERT(::fcntl(fds[0], F_SETFL, O_NONBLOCK) == 0);
+    FUZZ_ASSERT(::fcntl(fds[1], F_SETFL, O_NONBLOCK) == 0);
     std::string data(max_length, insert_value);
-    const ssize_t rc = ::write(pipe_fds[1], data.data(), max_length);
+    const ssize_t rc = ::write(fds[1], data.data(), max_length);
     FUZZ_ASSERT(rc > 0);
     Api::IoCallUint64Result result = io_handle.read(target_buffer, max_length);
     FUZZ_ASSERT(result.return_value_ == static_cast<uint64_t>(rc));
-    FUZZ_ASSERT(::close(pipe_fds[1]) == 0);
+    FUZZ_ASSERT(::close(fds[1]) == 0);
     break;
   }
   case test::common::buffer::Action::kWrite: {
-    int pipe_fds[2] = {0, 0};
-    FUZZ_ASSERT(::pipe(pipe_fds) == 0);
-    Network::IoSocketHandleImpl io_handle(pipe_fds[1]);
-    FUZZ_ASSERT(::fcntl(pipe_fds[0], F_SETFL, O_NONBLOCK) == 0);
-    FUZZ_ASSERT(::fcntl(pipe_fds[1], F_SETFL, O_NONBLOCK) == 0);
+    int fds[2] = {0, 0};
+    auto& os_sys_calls = Api::OsSysCallsSingleton::get();
+    FUZZ_ASSERT(os_sys_calls.socketpair(AF_UNIX, SOCK_STREAM, 0, fds).return_value_ == 0);
+    Network::IoSocketHandleImpl io_handle(fds[1]);
+    FUZZ_ASSERT(::fcntl(fds[0], F_SETFL, O_NONBLOCK) == 0);
+    FUZZ_ASSERT(::fcntl(fds[1], F_SETFL, O_NONBLOCK) == 0);
     uint64_t return_value;
     do {
       const bool empty = target_buffer.length() == 0;
@@ -423,12 +426,11 @@ uint32_t bufferAction(Context& ctxt, char insert_value, uint32_t max_alloc, Buff
         FUZZ_ASSERT(return_value == 0);
       } else {
         auto buf = std::make_unique<char[]>(return_value);
-        FUZZ_ASSERT(static_cast<uint64_t>(::read(pipe_fds[0], buf.get(), return_value)) ==
-                    return_value);
+        FUZZ_ASSERT(static_cast<uint64_t>(::read(fds[0], buf.get(), return_value)) == return_value);
         FUZZ_ASSERT(::memcmp(buf.get(), previous_data.data(), return_value) == 0);
       }
     } while (return_value > 0);
-    FUZZ_ASSERT(::close(pipe_fds[0]) == 0);
+    FUZZ_ASSERT(::close(fds[0]) == 0);
     break;
   }
   case test::common::buffer::Action::kGetRawSlices: {

test/common/buffer/owned_impl_test.cc

@@ -1256,39 +1256,39 @@ TYPED_TEST(OwnedImplTypedTest, Write) {
   using IoSocketHandleType = typename TestFixture::IoSocketHandleTestType;
   IoSocketHandleType io_handle;
   buffer.add("example");
-  EXPECT_CALL(os_sys_calls, writev(_, _, _)).WillOnce(Return(Api::SysCallSizeResult{7, 0}));
+  EXPECT_CALL(os_sys_calls, send(_, _, _, _)).WillOnce(Return(Api::SysCallSizeResult{7, 0}));
   Api::IoCallUint64Result result = io_handle.write(buffer);
   EXPECT_TRUE(result.ok());
   EXPECT_EQ(7, result.return_value_);
   EXPECT_EQ(0, buffer.length());
 
   buffer.add("example");
-  EXPECT_CALL(os_sys_calls, writev(_, _, _)).WillOnce(Return(Api::SysCallSizeResult{6, 0}));
+  EXPECT_CALL(os_sys_calls, send(_, _, _, _)).WillOnce(Return(Api::SysCallSizeResult{6, 0}));
   result = io_handle.write(buffer);
   EXPECT_TRUE(result.ok());
   EXPECT_EQ(6, result.return_value_);
   EXPECT_EQ(1, buffer.length());
 
-  EXPECT_CALL(os_sys_calls, writev(_, _, _)).WillOnce(Return(Api::SysCallSizeResult{0, 0}));
+  EXPECT_CALL(os_sys_calls, send(_, _, _, _)).WillOnce(Return(Api::SysCallSizeResult{0, 0}));
   result = io_handle.write(buffer);
   EXPECT_TRUE(result.ok());
   EXPECT_EQ(0, result.return_value_);
   EXPECT_EQ(1, buffer.length());
 
-  EXPECT_CALL(os_sys_calls, writev(_, _, _)).WillOnce(Return(Api::SysCallSizeResult{-1, 0}));
+  EXPECT_CALL(os_sys_calls, send(_, _, _, _)).WillOnce(Return(Api::SysCallSizeResult{-1, 0}));
   result = io_handle.write(buffer);
   EXPECT_EQ(Api::IoError::IoErrorCode::UnknownError, result.err_->getErrorCode());
   EXPECT_EQ(0, result.return_value_);
   EXPECT_EQ(1, buffer.length());
 
-  EXPECT_CALL(os_sys_calls, writev(_, _, _))
+  EXPECT_CALL(os_sys_calls, send(_, _, _, _))
       .WillOnce(Return(Api::SysCallSizeResult{-1, SOCKET_ERROR_AGAIN}));
   result = io_handle.write(buffer);
   EXPECT_EQ(Api::IoError::IoErrorCode::Again, result.err_->getErrorCode());
   EXPECT_EQ(0, result.return_value_);
   EXPECT_EQ(1, buffer.length());
 
-  EXPECT_CALL(os_sys_calls, writev(_, _, _)).WillOnce(Return(Api::SysCallSizeResult{1, 0}));
+  EXPECT_CALL(os_sys_calls, send(_, _, _, _)).WillOnce(Return(Api::SysCallSizeResult{1, 0}));
   result = io_handle.write(buffer);
   EXPECT_TRUE(result.ok());
   EXPECT_EQ(1, result.return_value_);
@@ -1307,29 +1307,29 @@ TYPED_TEST(OwnedImplTypedTest, Read) {
   Buffer::OwnedImpl buffer;
   using IoSocketHandleType = typename TestFixture::IoSocketHandleTestType;
   IoSocketHandleType io_handle;
-  EXPECT_CALL(os_sys_calls, readv(_, _, _)).WillOnce(Return(Api::SysCallSizeResult{0, 0}));
+  EXPECT_CALL(os_sys_calls, recv(_, _, _, _)).WillOnce(Return(Api::SysCallSizeResult{0, 0}));
   Api::IoCallUint64Result result = io_handle.read(buffer, 100);
   EXPECT_TRUE(result.ok());
   EXPECT_EQ(0, result.return_value_);
   EXPECT_EQ(0, buffer.length());
   EXPECT_THAT(buffer.describeSlicesForTest(), testing::IsEmpty());
 
-  EXPECT_CALL(os_sys_calls, readv(_, _, _)).WillOnce(Return(Api::SysCallSizeResult{-1, 0}));
+  EXPECT_CALL(os_sys_calls, recv(_, _, _, _)).WillOnce(Return(Api::SysCallSizeResult{-1, 0}));
   result = io_handle.read(buffer, 100);
   EXPECT_EQ(Api::IoError::IoErrorCode::UnknownError, result.err_->getErrorCode());
   EXPECT_EQ(0, result.return_value_);
   EXPECT_EQ(0, buffer.length());
   EXPECT_THAT(buffer.describeSlicesForTest(), testing::IsEmpty());
 
-  EXPECT_CALL(os_sys_calls, readv(_, _, _))
+  EXPECT_CALL(os_sys_calls, recv(_, _, _, _))
       .WillOnce(Return(Api::SysCallSizeResult{-1, SOCKET_ERROR_AGAIN}));
   result = io_handle.read(buffer, 100);
   EXPECT_EQ(Api::IoError::IoErrorCode::Again, result.err_->getErrorCode());
   EXPECT_EQ(0, result.return_value_);
   EXPECT_EQ(0, buffer.length());
   EXPECT_THAT(buffer.describeSlicesForTest(), testing::IsEmpty());
 
-  EXPECT_CALL(os_sys_calls, readv(_, _, _)).Times(0);
+  EXPECT_CALL(os_sys_calls, recv(_, _, _, _)).Times(0);
   result = io_handle.read(buffer, 0);
   EXPECT_EQ(0, result.return_value_);
   EXPECT_EQ(0, buffer.length());
@@ -1348,25 +1348,21 @@ TYPED_TEST(OwnedImplTypedTest, ReserveZeroCommit) {
   OwnedImplTest::expectSlices({{5, 0, 4096}, {0, 0, 0}}, buf);
   { auto reservation = buf.reserveSingleSlice(1280); }
   OwnedImplTest::expectSlices({{5, 0, 4096}}, buf);
-  os_fd_t pipe_fds[2] = {0, 0};
+  os_fd_t fds[2] = {0, 0};
   auto& os_sys_calls = Api::OsSysCallsSingleton::get();
-#ifdef WIN32
-  ASSERT_EQ(os_sys_calls.socketpair(AF_INET, SOCK_STREAM, 0, pipe_fds).return_value_, 0);
-#else
-  ASSERT_EQ(pipe(pipe_fds), 0);
-#endif
+  ASSERT_EQ(os_sys_calls.socketpair(ENVOY_DEFAULT_PIPE_TYPE, SOCK_STREAM, 0, fds).return_value_, 0);
   using IoSocketHandleType = typename TestFixture::IoSocketHandleTestType;
-  IoSocketHandleType io_handle(pipe_fds[0]);
-  ASSERT_EQ(os_sys_calls.setsocketblocking(pipe_fds[0], false).return_value_, 0);
-  ASSERT_EQ(os_sys_calls.setsocketblocking(pipe_fds[1], false).return_value_, 0);
+  IoSocketHandleType io_handle(fds[0]);
+  ASSERT_EQ(os_sys_calls.setsocketblocking(fds[0], false).return_value_, 0);
+  ASSERT_EQ(os_sys_calls.setsocketblocking(fds[1], false).return_value_, 0);
   const uint32_t max_length = 1953;
   std::string data(max_length, 'e');
-  const ssize_t rc = os_sys_calls.write(pipe_fds[1], data.data(), max_length).return_value_;
+  const ssize_t rc = os_sys_calls.write(fds[1], data.data(), max_length).return_value_;
   ASSERT_GT(rc, 0);
   const uint32_t previous_length = buf.length();
   Api::IoCallUint64Result result = io_handle.read(buf, max_length);
   ASSERT_EQ(result.return_value_, static_cast<uint64_t>(rc));
-  ASSERT_EQ(os_sys_calls.close(pipe_fds[1]).return_value_, 0);
+  ASSERT_EQ(os_sys_calls.close(fds[1]).return_value_, 0);
   ASSERT_EQ(previous_length, buf.search(data.data(), rc, previous_length, 0));
   EXPECT_EQ("bbbbb", buf.toString().substr(0, 5));
   OwnedImplTest::expectSlices({{5, 0, 4096}, {1953, 14431, 16384}}, buf);
@@ -1377,25 +1373,21 @@ TYPED_TEST(OwnedImplTypedTest, ReadReserveAndCommit) {
   Buffer::OwnedImpl buf;
   buf.add("bbbbb");
 
-  os_fd_t pipe_fds[2] = {0, 0};
+  os_fd_t fds[2] = {0, 0};
   auto& os_sys_calls = Api::OsSysCallsSingleton::get();
-#ifdef WIN32
-  ASSERT_EQ(os_sys_calls.socketpair(AF_INET, SOCK_STREAM, 0, pipe_fds).return_value_, 0);
-#else
-  ASSERT_EQ(pipe(pipe_fds), 0);
-#endif
+  ASSERT_EQ(os_sys_calls.socketpair(ENVOY_DEFAULT_PIPE_TYPE, SOCK_STREAM, 0, fds).return_value_, 0);
   using IoSocketHandleType = typename TestFixture::IoSocketHandleTestType;
-  IoSocketHandleType io_handle(pipe_fds[0]);
-  ASSERT_EQ(os_sys_calls.setsocketblocking(pipe_fds[0], false).return_value_, 0);
-  ASSERT_EQ(os_sys_calls.setsocketblocking(pipe_fds[1], false).return_value_, 0);
+  IoSocketHandleType io_handle(fds[0]);
+  ASSERT_EQ(os_sys_calls.setsocketblocking(fds[0], false).return_value_, 0);
+  ASSERT_EQ(os_sys_calls.setsocketblocking(fds[1], false).return_value_, 0);
 
   const uint32_t read_length = 32768;
   std::string data = "e";
-  const ssize_t rc = os_sys_calls.write(pipe_fds[1], data.data(), data.size()).return_value_;
+  const ssize_t rc = os_sys_calls.write(fds[1], data.data(), data.size()).return_value_;
   ASSERT_GT(rc, 0);
   Api::IoCallUint64Result result = io_handle.read(buf, read_length);
   ASSERT_EQ(result.return_value_, static_cast<uint64_t>(rc));
-  ASSERT_EQ(os_sys_calls.close(pipe_fds[1]).return_value_, 0);
+  ASSERT_EQ(os_sys_calls.close(fds[1]).return_value_, 0);
   EXPECT_EQ("bbbbbe", buf.toString());
   OwnedImplTest::expectSlices({{6, 4090, 4096}}, buf);
 }

test/common/buffer/watermark_buffer_test.cc

@@ -243,21 +243,17 @@ TEST_F(WatermarkBufferTest, MoveOneByte) {
 }
 
 TEST_F(WatermarkBufferTest, WatermarkFdFunctions) {
-  os_fd_t pipe_fds[2] = {0, 0};
-#ifdef WIN32
+  os_fd_t fds[2] = {0, 0};
   auto& os_sys_calls = Api::OsSysCallsSingleton::get();
-  ASSERT_EQ(0, os_sys_calls.socketpair(AF_INET, SOCK_STREAM, 0, pipe_fds).return_value_);
-#else
-  ASSERT_EQ(0, pipe(pipe_fds));
-#endif
+  ASSERT_EQ(os_sys_calls.socketpair(ENVOY_DEFAULT_PIPE_TYPE, SOCK_STREAM, 0, fds).return_value_, 0);
 
   buffer_.add(TEN_BYTES, 10);
   buffer_.add(TEN_BYTES, 10);
   EXPECT_EQ(1, times_high_watermark_called_);
   EXPECT_EQ(0, times_low_watermark_called_);
 
   int bytes_written_total = 0;
-  Network::IoSocketHandleImpl io_handle1(pipe_fds[1]);
+  Network::IoSocketHandleImpl io_handle1(fds[1]);
   while (bytes_written_total < 20) {
     Api::IoCallUint64Result result = io_handle1.write(buffer_);
     if (!result.ok()) {
@@ -271,7 +267,7 @@ TEST_F(WatermarkBufferTest, WatermarkFdFunctions) {
   EXPECT_EQ(0, buffer_.length());
 
   int bytes_read_total = 0;
-  Network::IoSocketHandleImpl io_handle2(pipe_fds[0]);
+  Network::IoSocketHandleImpl io_handle2(fds[0]);
   while (bytes_read_total < 20) {
     Api::IoCallUint64Result result = io_handle2.read(buffer_, 20);
     bytes_read_total += result.return_value_;

test/common/network/connection_impl_test.cc

@@ -1255,13 +1255,19 @@ TEST_P(ConnectionImplTest, WriteWithWatermarks) {
         return {-1, SOCKET_ERROR_AGAIN};
       }));
 
-  EXPECT_CALL(os_sys_calls, writev(_, _, _))
-      .WillOnce(Invoke([&](os_fd_t, const iovec*, int) -> Api::SysCallSizeResult {
+  EXPECT_CALL(os_sys_calls, recv(_, _, _, _))
+      .WillRepeatedly(Invoke([&](os_fd_t, void*, size_t, int) -> Api::SysCallSizeResult {
+        return {-1, SOCKET_ERROR_AGAIN};
+      }));
+
+  EXPECT_CALL(os_sys_calls, send(_, _, _, _))
+      .WillOnce(Invoke([&](os_fd_t, void*, size_t, int) -> Api::SysCallSizeResult {
         dispatcher_->exit();
         // Return to default os_sys_calls implementation
         os_calls.reset();
         return {-1, SOCKET_ERROR_AGAIN};
       }));
+
   // The write() call on the connection will buffer enough data to bring the connection above the
   // high watermark and as the data will not flush it should not return below the watermark.
   EXPECT_CALL(client_callbacks_, onAboveWriteBufferHighWatermark());
@@ -1342,13 +1348,13 @@ TEST_P(ConnectionImplTest, WatermarkFuzzing) {
     // drain |bytes_to_flush| before having writev syscall fail with EAGAIN
     EXPECT_CALL(*client_write_buffer_, move(_))
         .WillOnce(Invoke(client_write_buffer_, &MockWatermarkBuffer::baseMove));
-    EXPECT_CALL(os_sys_calls, writev(_, _, _))
-        .WillOnce(Invoke([&](os_fd_t, const iovec*, int) -> Api::SysCallSizeResult {
+    EXPECT_CALL(os_sys_calls, send(_, _, _, _))
+        .WillOnce(Invoke([&](os_fd_t, void*, size_t, int) -> Api::SysCallSizeResult {
           client_write_buffer_->drain(bytes_to_flush);
           dispatcher_->exit();
           return {-1, SOCKET_ERROR_AGAIN};
         }))
-        .WillRepeatedly(Invoke([&](os_fd_t, const iovec*, int) -> Api::SysCallSizeResult {
+        .WillRepeatedly(Invoke([&](os_fd_t, void*, size_t, int) -> Api::SysCallSizeResult {
           return {-1, SOCKET_ERROR_AGAIN};
         }));