[lldb] Added Pipe::WriteWithTimeout()

Fixed few bugs in PipeWindows. Added the test for async read/write.

Author: slydiman

lldb/include/lldb/Host/PipeBase.h

@@ -56,7 +56,10 @@ class PipeBase {
   // Delete named pipe.
   virtual Status Delete(llvm::StringRef name) = 0;
 
-  virtual Status Write(const void *buf, size_t size, size_t &bytes_written) = 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;

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

@@ -64,7 +64,9 @@ class PipePosix : public PipeBase {
 
   Status Delete(llvm::StringRef name) override;
 
-  Status Write(const void *buf, size_t size, size_t &bytes_written) 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;

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

@@ -32,7 +32,6 @@ class PipeWindows : public PipeBase {
   Status CreateNew(bool child_process_inherit) override;
 
   // Create a named pipe.
-  Status CreateNewNamed(bool child_process_inherit);
   Status CreateNew(llvm::StringRef name, bool child_process_inherit) override;
   Status CreateWithUniqueName(llvm::StringRef prefix,
                               bool child_process_inherit,
@@ -60,7 +59,9 @@ class PipeWindows : public PipeBase {
 
   Status Delete(llvm::StringRef name) override;
 
-  Status Write(const void *buf, size_t size, size_t &bytes_written) 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;

lldb/source/Host/common/PipeBase.cpp

@@ -18,6 +18,11 @@ Status PipeBase::OpenAsWriter(llvm::StringRef name,
                                  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/posix/PipePosix.cpp

@@ -335,15 +335,17 @@ Status PipePosix::ReadWithTimeout(void *buf, size_t size,
   return error;
 }
 
-Status PipePosix::Write(const void *buf, size_t size, size_t &bytes_written) {
+Status PipePosix::WriteWithTimeout(const void *buf, size_t size,
+                                   const std::chrono::microseconds &timeout,
+                                   size_t &bytes_written) {
   std::lock_guard<std::mutex> guard(m_write_mutex);
   bytes_written = 0;
   if (!CanWriteUnlocked())
     return Status(EINVAL, eErrorTypePOSIX);
 
   const int fd = GetWriteFileDescriptorUnlocked();
   SelectHelper select_helper;
-  select_helper.SetTimeout(std::chrono::seconds(0));
+  select_helper.SetTimeout(timeout);
   select_helper.FDSetWrite(fd);
 
   Status error;

lldb/source/Host/windows/PipeWindows.cpp

@@ -58,30 +58,15 @@ PipeWindows::PipeWindows(pipe_t read, pipe_t write)
   }
 
   ZeroMemory(&m_read_overlapped, sizeof(m_read_overlapped));
+  m_read_overlapped.hEvent = ::CreateEventA(nullptr, TRUE, FALSE, nullptr);
+
   ZeroMemory(&m_write_overlapped, sizeof(m_write_overlapped));
+  m_write_overlapped.hEvent = ::CreateEventA(nullptr, TRUE, FALSE, nullptr);
 }
 
 PipeWindows::~PipeWindows() { Close(); }
 
 Status PipeWindows::CreateNew(bool child_process_inherit) {
-  // Create an anonymous pipe with the specified inheritance.
-  SECURITY_ATTRIBUTES sa{sizeof(SECURITY_ATTRIBUTES), 0,
-                         child_process_inherit ? TRUE : FALSE};
-  BOOL result = ::CreatePipe(&m_read, &m_write, &sa, 1024);
-  if (result == FALSE)
-    return Status(::GetLastError(), eErrorTypeWin32);
-
-  m_read_fd = _open_osfhandle((intptr_t)m_read, _O_RDONLY);
-  ZeroMemory(&m_read_overlapped, sizeof(m_read_overlapped));
-  m_read_overlapped.hEvent = ::CreateEventA(nullptr, TRUE, FALSE, nullptr);
-
-  m_write_fd = _open_osfhandle((intptr_t)m_write, _O_WRONLY);
-  ZeroMemory(&m_write_overlapped, sizeof(m_write_overlapped));
-
-  return Status();
-}
-
-Status PipeWindows::CreateNewNamed(bool child_process_inherit) {
   // Even for anonymous pipes, we open a named pipe.  This is because you
   // cannot get overlapped i/o on Windows without using a named pipe.  So we
   // synthesize a unique name.
@@ -105,12 +90,19 @@ Status PipeWindows::CreateNew(llvm::StringRef name,
   std::string pipe_path = g_pipe_name_prefix.str();
   pipe_path.append(name.str());
 
+  SECURITY_ATTRIBUTES sa{sizeof(SECURITY_ATTRIBUTES), 0,
+                         child_process_inherit ? TRUE : FALSE};
+
   // Always open for overlapped i/o.  We implement blocking manually in Read
   // and Write.
   DWORD read_mode = FILE_FLAG_OVERLAPPED;
-  m_read = ::CreateNamedPipeA(
-      pipe_path.c_str(), PIPE_ACCESS_INBOUND | read_mode,
-      PIPE_TYPE_BYTE | PIPE_WAIT, 1, 1024, 1024, 120 * 1000, NULL);
+  m_read =
+      ::CreateNamedPipeA(pipe_path.c_str(), PIPE_ACCESS_INBOUND | read_mode,
+                         PIPE_TYPE_BYTE | PIPE_WAIT, 1,
+                         1024, // Out buffer size
+                         1024, // In buffer size
+                         0,    // Default timeout in ms, 0 means 50ms
+                         &sa);
   if (INVALID_HANDLE_VALUE == m_read)
     return Status(::GetLastError(), eErrorTypeWin32);
   m_read_fd = _open_osfhandle((intptr_t)m_read, _O_RDONLY);
@@ -155,7 +147,7 @@ Status PipeWindows::CreateWithUniqueName(llvm::StringRef prefix,
 Status PipeWindows::OpenAsReader(llvm::StringRef name,
                                  bool child_process_inherit) {
   if (CanRead())
-    return Status(ERROR_ALREADY_EXISTS, eErrorTypeWin32);
+    return Status(); // Note the name is ignored.
 
   return OpenNamedPipe(name, child_process_inherit, true);
 }
@@ -165,7 +157,7 @@ PipeWindows::OpenAsWriterWithTimeout(llvm::StringRef name,
                                      bool child_process_inherit,
                                      const std::chrono::microseconds &timeout) {
   if (CanWrite())
-    return Status(ERROR_ALREADY_EXISTS, eErrorTypeWin32);
+    return Status(); // Note the name is ignored.
 
   return OpenNamedPipe(name, child_process_inherit, false);
 }
@@ -177,8 +169,8 @@ Status PipeWindows::OpenNamedPipe(llvm::StringRef name,
 
   assert(is_read ? !CanRead() : !CanWrite());
 
-  SECURITY_ATTRIBUTES attributes = {};
-  attributes.bInheritHandle = child_process_inherit;
+  SECURITY_ATTRIBUTES attributes{sizeof(SECURITY_ATTRIBUTES), 0,
+                                 child_process_inherit ? TRUE : FALSE};
 
   std::string pipe_path = g_pipe_name_prefix.str();
   pipe_path.append(name.str());
@@ -202,6 +194,7 @@ Status PipeWindows::OpenNamedPipe(llvm::StringRef name,
     m_write_fd = _open_osfhandle((intptr_t)m_write, _O_WRONLY);
 
     ZeroMemory(&m_write_overlapped, sizeof(m_write_overlapped));
+    m_write_overlapped.hEvent = ::CreateEventA(nullptr, TRUE, FALSE, nullptr);
   }
 
   return Status();
@@ -228,6 +221,8 @@ int PipeWindows::ReleaseWriteFileDescriptor() {
     return PipeWindows::kInvalidDescriptor;
   int result = m_write_fd;
   m_write_fd = PipeWindows::kInvalidDescriptor;
+  if (m_write_overlapped.hEvent)
+    ::CloseHandle(m_write_overlapped.hEvent);
   m_write = INVALID_HANDLE_VALUE;
   ZeroMemory(&m_write_overlapped, sizeof(m_write_overlapped));
   return result;
@@ -250,6 +245,9 @@ void PipeWindows::CloseWriteFileDescriptor() {
   if (!CanWrite())
     return;
 
+  if (m_write_overlapped.hEvent)
+    ::CloseHandle(m_write_overlapped.hEvent);
+
   _close(m_write_fd);
   m_write = INVALID_HANDLE_VALUE;
   m_write_fd = PipeWindows::kInvalidDescriptor;
@@ -280,15 +278,21 @@ Status PipeWindows::ReadWithTimeout(void *buf, size_t size,
     return Status(ERROR_INVALID_HANDLE, eErrorTypeWin32);
 
   bytes_read = 0;
-  DWORD sys_bytes_read = size;
-  BOOL result = ::ReadFile(m_read, buf, sys_bytes_read, &sys_bytes_read,
-                           &m_read_overlapped);
-  if (!result && GetLastError() != ERROR_IO_PENDING)
-    return Status(::GetLastError(), eErrorTypeWin32);
+  DWORD sys_bytes_read = 0;
+  BOOL result =
+      ::ReadFile(m_read, buf, size, &sys_bytes_read, &m_read_overlapped);
+  if (result) {
+    bytes_read = sys_bytes_read;
+    return Status();
+  }
+
+  DWORD failure_error = ::GetLastError();
+  if (failure_error != ERROR_IO_PENDING)
+    return Status(failure_error, eErrorTypeWin32);
 
   DWORD timeout = (duration == std::chrono::microseconds::zero())
                       ? INFINITE
-                      : duration.count() * 1000;
+                      : duration.count() / 1000;
   DWORD wait_result = ::WaitForSingleObject(m_read_overlapped.hEvent, timeout);
   if (wait_result != WAIT_OBJECT_0) {
     // The operation probably failed.  However, if it timed out, we need to
@@ -298,10 +302,10 @@ Status PipeWindows::ReadWithTimeout(void *buf, size_t size,
     // happens, the original operation should be considered to have been
     // successful.
     bool failed = true;
-    DWORD failure_error = ::GetLastError();
+    failure_error = ::GetLastError();
     if (wait_result == WAIT_TIMEOUT) {
-      BOOL cancel_result = CancelIoEx(m_read, &m_read_overlapped);
-      if (!cancel_result && GetLastError() == ERROR_NOT_FOUND)
+      BOOL cancel_result = ::CancelIoEx(m_read, &m_read_overlapped);
+      if (!cancel_result && ::GetLastError() == ERROR_NOT_FOUND)
         failed = false;
     }
     if (failed)
@@ -310,27 +314,61 @@ Status PipeWindows::ReadWithTimeout(void *buf, size_t size,
 
   // Now we call GetOverlappedResult setting bWait to false, since we've
   // already waited as long as we're willing to.
-  if (!GetOverlappedResult(m_read, &m_read_overlapped, &sys_bytes_read, FALSE))
+  if (!::GetOverlappedResult(m_read, &m_read_overlapped, &sys_bytes_read,
+                             FALSE))
     return Status(::GetLastError(), eErrorTypeWin32);
 
   bytes_read = sys_bytes_read;
   return Status();
 }
 
-Status PipeWindows::Write(const void *buf, size_t num_bytes,
-                          size_t &bytes_written) {
+Status PipeWindows::WriteWithTimeout(const void *buf, size_t size,
+                                     const std::chrono::microseconds &duration,
+                                     size_t &bytes_written) {
   if (!CanWrite())
     return Status(ERROR_INVALID_HANDLE, eErrorTypeWin32);
 
-  DWORD sys_bytes_written = 0;
-  BOOL write_result = ::WriteFile(m_write, buf, num_bytes, &sys_bytes_written,
-                                  &m_write_overlapped);
-  if (!write_result && GetLastError() != ERROR_IO_PENDING)
-    return Status(::GetLastError(), eErrorTypeWin32);
+  bytes_written = 0;
+  DWORD sys_bytes_write = 0;
+  BOOL result =
+      ::WriteFile(m_write, buf, size, &sys_bytes_write, &m_write_overlapped);
+  if (result) {
+    bytes_written = sys_bytes_write;
+    return Status();
+  }
+
+  DWORD failure_error = ::GetLastError();
+  if (failure_error != ERROR_IO_PENDING)
+    return Status(failure_error, eErrorTypeWin32);
 
-  BOOL result = GetOverlappedResult(m_write, &m_write_overlapped,
-                                    &sys_bytes_written, TRUE);
-  if (!result)
+  DWORD timeout = (duration == std::chrono::microseconds::zero())
+                      ? INFINITE
+                      : duration.count() / 1000;
+  DWORD wait_result = ::WaitForSingleObject(m_write_overlapped.hEvent, timeout);
+  if (wait_result != WAIT_OBJECT_0) {
+    // The operation probably failed.  However, if it timed out, we need to
+    // cancel the I/O. Between the time we returned from WaitForSingleObject
+    // and the time we call CancelIoEx, the operation may complete.  If that
+    // hapens, CancelIoEx will fail and return ERROR_NOT_FOUND. If that
+    // happens, the original operation should be considered to have been
+    // successful.
+    bool failed = true;
+    failure_error = ::GetLastError();
+    if (wait_result == WAIT_TIMEOUT) {
+      BOOL cancel_result = ::CancelIoEx(m_write, &m_write_overlapped);
+      if (!cancel_result && ::GetLastError() == ERROR_NOT_FOUND)
+        failed = false;
+    }
+    if (failed)
+      return Status(failure_error, eErrorTypeWin32);
+  }
+
+  // Now we call GetOverlappedResult setting bWait to false, since we've
+  // already waited as long as we're willing to.
+  if (!::GetOverlappedResult(m_write, &m_write_overlapped, &sys_bytes_write,
+                             FALSE))
     return Status(::GetLastError(), eErrorTypeWin32);
+
+  bytes_written = sys_bytes_write;
   return Status();
 }

lldb/unittests/Host/PipeTest.cpp

@@ -27,10 +27,10 @@ TEST_F(PipeTest, CreateWithUniqueName) {
                                               name)
                         .ToError(),
                     llvm::Succeeded());
+  pipe.Close();
+  ASSERT_THAT_ERROR(pipe.Delete(name).ToError(), llvm::Succeeded());
 }
 
-// Test broken
-#ifndef _WIN32
 TEST_F(PipeTest, OpenAsReader) {
   Pipe pipe;
   llvm::SmallString<0> name;
@@ -44,8 +44,72 @@ TEST_F(PipeTest, OpenAsReader) {
   size_t name_len = name.size();
   name += "foobar";
   llvm::StringRef name_ref(name.data(), name_len);
+  // Note OpenAsReader() do nothing on Windows, the pipe is already opened for
+  // read and write.
   ASSERT_THAT_ERROR(
       pipe.OpenAsReader(name_ref, /*child_process_inherit=*/false).ToError(),
       llvm::Succeeded());
+
+  ASSERT_TRUE(pipe.CanRead());
+  pipe.Close();
+  ASSERT_THAT_ERROR(pipe.Delete(name_ref).ToError(), llvm::Succeeded());
+}
+
+TEST_F(PipeTest, WriteWithTimeout) {
+  Pipe pipe;
+  ASSERT_THAT_ERROR(pipe.CreateNew(false).ToError(), llvm::Succeeded());
+  // Note write_chunk_size must be less than the pipe buffer.
+  // The pipe buffer is 1024 for PipeWindows and 4096 for PipePosix.
+  const size_t buf_size = 8192;
+  const size_t write_chunk_size = 256;
+  const size_t read_chunk_size = 300;
+  std::unique_ptr<int32_t[]> write_buf_ptr(
+      new int32_t[buf_size / sizeof(int32_t)]);
+  int32_t *write_buf = write_buf_ptr.get();
+  std::unique_ptr<int32_t[]> read_buf_ptr(
+      new int32_t[(buf_size + 100) / sizeof(int32_t)]);
+  int32_t *read_buf = read_buf_ptr.get();
+  for (int i = 0; i < buf_size / sizeof(int32_t); ++i) {
+    write_buf[i] = i;
+    read_buf[i] = -i;
+  }
+
+  char *write_ptr = (char *)write_buf;
+  size_t write_bytes = 0;
+  char *read_ptr = (char *)read_buf;
+  size_t read_bytes = 0;
+  size_t num_bytes = 0;
+  Status error;
+  while (write_bytes < buf_size) {
+    error = pipe.WriteWithTimeout(write_ptr + write_bytes, write_chunk_size,
+                                  std::chrono::milliseconds(10), num_bytes);
+    if (error.Fail()) {
+      ASSERT_TRUE(read_bytes < buf_size);
+      error = pipe.ReadWithTimeout(read_ptr + read_bytes, read_chunk_size,
+                                   std::chrono::milliseconds(10), num_bytes);
+      if (error.Fail())
+        FAIL();
+      else
+        read_bytes += num_bytes;
+    } else
+      write_bytes += num_bytes;
+  }
+  // Read the rest data.
+  while (read_bytes < buf_size) {
+    error = pipe.ReadWithTimeout(read_ptr + read_bytes, buf_size - read_bytes,
+                                 std::chrono::milliseconds(10), num_bytes);
+    if (error.Fail())
+      FAIL();
+    else
+      read_bytes += num_bytes;
+  }
+
+  // Be sure the pipe is empty.
+  error = pipe.ReadWithTimeout(read_ptr + read_bytes, 100,
+                               std::chrono::milliseconds(10), num_bytes);
+  ASSERT_TRUE(error.Fail());
+
+  // Compare the data.
+  ASSERT_EQ(write_bytes, read_bytes);
+  ASSERT_EQ(memcmp(write_buf, read_buf, buf_size), 0);
 }
-#endif