Add Pipe::WriteWithTimeout method

[labath]

Summary

This commit adds a WriteWithTimeout method to time Pipe class, analogous to the existing
ReadWithTimeout(). It also changes the meaning of passing zero as a timeout value. Previously,
zero was used as an infinite timeout value. Now, the meaning of zero timeout to return the data
avaiable without sleeping (basically, a non-blocking operation). This makes the behaviour of Pipe
consistent with the Communication/Connection classes. For blocking operatios with infinite
timeout, I introduce a special constant for this purpose.

[zturner]

windows stuff lgtm

[ovyalov]

LGTM

lldb/trunk/include/lldb/Host/PipeBase.h

public:
     virtual void CloseWriteFileDescriptor() = 0;
 
     // Close both descriptors
     virtual void Close() = 0;
 
     // Delete named pipe.
     virtual Error Delete(llvm::StringRef name) = 0;
 
-    virtual Error Write(const void *buf, size_t size, size_t &bytes_written) = 0;
+    virtual Error WriteWithTimeout(const void *buf, size_t size, const std::chrono::microseconds &timeout, size_t &bytes_written) = 0;
     virtual Error ReadWithTimeout(void *buf, size_t size, const std::chrono::microseconds &timeout, size_t &bytes_read) = 0;
+
     Error Read(void *buf, size_t size, size_t &bytes_read);
+    Error Write(const void *buf, size_t size, size_t &bytes_written);
+
+    static const std::chrono::microseconds kInfiniteTimeout;
 };
 }
 
 #endif

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

public:
     // Close both descriptors
     void
     Close() override;
 
     Error
     Delete(llvm::StringRef name) override;
 
     Error
-    Write(const void *buf, size_t size, size_t &bytes_written) override;
+    WriteWithTimeout(const void *buf, size_t size, const std::chrono::microseconds &timeout, size_t &bytes_written) override;
     Error
     ReadWithTimeout(void *buf, size_t size, const std::chrono::microseconds &timeout, size_t &bytes_read) override;
 
 private:
     int m_fds[2];
 };
 
 } // namespace lldb_private
 
 #endif  // #if defined(__cplusplus)
 #endif  // liblldb_Host_posix_PipePosix_h_

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

public:
     int ReleaseWriteFileDescriptor() override;
     void CloseReadFileDescriptor() override;
     void CloseWriteFileDescriptor() override;
 
     void Close() override;
 
     Error Delete(llvm::StringRef name) override;
 
-    Error Write(const void *buf, size_t size, size_t &bytes_written) override;
+    Error WriteWithTimeout(const void *buf, size_t size, const std::chrono::microseconds &timeout, size_t &bytes_written) override;
     Error ReadWithTimeout(void *buf, size_t size, const std::chrono::microseconds &timeout, size_t &bytes_read) override;
 
     // PipeWindows specific methods.  These allow access to the underlying OS handle.
     HANDLE GetReadNativeHandle();
     HANDLE GetWriteNativeHandle();
 
   private:
     Error OpenNamedPipe(llvm::StringRef name, bool child_process_inherit, bool is_read);

lldb/trunk/source/Host/common/PipeBase.cpp

 //===-- source/Host/common/PipeBase.cpp -------------------------*- C++ -*-===//
 //
 //                     The LLVM Compiler Infrastructure
 //
 // This file is distributed under the University of Illinois Open Source
 // License. See LICENSE.TXT for details.
 //
 //===----------------------------------------------------------------------===//
 
 #include "lldb/Host/PipeBase.h"
 
 using namespace lldb_private;
 
+const std::chrono::microseconds PipeBase::kInfiniteTimeout(-1);
 
 PipeBase::~PipeBase() = default;
 
 Error
 PipeBase::OpenAsWriter(llvm::StringRef name, bool child_process_inherit)
 {
-    return OpenAsWriterWithTimeout(name, child_process_inherit, std::chrono::microseconds::zero());
+    return OpenAsWriterWithTimeout(name, child_process_inherit, kInfiniteTimeout);
 }
 
 Error
 PipeBase::Read(void *buf, size_t size, size_t &bytes_read)
 {
-    return ReadWithTimeout(buf, size, std::chrono::microseconds::zero(), bytes_read);
+    return ReadWithTimeout(buf, size, kInfiniteTimeout, bytes_read);
+}
+
+Error
+PipeBase::Write(const void *buf, size_t size, size_t &bytes_written)
+{
+    return WriteWithTimeout(buf, size, kInfiniteTimeout, bytes_written);
 }

lldb/trunk/source/Host/posix/PipePosix.cpp

SelectIO(int handle, bool is_read, const std::function<Error(bool&)> &io_handler, const std::chrono::microseconds &timeout)
 
     using namespace std::chrono;
 
     const auto finish_time = Now() + timeout;
 
     while (!done)
     {
         struct timeval tv = {0, 0};
-        if (timeout != microseconds::zero())
+        struct timeval *tv_p = &tv;
+        if (timeout != PipeBase::kInfiniteTimeout)
         {
             const auto remaining_dur = duration_cast<microseconds>(finish_time - Now());
             if (remaining_dur.count() <= 0)
             {
                 error.SetErrorString("timeout exceeded");
                 break;
             }
             const auto dur_secs = duration_cast<seconds>(remaining_dur);
             const auto dur_usecs = remaining_dur % seconds(1);
 
             tv.tv_sec = dur_secs.count();
             tv.tv_usec = dur_usecs.count();
         }
         else
-            tv.tv_sec = 1;
+            tv_p = nullptr;
 
         FD_ZERO(&fds);
         FD_SET(handle, &fds);
 
         const auto retval = ::select(handle + 1,
                                      (is_read) ? &fds : nullptr,
                                      (is_read) ? nullptr : &fds,
-                                     nullptr, &tv);
+                                     nullptr, tv_p);
         if (retval == -1)
         {
             if (errno == EINTR)
                 continue;
             error.SetErrorToErrno();
             break;
         }
         if (retval == 0)
PipePosix::OpenAsWriterWithTimeout(llvm::StringRef name, bool child_process_inherit, const std::chrono::microseconds &timeout)
     if (!child_process_inherit)
         flags |= O_CLOEXEC;
 
     using namespace std::chrono;
     const auto finish_time = Now() + timeout;
 
     while (!CanWrite())
     {
-        if (timeout != microseconds::zero())
+        if (timeout != kInfiniteTimeout)
         {
             const auto dur = duration_cast<microseconds>(finish_time - Now()).count();
             if (dur <= 0)
                 return Error("timeout exceeded - reader hasn't opened so far");
         }
 
         errno = 0;
         int fd = ::open(name.data(), flags);
return SelectIO(handle,
                           error.SetErrorToErrno();
 
                       return error;
                   },
                   timeout);
 }
 
 Error
-PipePosix::Write(const void *buf, size_t size, size_t &bytes_written)
+PipePosix::WriteWithTimeout(const void *buf,
+                            size_t size,
+                            const std::chrono::microseconds &timeout,
+                            size_t &bytes_written)
 {
     bytes_written = 0;
     if (!CanWrite())
         return Error(EINVAL, eErrorTypePOSIX);
 
     auto handle = GetWriteFileDescriptor();
     return SelectIO(handle,
                     false,
return SelectIO(handle,
                             if (bytes_written == size)
                                 done = true;
                         }
                         else
                             error.SetErrorToErrno();
 
                         return error;
                     },
-                    std::chrono::microseconds::zero());
+                    timeout);
 }

lldb/trunk/source/Host/windows/PipeWindows.cpp

if (!CanRead())
         return Error(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 Error(::GetLastError(), eErrorTypeWin32);
 
-    DWORD timeout = (duration == std::chrono::microseconds::zero()) ? INFINITE : duration.count() * 1000;
+    DWORD timeout = (duration == kInfiniteTimeout) ? INFINITE : 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 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;
PipeWindows::ReadWithTimeout(void *buf, size_t size, const std::chrono::microseconds &duration, size_t &bytes_read)
     if (!GetOverlappedResult(m_read, &m_read_overlapped, &sys_bytes_read, FALSE))
         return Error(::GetLastError(), eErrorTypeWin32);
 
     bytes_read = sys_bytes_read;
     return Error();
 }
 
 Error
-PipeWindows::Write(const void *buf, size_t num_bytes, size_t &bytes_written)
+PipeWindows::WriteWithTimeout(const void *buf,
+                              size_t num_bytes,
+                              const std::chrono::microseconds &duration,
+                              size_t &bytes_written)
 {
     if (!CanWrite())
         return Error(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 Error(::GetLastError(), eErrorTypeWin32);
 
-    BOOL result = GetOverlappedResult(m_write, &m_write_overlapped, &sys_bytes_written, TRUE);
+    DWORD timeout = (duration == kInfiniteTimeout) ? 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;
+        DWORD failure_error = ::GetLastError();
+        if (wait_result == WAIT_TIMEOUT)
+        {
+            BOOL cancel_result = CancelIoEx(m_read, &m_write_overlapped);
+            if (!cancel_result && GetLastError() == ERROR_NOT_FOUND)
+                failed = false;
+        }
+        if (failed)
+            return Error(failure_error, eErrorTypeWin32);
+    }
+
+    // Now we call GetOverlappedResult setting bWait to false, since we've already waited
+    // as long as we're willing to.
+    BOOL result = GetOverlappedResult(m_write, &m_write_overlapped, &sys_bytes_written, FALSE);
     if (!result)
         return Error(::GetLastError(), eErrorTypeWin32);
+
+    bytes_written = sys_bytes_written;
     return Error();
 }