[lldb] Assorted improvements to the Pipe class (#128719)

The main motivation for this was the inconsistency in handling of
partial reads/writes between the windows and posix implementations
(windows was returning partial reads, posix was trying to fill the
buffer completely). I settle on the windows implementation, as that's
the more common behavior, and the "eager" version can be implemented on
top of that (in most cases, it isn't necessary, since we're writing just
a single byte).

Since this also required auditing the callers to make sure they're
handling partial reads/writes correctly, I used the opportunity to
modernize the function signatures as a forcing function. They now use
the `Timeout` class (basically an `optional<duration>`) to support both
polls (timeout=0) and blocking (timeout=nullopt) operations in a single
function, and use an `Expected` instead of a by-ref result to return the
number of bytes read/written.

As a drive-by, I also fix a problem with the windows implementation
where we were rounding the timeout value down, which meant that calls
could time out slightly sooner than expected.

Author: labath

lldb/include/lldb/Host/PipeBase.h

@@ -10,12 +10,11 @@
 #ifndef LLDB_HOST_PIPEBASE_H
 #define LLDB_HOST_PIPEBASE_H
 
-#include <chrono>
-#include <string>
-
 #include "lldb/Utility/Status.h"
+#include "lldb/Utility/Timeout.h"
 #include "llvm/ADT/SmallVector.h"
 #include "llvm/ADT/StringRef.h"
+#include "llvm/Support/Error.h"
 
 namespace lldb_private {
 class PipeBase {
@@ -32,10 +31,9 @@ class PipeBase {
   virtual Status OpenAsReader(llvm::StringRef name,
                               bool child_process_inherit) = 0;
 
-  Status OpenAsWriter(llvm::StringRef name, bool child_process_inherit);
-  virtual Status
-  OpenAsWriterWithTimeout(llvm::StringRef name, bool child_process_inherit,
-                          const std::chrono::microseconds &timeout) = 0;
+  virtual llvm::Error OpenAsWriter(llvm::StringRef name,
+                                   bool child_process_inherit,
+                                   const Timeout<std::micro> &timeout) = 0;
 
   virtual bool CanRead() const = 0;
   virtual bool CanWrite() const = 0;
@@ -56,14 +54,13 @@ class PipeBase {
   // Delete named pipe.
   virtual Status Delete(llvm::StringRef name) = 0;
 
-  virtual Status WriteWithTimeout(const void *buf, size_t size,
-                                  const std::chrono::microseconds &timeout,
-                                  size_t &bytes_written) = 0;
-  Status Write(const void *buf, size_t size, size_t &bytes_written);
-  virtual Status ReadWithTimeout(void *buf, size_t size,
-                                 const std::chrono::microseconds &timeout,
-                                 size_t &bytes_read) = 0;
-  Status Read(void *buf, size_t size, size_t &bytes_read);
+  virtual llvm::Expected<size_t>
+  Write(const void *buf, size_t size,
+        const Timeout<std::micro> &timeout = std::nullopt) = 0;
+
+  virtual llvm::Expected<size_t>
+  Read(void *buf, size_t size,
+       const Timeout<std::micro> &timeout = std::nullopt) = 0;
 };
 }
 

lldb/include/lldb/Host/posix/PipePosix.h

@@ -8,6 +8,7 @@
 
 #ifndef LLDB_HOST_POSIX_PIPEPOSIX_H
 #define LLDB_HOST_POSIX_PIPEPOSIX_H
+
 #include "lldb/Host/PipeBase.h"
 #include <mutex>
 
@@ -38,9 +39,8 @@ class PipePosix : public PipeBase {
                               llvm::SmallVectorImpl<char> &name) override;
   Status OpenAsReader(llvm::StringRef name,
                       bool child_process_inherit) override;
-  Status
-  OpenAsWriterWithTimeout(llvm::StringRef name, bool child_process_inherit,
-                          const std::chrono::microseconds &timeout) override;
+  llvm::Error OpenAsWriter(llvm::StringRef name, bool child_process_inherit,
+                           const Timeout<std::micro> &timeout) override;
 
   bool CanRead() const override;
   bool CanWrite() const override;
@@ -64,12 +64,13 @@ class PipePosix : public PipeBase {
 
   Status Delete(llvm::StringRef name) override;
 
-  Status WriteWithTimeout(const void *buf, size_t size,
-                          const std::chrono::microseconds &timeout,
-                          size_t &bytes_written) override;
-  Status ReadWithTimeout(void *buf, size_t size,
-                         const std::chrono::microseconds &timeout,
-                         size_t &bytes_read) override;
+  llvm::Expected<size_t>
+  Write(const void *buf, size_t size,
+        const Timeout<std::micro> &timeout = std::nullopt) override;
+
+  llvm::Expected<size_t>
+  Read(void *buf, size_t size,
+       const Timeout<std::micro> &timeout = std::nullopt) override;
 
 private:
   bool CanReadUnlocked() const;

lldb/include/lldb/Host/windows/PipeWindows.h

@@ -38,9 +38,8 @@ class PipeWindows : public PipeBase {
                               llvm::SmallVectorImpl<char> &name) override;
   Status OpenAsReader(llvm::StringRef name,
                       bool child_process_inherit) override;
-  Status
-  OpenAsWriterWithTimeout(llvm::StringRef name, bool child_process_inherit,
-                          const std::chrono::microseconds &timeout) override;
+  llvm::Error OpenAsWriter(llvm::StringRef name, bool child_process_inherit,
+                           const Timeout<std::micro> &timeout) override;
 
   bool CanRead() const override;
   bool CanWrite() const override;
@@ -59,12 +58,13 @@ class PipeWindows : public PipeBase {
 
   Status Delete(llvm::StringRef name) override;
 
-  Status WriteWithTimeout(const void *buf, size_t size,
-                          const std::chrono::microseconds &timeout,
-                          size_t &bytes_written) override;
-  Status ReadWithTimeout(void *buf, size_t size,
-                         const std::chrono::microseconds &timeout,
-                         size_t &bytes_read) override;
+  llvm::Expected<size_t>
+  Write(const void *buf, size_t size,
+        const Timeout<std::micro> &timeout = std::nullopt) override;
+
+  llvm::Expected<size_t>
+  Read(void *buf, size_t size,
+       const Timeout<std::micro> &timeout = std::nullopt) override;
 
   // PipeWindows specific methods.  These allow access to the underlying OS
   // handle.

lldb/source/Host/common/PipeBase.cpp

@@ -11,19 +11,3 @@
 using namespace lldb_private;
 
 PipeBase::~PipeBase() = default;
-
-Status PipeBase::OpenAsWriter(llvm::StringRef name,
-                              bool child_process_inherit) {
-  return OpenAsWriterWithTimeout(name, child_process_inherit,
-                                 std::chrono::microseconds::zero());
-}
-
-Status PipeBase::Write(const void *buf, size_t size, size_t &bytes_written) {
-  return WriteWithTimeout(buf, size, std::chrono::microseconds::zero(),
-                          bytes_written);
-}
-
-Status PipeBase::Read(void *buf, size_t size, size_t &bytes_read) {
-  return ReadWithTimeout(buf, size, std::chrono::microseconds::zero(),
-                         bytes_read);
-}

lldb/source/Host/common/Socket.cpp

@@ -93,15 +93,14 @@ Status SharedSocket::CompleteSending(lldb::pid_t child_pid) {
         "WSADuplicateSocket() failed, error: %d", last_error);
   }
 
-  size_t num_bytes;
-  Status error =
-      m_socket_pipe.WriteWithTimeout(&protocol_info, sizeof(protocol_info),
-                                     std::chrono::seconds(10), num_bytes);
-  if (error.Fail())
-    return error;
-  if (num_bytes != sizeof(protocol_info))
+  llvm::Expected<size_t> num_bytes = m_socket_pipe.Write(
+      &protocol_info, sizeof(protocol_info), std::chrono::seconds(10));
+  if (!num_bytes)
+    return Status::FromError(num_bytes.takeError());
+  if (*num_bytes != sizeof(protocol_info))
     return Status::FromErrorStringWithFormatv(
-        "WriteWithTimeout(WSAPROTOCOL_INFO) failed: {0} bytes", num_bytes);
+        "Write(WSAPROTOCOL_INFO) failed: wrote {0}/{1} bytes", *num_bytes,
+        sizeof(protocol_info));
 #endif
   return Status();
 }
@@ -113,16 +112,14 @@ Status SharedSocket::GetNativeSocket(shared_fd_t fd, NativeSocket &socket) {
   WSAPROTOCOL_INFO protocol_info;
   {
     Pipe socket_pipe(fd, LLDB_INVALID_PIPE);
-    size_t num_bytes;
-    Status error =
-        socket_pipe.ReadWithTimeout(&protocol_info, sizeof(protocol_info),
-                                    std::chrono::seconds(10), num_bytes);
-    if (error.Fail())
-      return error;
-    if (num_bytes != sizeof(protocol_info)) {
+    llvm::Expected<size_t> num_bytes = socket_pipe.Read(
+        &protocol_info, sizeof(protocol_info), std::chrono::seconds(10));
+    if (!num_bytes)
+      return Status::FromError(num_bytes.takeError());
+    if (*num_bytes != sizeof(protocol_info)) {
       return Status::FromErrorStringWithFormatv(
-          "socket_pipe.ReadWithTimeout(WSAPROTOCOL_INFO) failed: {0} bytes",
-          num_bytes);
+          "Read(WSAPROTOCOL_INFO) failed: read {0}/{1} bytes", *num_bytes,
+          sizeof(protocol_info));
     }
   }
   socket = ::WSASocket(FROM_PROTOCOL_INFO, FROM_PROTOCOL_INFO,

lldb/source/Host/posix/ConnectionFileDescriptorPosix.cpp

@@ -178,9 +178,7 @@ ConnectionFileDescriptor::Connect(llvm::StringRef path,
 }
 
 bool ConnectionFileDescriptor::InterruptRead() {
-  size_t bytes_written = 0;
-  Status result = m_pipe.Write("i", 1, bytes_written);
-  return result.Success();
+  return !errorToBool(m_pipe.Write("i", 1).takeError());
 }
 
 ConnectionStatus ConnectionFileDescriptor::Disconnect(Status *error_ptr) {
@@ -205,13 +203,11 @@ ConnectionStatus ConnectionFileDescriptor::Disconnect(Status *error_ptr) {
   std::unique_lock<std::recursive_mutex> locker(m_mutex, std::defer_lock);
   if (!locker.try_lock()) {
     if (m_pipe.CanWrite()) {
-      size_t bytes_written = 0;
-      Status result = m_pipe.Write("q", 1, bytes_written);
-      LLDB_LOGF(log,
-                "%p ConnectionFileDescriptor::Disconnect(): Couldn't get "
-                "the lock, sent 'q' to %d, error = '%s'.",
-                static_cast<void *>(this), m_pipe.GetWriteFileDescriptor(),
-                result.AsCString());
+      llvm::Error err = m_pipe.Write("q", 1).takeError();
+      LLDB_LOG(log,
+               "{0}: Couldn't get the lock, sent 'q' to {1}, error = '{2}'.",
+               this, m_pipe.GetWriteFileDescriptor(), err);
+      consumeError(std::move(err));
     } else if (log) {
       LLDB_LOGF(log,
                 "%p ConnectionFileDescriptor::Disconnect(): Couldn't get the "

lldb/source/Host/posix/MainLoopPosix.cpp

@@ -392,9 +392,5 @@ void MainLoopPosix::Interrupt() {
     return;
 
   char c = '.';
-  size_t bytes_written;
-  Status error = m_interrupt_pipe.Write(&c, 1, bytes_written);
-  assert(error.Success());
-  UNUSED_IF_ASSERT_DISABLED(error);
-  assert(bytes_written == 1);
+  cantFail(m_interrupt_pipe.Write(&c, 1));
 }

lldb/source/Host/posix/PipePosix.cpp

@@ -12,7 +12,9 @@
 #include "lldb/Utility/SelectHelper.h"
 #include "llvm/ADT/SmallString.h"
 #include "llvm/Support/Errno.h"
+#include "llvm/Support/Error.h"
 #include <functional>
+#include <system_error>
 #include <thread>
 
 #include <cerrno>
@@ -164,26 +166,27 @@ Status PipePosix::OpenAsReader(llvm::StringRef name,
   return error;
 }
 
-Status
-PipePosix::OpenAsWriterWithTimeout(llvm::StringRef name,
-                                   bool child_process_inherit,
-                                   const std::chrono::microseconds &timeout) {
+llvm::Error PipePosix::OpenAsWriter(llvm::StringRef name,
+                                    bool child_process_inherit,
+                                    const Timeout<std::micro> &timeout) {
   std::lock_guard<std::mutex> guard(m_write_mutex);
   if (CanReadUnlocked() || CanWriteUnlocked())
-    return Status::FromErrorString("Pipe is already opened");
+    return llvm::createStringError("Pipe is already opened");
 
   int flags = O_WRONLY | O_NONBLOCK;
   if (!child_process_inherit)
     flags |= O_CLOEXEC;
 
   using namespace std::chrono;
-  const auto finish_time = Now() + timeout;
+  std::optional<time_point<steady_clock>> finish_time;
+  if (timeout)
+    finish_time = Now() + *timeout;
 
   while (!CanWriteUnlocked()) {
-    if (timeout != microseconds::zero()) {
-      const auto dur = duration_cast<microseconds>(finish_time - Now()).count();
-      if (dur <= 0)
-        return Status::FromErrorString(
+    if (timeout) {
+      if (Now() > finish_time)
+        return llvm::createStringError(
+            std::make_error_code(std::errc::timed_out),
             "timeout exceeded - reader hasn't opened so far");
     }
 
@@ -193,7 +196,8 @@ PipePosix::OpenAsWriterWithTimeout(llvm::StringRef name,
       const auto errno_copy = errno;
       // We may get ENXIO if a reader side of the pipe hasn't opened yet.
       if (errno_copy != ENXIO && errno_copy != EINTR)
-        return Status(errno_copy, eErrorTypePOSIX);
+        return llvm::errorCodeToError(
+            std::error_code(errno_copy, std::generic_category()));
 
       std::this_thread::sleep_for(
           milliseconds(OPEN_WRITER_SLEEP_TIMEOUT_MSECS));
@@ -202,7 +206,7 @@ PipePosix::OpenAsWriterWithTimeout(llvm::StringRef name,
     }
   }
 
-  return Status();
+  return llvm::Error::success();
 }
 
 int PipePosix::GetReadFileDescriptor() const {
@@ -300,70 +304,51 @@ void PipePosix::CloseWriteFileDescriptorUnlocked() {
   }
 }
 
-Status PipePosix::ReadWithTimeout(void *buf, size_t size,
-                                  const std::chrono::microseconds &timeout,
-                                  size_t &bytes_read) {
+llvm::Expected<size_t> PipePosix::Read(void *buf, size_t size,
+                                       const Timeout<std::micro> &timeout) {
   std::lock_guard<std::mutex> guard(m_read_mutex);
-  bytes_read = 0;
   if (!CanReadUnlocked())
-    return Status(EINVAL, eErrorTypePOSIX);
+    return llvm::errorCodeToError(
+        std::make_error_code(std::errc::invalid_argument));
 
   const int fd = GetReadFileDescriptorUnlocked();
 
   SelectHelper select_helper;
-  select_helper.SetTimeout(timeout);
+  if (timeout)
+    select_helper.SetTimeout(*timeout);
   select_helper.FDSetRead(fd);
 
-  Status error;
-  while (error.Success()) {
-    error = select_helper.Select();
-    if (error.Success()) {
-      auto result =
-          ::read(fd, static_cast<char *>(buf) + bytes_read, size - bytes_read);
-      if (result != -1) {
-        bytes_read += result;
-        if (bytes_read == size || result == 0)
-          break;
-      } else if (errno == EINTR) {
-        continue;
-      } else {
-        error = Status::FromErrno();
-        break;
-      }
-    }
-  }
-  return error;
+  if (llvm::Error error = select_helper.Select().takeError())
+    return error;
+
+  ssize_t result = ::read(fd, buf, size);
+  if (result == -1)
+    return llvm::errorCodeToError(
+        std::error_code(errno, std::generic_category()));
+
+  return result;
 }
 
-Status PipePosix::WriteWithTimeout(const void *buf, size_t size,
-                                   const std::chrono::microseconds &timeout,
-                                   size_t &bytes_written) {
+llvm::Expected<size_t> PipePosix::Write(const void *buf, size_t size,
+                                        const Timeout<std::micro> &timeout) {
   std::lock_guard<std::mutex> guard(m_write_mutex);
-  bytes_written = 0;
   if (!CanWriteUnlocked())
-    return Status(EINVAL, eErrorTypePOSIX);
+    return llvm::errorCodeToError(
+        std::make_error_code(std::errc::invalid_argument));
 
   const int fd = GetWriteFileDescriptorUnlocked();
   SelectHelper select_helper;
-  select_helper.SetTimeout(timeout);
+  if (timeout)
+    select_helper.SetTimeout(*timeout);
   select_helper.FDSetWrite(fd);
 
-  Status error;
-  while (error.Success()) {
-    error = select_helper.Select();
-    if (error.Success()) {
-      auto result = ::write(fd, static_cast<const char *>(buf) + bytes_written,
-                            size - bytes_written);
-      if (result != -1) {
-        bytes_written += result;
-        if (bytes_written == size)
-          break;
-      } else if (errno == EINTR) {
-        continue;
-      } else {
-        error = Status::FromErrno();
-      }
-    }
-  }
-  return error;
+  if (llvm::Error error = select_helper.Select().takeError())
+    return error;
+
+  ssize_t result = ::write(fd, buf, size);
+  if (result == -1)
+    return llvm::errorCodeToError(
+        std::error_code(errno, std::generic_category()));
+
+  return result;
 }