Debugger: Replace libuv with platform-specific connection code

Remove libuv usage from CMake debugger. Libuv has an async io model
and cppdap uses a sync model, so an extra thread and a buffer copy
were necessary to match semantics. In order to eliminate those
costs this commit implements the IO using platform specific APIs.

10 files changed, 690 insertions, 440 deletions

diff --git a/Source/CMakeLists.txt b/Source/CMakeLists.txt
index 708aec7..4c5b4e3 100644
--- a/Source/CMakeLists.txt
+++ b/Source/CMakeLists.txt
@@ -775,8 +775,6 @@ if(CMake_ENABLE_DEBUGGER)
       cmDebuggerBreakpointManager.h
       cmDebuggerExceptionManager.cxx
       cmDebuggerExceptionManager.h
-      cmDebuggerPipeConnection.cxx
-      cmDebuggerPipeConnection.h
       cmDebuggerProtocol.cxx
       cmDebuggerProtocol.h
       cmDebuggerSourceBreakpoint.cxx
@@ -794,6 +792,21 @@ if(CMake_ENABLE_DEBUGGER)
       cmDebuggerVariablesManager.cxx
       cmDebuggerVariablesManager.h
     )
+  if(WIN32)
+    target_sources(
+    CMakeLib
+    PRIVATE
+      cmDebuggerWindowsPipeConnection.cxx
+      cmDebuggerWindowsPipeConnection.h
+    )
+  else()
+    target_sources(
+    CMakeLib
+    PRIVATE
+      cmDebuggerPosixPipeConnection.cxx
+      cmDebuggerPosixPipeConnection.h
+    )
+  endif()
   target_link_libraries(CMakeLib PUBLIC cppdap::cppdap)
 endif()
 
@@ -945,7 +958,6 @@ if(CMake_BUILD_PCH)
     "$<$<COMPILE_LANGUAGE:CXX>:cmArgumentParser.h>"
     "$<$<COMPILE_LANGUAGE:CXX>:cmake.h>"
     "$<$<COMPILE_LANGUAGE:CXX>:cmCMakePath.h>"
-    "$<$<COMPILE_LANGUAGE:CXX>:cmDebuggerPipeConnection.h>"
     "$<$<COMPILE_LANGUAGE:CXX>:cmCurl.h>")
 
     set_source_files_properties(
diff --git a/Source/cmDebuggerPipeConnection.cxx b/Source/cmDebuggerPipeConnection.cxx
deleted file mode 100644
index 1b54346..0000000
--- a/Source/cmDebuggerPipeConnection.cxx
+++ /dev/null
@@ -1,293 +0,0 @@
-/* Distributed under the OSI-approved BSD 3-Clause License.  See accompanying
-   file Copyright.txt or https://cmake.org/licensing for details.  */
-#include "cmDebuggerPipeConnection.h"
-
-#include <algorithm>
-#include <cassert>
-#include <cstring>
-#include <stdexcept>
-#include <utility>
-
-namespace cmDebugger {
-
-struct write_req_t
-{
-  uv_write_t req;
-  uv_buf_t buf;
-};
-
-cmDebuggerPipeBase::cmDebuggerPipeBase(std::string name)
-  : PipeName(std::move(name))
-{
-  Loop.init();
-  LoopExit.init(
-    *Loop, [](uv_async_t* handle) { uv_stop((uv_loop_t*)handle->data); },
-    Loop);
-  WriteEvent.init(
-    *Loop,
-    [](uv_async_t* handle) {
-      auto* conn = static_cast<cmDebuggerPipeBase*>(handle->data);
-      conn->WriteInternal();
-    },
-    this);
-  PipeClose.init(
-    *Loop,
-    [](uv_async_t* handle) {
-      auto* conn = static_cast<cmDebuggerPipeBase*>(handle->data);
-      if (conn->Pipe.get()) {
-        conn->Pipe->data = nullptr;
-        conn->Pipe.reset();
-      }
-    },
-    this);
-}
-
-void cmDebuggerPipeBase::WaitForConnection()
-{
-  std::unique_lock<std::mutex> lock(Mutex);
-  Connected.wait(lock, [this] { return isOpen() || FailedToOpen; });
-  if (FailedToOpen) {
-    throw std::runtime_error("Failed to open debugger connection.");
-  }
-}
-
-void cmDebuggerPipeBase::close()
-{
-  std::unique_lock<std::mutex> lock(Mutex);
-
-  CloseConnection();
-  PipeClose.send();
-  lock.unlock();
-  ReadReady.notify_all();
-}
-
-size_t cmDebuggerPipeBase::read(void* buffer, size_t n)
-{
-  std::unique_lock<std::mutex> lock(Mutex);
-  ReadReady.wait(lock, [this] { return !isOpen() || !ReadBuffer.empty(); });
-
-  if (!isOpen() && ReadBuffer.empty()) {
-    return 0;
-  }
-
-  auto size = std::min(n, ReadBuffer.size());
-  memcpy(buffer, ReadBuffer.data(), size);
-  ReadBuffer.erase(0, size);
-  return size;
-}
-
-bool cmDebuggerPipeBase::write(const void* buffer, size_t n)
-{
-  std::unique_lock<std::mutex> lock(Mutex);
-  WriteBuffer.append(static_cast<const char*>(buffer), n);
-  lock.unlock();
-  WriteEvent.send();
-
-  lock.lock();
-  WriteComplete.wait(lock, [this] { return WriteBuffer.empty(); });
-  return true;
-}
-
-void cmDebuggerPipeBase::StopLoop()
-{
-  LoopExit.send();
-
-  if (LoopThread.joinable()) {
-    LoopThread.join();
-  }
-}
-
-void cmDebuggerPipeBase::BufferData(const std::string& data)
-{
-  std::unique_lock<std::mutex> lock(Mutex);
-  ReadBuffer += data;
-  lock.unlock();
-  ReadReady.notify_all();
-}
-
-void cmDebuggerPipeBase::WriteInternal()
-{
-  std::unique_lock<std::mutex> lock(Mutex);
-  auto n = WriteBuffer.length();
-  assert(this->Pipe.get());
-  write_req_t* req = new write_req_t;
-  req->req.data = &WriteComplete;
-  char* rawBuffer = new char[n];
-  req->buf = uv_buf_init(rawBuffer, static_cast<unsigned int>(n));
-  memcpy(req->buf.base, WriteBuffer.data(), n);
-  WriteBuffer.clear();
-  lock.unlock();
-
-  uv_write(
-    reinterpret_cast<uv_write_t*>(req), this->Pipe, &req->buf, 1,
-    [](uv_write_t* cb_req, int status) {
-      (void)status; // We need to free memory even if the write failed.
-      write_req_t* wr = reinterpret_cast<write_req_t*>(cb_req);
-      reinterpret_cast<std::condition_variable*>(wr->req.data)->notify_all();
-      delete[] (wr->buf.base);
-      delete wr;
-    });
-
-#ifdef __clang_analyzer__
-  // Tell clang-analyzer that 'rawBuffer' does not leak.
-  // We pass ownership to the closure.
-  delete[] rawBuffer;
-#endif
-}
-
-cmDebuggerPipeConnection::cmDebuggerPipeConnection(std::string name)
-  : cmDebuggerPipeBase(std::move(name))
-{
-  ServerPipeClose.init(
-    *Loop,
-    [](uv_async_t* handle) {
-      auto* conn = static_cast<cmDebuggerPipeConnection*>(handle->data);
-      if (conn->ServerPipe.get()) {
-        conn->ServerPipe->data = nullptr;
-        conn->ServerPipe.reset();
-      }
-    },
-    this);
-}
-
-cmDebuggerPipeConnection::~cmDebuggerPipeConnection()
-{
-  StopLoop();
-}
-
-bool cmDebuggerPipeConnection::StartListening(std::string& errorMessage)
-{
-  this->ServerPipe.init(*Loop, 0,
-                        static_cast<cmDebuggerPipeConnection*>(this));
-
-  int r;
-  if ((r = uv_pipe_bind(this->ServerPipe, this->PipeName.c_str())) != 0) {
-    errorMessage =
-      "Internal Error with " + this->PipeName + ": " + uv_err_name(r);
-    return false;
-  }
-
-  r = uv_listen(this->ServerPipe, 1, [](uv_stream_t* stream, int status) {
-    if (status >= 0) {
-      auto* conn = static_cast<cmDebuggerPipeConnection*>(stream->data);
-      if (conn) {
-        conn->Connect(stream);
-      }
-    }
-  });
-
-  if (r != 0) {
-    errorMessage =
-      "Internal Error listening on " + this->PipeName + ": " + uv_err_name(r);
-    return false;
-  }
-
-  // Start the libuv event loop thread so that a client can connect.
-  LoopThread = std::thread([this] { uv_run(Loop, UV_RUN_DEFAULT); });
-
-  StartedListening.set_value();
-
-  return true;
-}
-
-std::shared_ptr<dap::Reader> cmDebuggerPipeConnection::GetReader()
-{
-  return std::static_pointer_cast<dap::Reader>(shared_from_this());
-}
-
-std::shared_ptr<dap::Writer> cmDebuggerPipeConnection::GetWriter()
-{
-  return std::static_pointer_cast<dap::Writer>(shared_from_this());
-}
-
-bool cmDebuggerPipeConnection::isOpen()
-{
-  return this->Pipe.get() != nullptr;
-}
-
-void cmDebuggerPipeConnection::CloseConnection()
-{
-  ServerPipeClose.send();
-}
-
-void cmDebuggerPipeConnection::Connect(uv_stream_t* server)
-{
-  if (this->Pipe.get()) {
-    // Accept and close all pipes but the first:
-    cm::uv_pipe_ptr rejectPipe;
-
-    rejectPipe.init(*Loop, 0);
-    uv_accept(server, rejectPipe);
-
-    return;
-  }
-
-  cm::uv_pipe_ptr ClientPipe;
-  ClientPipe.init(*Loop, 0, static_cast<cmDebuggerPipeConnection*>(this));
-
-  if (uv_accept(server, ClientPipe) != 0) {
-    return;
-  }
-
-  StartReading<cmDebuggerPipeConnection>(ClientPipe);
-
-  std::unique_lock<std::mutex> lock(Mutex);
-  Pipe = std::move(ClientPipe);
-  lock.unlock();
-  Connected.notify_all();
-}
-
-cmDebuggerPipeClient::~cmDebuggerPipeClient()
-{
-  StopLoop();
-}
-
-void cmDebuggerPipeClient::Start()
-{
-  this->Pipe.init(*Loop, 0, static_cast<cmDebuggerPipeClient*>(this));
-
-  uv_connect_t* connect = new uv_connect_t;
-  connect->data = this;
-  uv_pipe_connect(
-    connect, Pipe, PipeName.c_str(), [](uv_connect_t* cb_connect, int status) {
-      auto* conn = static_cast<cmDebuggerPipeClient*>(cb_connect->data);
-      if (status >= 0) {
-        conn->Connect();
-      } else {
-        conn->FailConnection();
-      }
-      delete cb_connect;
-    });
-
-  // Start the libuv event loop so that the pipe can connect.
-  LoopThread = std::thread([this] { uv_run(Loop, UV_RUN_DEFAULT); });
-}
-
-bool cmDebuggerPipeClient::isOpen()
-{
-  return IsConnected;
-}
-
-void cmDebuggerPipeClient::CloseConnection()
-{
-  IsConnected = false;
-}
-
-void cmDebuggerPipeClient::Connect()
-{
-  StartReading<cmDebuggerPipeClient>(Pipe);
-  std::unique_lock<std::mutex> lock(Mutex);
-  IsConnected = true;
-  lock.unlock();
-  Connected.notify_all();
-}
-
-void cmDebuggerPipeClient::FailConnection()
-{
-  std::unique_lock<std::mutex> lock(Mutex);
-  FailedToOpen = true;
-  lock.unlock();
-  Connected.notify_all();
-}
-
-} // namespace cmDebugger
diff --git a/Source/cmDebuggerPipeConnection.h b/Source/cmDebuggerPipeConnection.h
deleted file mode 100644
index 0991ff7..0000000
--- a/Source/cmDebuggerPipeConnection.h
+++ /dev/null
@@ -1,139 +0,0 @@
-/* Distributed under the OSI-approved BSD 3-Clause License.  See accompanying
-   file Copyright.txt or https://cmake.org/licensing for details.  */
-#pragma once
-
-#include "cmConfigure.h" // IWYU pragma: keep
-
-#include <condition_variable>
-#include <cstddef>
-#include <future>
-#include <memory>
-#include <mutex>
-#include <string>
-#include <thread>
-
-#include <cm3p/cppdap/io.h>
-#include <cm3p/uv.h>
-
-#include "cmDebuggerAdapter.h"
-#include "cmUVHandlePtr.h"
-
-namespace cmDebugger {
-
-class cmDebuggerPipeBase : public dap::ReaderWriter
-{
-public:
-  cmDebuggerPipeBase(std::string name);
-
-  void WaitForConnection();
-
-  // dap::ReaderWriter implementation
-
-  void close() final;
-  size_t read(void* buffer, size_t n) final;
-  bool write(const void* buffer, size_t n) final;
-
-protected:
-  virtual void CloseConnection(){};
-  template <typename T>
-  void StartReading(uv_stream_t* stream)
-  {
-    uv_read_start(
-      stream,
-      // alloc_cb
-      [](uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
-        (void)handle;
-        char* rawBuffer = new char[suggested_size];
-        *buf =
-          uv_buf_init(rawBuffer, static_cast<unsigned int>(suggested_size));
-      },
-      // read_cb
-      [](uv_stream_t* readStream, ssize_t nread, const uv_buf_t* buf) {
-        auto conn = static_cast<T*>(readStream->data);
-        if (conn) {
-          if (nread >= 0) {
-            conn->BufferData(std::string(buf->base, buf->base + nread));
-          } else {
-            conn->close();
-          }
-        }
-        delete[] (buf->base);
-      });
-  }
-  void StopLoop();
-
-  const std::string PipeName;
-  std::thread LoopThread;
-  cm::uv_loop_ptr Loop;
-  cm::uv_pipe_ptr Pipe;
-  std::mutex Mutex;
-  std::condition_variable Connected;
-  bool FailedToOpen = false;
-
-private:
-  void BufferData(const std::string& data);
-  void WriteInternal();
-
-  cm::uv_async_ptr LoopExit;
-  cm::uv_async_ptr WriteEvent;
-  cm::uv_async_ptr PipeClose;
-  std::string WriteBuffer;
-  std::string ReadBuffer;
-  std::condition_variable ReadReady;
-  std::condition_variable WriteComplete;
-};
-
-class cmDebuggerPipeConnection
-  : public cmDebuggerPipeBase
-  , public cmDebuggerConnection
-  , public std::enable_shared_from_this<cmDebuggerPipeConnection>
-{
-public:
-  cmDebuggerPipeConnection(std::string name);
-  ~cmDebuggerPipeConnection() override;
-
-  void WaitForConnection() override
-  {
-    cmDebuggerPipeBase::WaitForConnection();
-  }
-
-  bool StartListening(std::string& errorMessage) override;
-  std::shared_ptr<dap::Reader> GetReader() override;
-  std::shared_ptr<dap::Writer> GetWriter() override;
-
-  // dap::ReaderWriter implementation
-
-  bool isOpen() override;
-
-  // Used for unit test synchronization
-  std::promise<void> StartedListening;
-
-private:
-  void CloseConnection() override;
-  void Connect(uv_stream_t* server);
-
-  cm::uv_pipe_ptr ServerPipe;
-  cm::uv_async_ptr ServerPipeClose;
-};
-
-class cmDebuggerPipeClient : public cmDebuggerPipeBase
-{
-public:
-  using cmDebuggerPipeBase::cmDebuggerPipeBase;
-  ~cmDebuggerPipeClient() override;
-
-  void Start();
-
-  // dap::ReaderWriter implementation
-
-  bool isOpen() override;
-
-private:
-  void CloseConnection() override;
-  void Connect();
-  void FailConnection();
-
-  bool IsConnected = false;
-};
-
-} // namespace cmDebugger
diff --git a/Source/cmDebuggerPosixPipeConnection.cxx b/Source/cmDebuggerPosixPipeConnection.cxx
new file mode 100644
index 0000000..0587450
--- /dev/null
+++ b/Source/cmDebuggerPosixPipeConnection.cxx
@@ -0,0 +1,205 @@
+/* Distributed under the OSI-approved BSD 3-Clause License.  See accompanying
+   file Copyright.txt or https://cmake.org/licensing for details.  */
+#include "cmDebuggerPosixPipeConnection.h"
+
+#include <cerrno>
+#include <cstring>
+#include <stdexcept>
+#include <utility>
+
+#include <unistd.h>
+
+#include <sys/socket.h>
+
+namespace cmDebugger {
+
+#ifndef _WIN32
+
+cmDebuggerPipeConnection_POSIX::cmDebuggerPipeConnection_POSIX(
+  std::string name)
+  : PipeName(std::move(name))
+{
+  addr.sun_path[0] = '\0';
+}
+
+cmDebuggerPipeConnection_POSIX::~cmDebuggerPipeConnection_POSIX()
+{
+  if (isOpen()) {
+    close();
+  }
+}
+
+bool cmDebuggerPipeConnection_POSIX::StartListening(std::string& errorMessage)
+{
+  listen_fd = socket(AF_UNIX, SOCK_STREAM, 0);
+  if (listen_fd < 0) {
+    errorMessage = "Failed to create socket: ";
+    errorMessage += strerror(errno);
+    return false;
+  }
+
+  addr.sun_family = AF_UNIX;
+  strncpy(addr.sun_path, PipeName.c_str(), sizeof(addr.sun_path));
+  addr.sun_path[sizeof(addr.sun_path) - 1] = '\0';
+  if (bind(listen_fd, (sockaddr*)&addr, sizeof(addr)) == -1) {
+    errorMessage = "Failed to bind name '";
+    errorMessage += addr.sun_path;
+    errorMessage += "' to socket: ";
+    errorMessage += strerror(errno);
+    close_listen();
+    return false;
+  }
+
+  if (listen(listen_fd, 1) == -1) {
+    errorMessage = "Failed to listen on socket: ";
+    errorMessage += strerror(errno);
+    close_listen();
+    return false;
+  }
+
+  StartedListening.set_value();
+  return true;
+}
+
+std::shared_ptr<dap::Reader> cmDebuggerPipeConnection_POSIX::GetReader()
+{
+  return std::static_pointer_cast<dap::Reader>(shared_from_this());
+}
+
+std::shared_ptr<dap::Writer> cmDebuggerPipeConnection_POSIX::GetWriter()
+{
+  return std::static_pointer_cast<dap::Writer>(shared_from_this());
+}
+
+bool cmDebuggerPipeConnection_POSIX::isOpen()
+{
+  return rw_pipe >= 0;
+}
+
+void cmDebuggerPipeConnection_POSIX::close()
+{
+  close_listen();
+  ::close(rw_pipe);
+  rw_pipe = -1;
+}
+
+void cmDebuggerPipeConnection_POSIX::close_listen()
+{
+  if (strlen(addr.sun_path) > 0) {
+    unlink(addr.sun_path);
+    addr.sun_path[0] = '\0';
+  }
+  ::close(listen_fd);
+  listen_fd = -1;
+}
+
+void cmDebuggerPipeConnection_POSIX::WaitForConnection()
+{
+  sockaddr_un laddr;
+  socklen_t len = sizeof(laddr);
+  rw_pipe = accept(listen_fd, (sockaddr*)&laddr, &len);
+  if (rw_pipe < 0) {
+    close();
+    return;
+  }
+
+  close_listen(); // no longer need the listen resources
+}
+
+size_t cmDebuggerPipeConnection_POSIX::read(void* buffer, size_t n)
+{
+  size_t result = 0;
+  if (rw_pipe >= 0) {
+    result = ::read(rw_pipe, buffer, n);
+    if (result == 0) {
+      close();
+    }
+  }
+
+  return result;
+}
+
+bool cmDebuggerPipeConnection_POSIX::write(void const* buffer, size_t n)
+{
+  bool result = false;
+  if (rw_pipe >= 0) {
+    result = ::write(rw_pipe, buffer, n) >= 0;
+    if (!result) {
+      close();
+    }
+  }
+
+  return result;
+}
+
+cmDebuggerPipeClient_POSIX::cmDebuggerPipeClient_POSIX(std::string name)
+  : PipeName(std::move(name))
+{
+}
+
+cmDebuggerPipeClient_POSIX::~cmDebuggerPipeClient_POSIX()
+{
+  close();
+}
+
+void cmDebuggerPipeClient_POSIX::WaitForConnection()
+{
+  rw_pipe = socket(AF_UNIX, SOCK_STREAM, 0);
+  if (rw_pipe < 0) {
+    throw std::runtime_error(std::string("Failed to create socket: ") +
+                             strerror(errno));
+  }
+
+  sockaddr_un addr;
+  addr.sun_family = AF_UNIX;
+  strncpy(addr.sun_path, PipeName.c_str(), sizeof(addr.sun_path));
+  addr.sun_path[sizeof(addr.sun_path) - 1] = '\0';
+  if (connect(rw_pipe, (sockaddr*)&addr, sizeof(addr)) == -1) {
+    close();
+    throw std::runtime_error(
+      std::string("Failed to connect path to socket: ") + strerror(errno));
+  }
+}
+
+bool cmDebuggerPipeClient_POSIX::isOpen()
+{
+  return rw_pipe >= 0;
+}
+
+void cmDebuggerPipeClient_POSIX::close()
+{
+  if (isOpen()) {
+    ::close(rw_pipe);
+    rw_pipe = -1;
+  }
+}
+
+size_t cmDebuggerPipeClient_POSIX::read(void* buffer, size_t n)
+{
+  int count = 0;
+  if (isOpen()) {
+    count = static_cast<int>(::read(rw_pipe, buffer, n));
+    if (count == 0) {
+      close();
+    }
+  }
+
+  return count;
+}
+
+bool cmDebuggerPipeClient_POSIX::write(void const* buffer, size_t n)
+{
+  int count = 0;
+  if (isOpen()) {
+    count = static_cast<int>(::write(rw_pipe, buffer, n));
+    if (count < 0) {
+      close();
+    }
+  }
+
+  return count > 0;
+}
+
+#endif // !_WIN32
+
+} // namespace cmDebugger
diff --git a/Source/cmDebuggerPosixPipeConnection.h b/Source/cmDebuggerPosixPipeConnection.h
new file mode 100644
index 0000000..42642fa
--- /dev/null
+++ b/Source/cmDebuggerPosixPipeConnection.h
@@ -0,0 +1,81 @@
+/* Distributed under the OSI-approved BSD 3-Clause License.  See accompanying
+   file Copyright.txt or https://cmake.org/licensing for details.  */
+#pragma once
+
+#include "cmConfigure.h" // IWYU pragma: keep
+
+#include <cstddef>
+#include <future>
+#include <memory>
+#include <string>
+
+#include <cm3p/cppdap/io.h>
+
+#include <sys/un.h>
+
+#include "cmDebuggerAdapter.h"
+
+namespace cmDebugger {
+
+#ifndef _WIN32
+
+class cmDebuggerPipeConnection_POSIX
+  : public dap::ReaderWriter
+  , public cmDebuggerConnection
+  , public std::enable_shared_from_this<cmDebuggerPipeConnection_POSIX>
+{
+public:
+  cmDebuggerPipeConnection_POSIX(std::string name);
+  ~cmDebuggerPipeConnection_POSIX() override;
+
+  void WaitForConnection() override;
+
+  bool StartListening(std::string& errorMessage) override;
+  std::shared_ptr<dap::Reader> GetReader() override;
+  std::shared_ptr<dap::Writer> GetWriter() override;
+
+  // dap::ReaderWriter implementation
+
+  bool isOpen() override;
+  void close() override;
+  size_t read(void* buffer, size_t n) override;
+  bool write(void const* buffer, size_t n) override;
+
+  // Used for unit test synchronization
+  std::promise<void> StartedListening;
+
+private:
+  void close_listen(); // release listen resources
+
+  std::string const PipeName;
+  sockaddr_un addr;
+  int listen_fd = -1; // listen fd
+  int rw_pipe = -1;   // rw fd
+};
+
+using cmDebuggerPipeConnection = cmDebuggerPipeConnection_POSIX;
+
+class cmDebuggerPipeClient_POSIX
+  : public dap::ReaderWriter
+  , public std::enable_shared_from_this<cmDebuggerPipeClient_POSIX>
+{
+public:
+  cmDebuggerPipeClient_POSIX(std::string name);
+  ~cmDebuggerPipeClient_POSIX() override;
+  void WaitForConnection();
+
+  bool isOpen() override;
+  void close() override;
+  size_t read(void* buffer, size_t n) override;
+  bool write(void const* buffer, size_t n) override;
+
+private:
+  std::string const PipeName;
+  int rw_pipe = -1;
+};
+
+using cmDebuggerPipeClient = cmDebuggerPipeClient_POSIX;
+
+#endif // !_WIN32
+
+} // namespace cmDebugger
diff --git a/Source/cmDebuggerWindowsPipeConnection.cxx b/Source/cmDebuggerWindowsPipeConnection.cxx
new file mode 100644
index 0000000..1c6a2a7
--- /dev/null
+++ b/Source/cmDebuggerWindowsPipeConnection.cxx
@@ -0,0 +1,272 @@
+/* Distributed under the OSI-approved BSD 3-Clause License.  See accompanying
+   file Copyright.txt or https://cmake.org/licensing for details.  */
+#include "cmDebuggerWindowsPipeConnection.h"
+
+#include <algorithm>
+#include <cassert>
+#include <cstring>
+#include <stdexcept>
+#include <utility>
+
+namespace cmDebugger {
+
+#ifdef _WIN32
+
+DuplexPipe_WIN32::DuplexPipe_WIN32(HANDLE pipe)
+  : hPipe(pipe)
+{
+  readOp.Offset = readOp.OffsetHigh = 0;
+  readOp.hEvent = CreateEvent(NULL, true, false, NULL);
+  writeOp.Offset = readOp.OffsetHigh = 0;
+  writeOp.hEvent = CreateEvent(NULL, true, false, NULL);
+}
+
+DuplexPipe_WIN32::~DuplexPipe_WIN32()
+{
+  close();
+}
+
+size_t DuplexPipe_WIN32::read(void* buffer, size_t n)
+{
+  if (hPipe != INVALID_HANDLE_VALUE) {
+    readOp.Offset = readOp.OffsetHigh = 0;
+    ResetEvent(readOp.hEvent);
+    auto r = ReadFile(hPipe, buffer, n, NULL, &readOp);
+    auto err = GetLastError();
+    if (r || err == ERROR_IO_PENDING) {
+      DWORD nRead = 0;
+      if (GetOverlappedResult(hPipe, &readOp, &nRead, true)) {
+        return nRead;
+      }
+    }
+  }
+
+  return 0;
+}
+
+bool DuplexPipe_WIN32::write(void const* buffer, size_t n)
+{
+  if (hPipe != INVALID_HANDLE_VALUE) {
+    writeOp.Offset = writeOp.OffsetHigh = 0;
+    ResetEvent(writeOp.hEvent);
+    auto w = WriteFile(hPipe, buffer, n, NULL, &writeOp);
+    auto err = GetLastError();
+    if (w || err == ERROR_IO_PENDING) {
+      DWORD nWrite = 0;
+      if (GetOverlappedResult(hPipe, &writeOp, &nWrite, true)) {
+        return n == nWrite;
+      }
+    }
+  }
+
+  return false;
+}
+
+void DuplexPipe_WIN32::close()
+{
+  CloseHandle(hPipe);
+  hPipe = INVALID_HANDLE_VALUE;
+  CloseHandle(readOp.hEvent);
+  CloseHandle(writeOp.hEvent);
+  readOp.hEvent = writeOp.hEvent = INVALID_HANDLE_VALUE;
+}
+
+bool DuplexPipe_WIN32::WaitForConnection()
+{
+  auto connect = ConnectNamedPipe(hPipe, &readOp);
+  auto err = GetLastError();
+  if (!connect && err == ERROR_IO_PENDING) {
+    DWORD ignored;
+    if (GetOverlappedResult(hPipe, &readOp, &ignored, true)) {
+      return true;
+    }
+  }
+
+  return connect || err == ERROR_PIPE_CONNECTED;
+}
+
+cmDebuggerPipeConnection_WIN32::cmDebuggerPipeConnection_WIN32(
+  std::string name)
+  : PipeName(std::move(name))
+  , pipes(nullptr)
+{
+}
+
+cmDebuggerPipeConnection_WIN32::~cmDebuggerPipeConnection_WIN32()
+{
+  if (isOpen()) {
+    pipes = nullptr;
+  }
+}
+
+bool cmDebuggerPipeConnection_WIN32::StartListening(std::string& errorMessage)
+{
+  bool result = true;
+
+  auto hPipe = CreateNamedPipeA(
+    PipeName.c_str(),
+    PIPE_ACCESS_DUPLEX | FILE_FLAG_FIRST_PIPE_INSTANCE | FILE_FLAG_OVERLAPPED,
+    PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_REJECT_REMOTE_CLIENTS, 1,
+    1024 * 16, 1024 * 16, NMPWAIT_USE_DEFAULT_WAIT, NULL);
+
+  if (hPipe == INVALID_HANDLE_VALUE) {
+    auto err = GetLastError();
+    errorMessage = GetErrorMessage(err);
+    result = false;
+  }
+
+  if (result) {
+    pipes = std::make_unique<DuplexPipe_WIN32>(hPipe);
+  }
+
+  StartedListening.set_value();
+  return result;
+}
+
+std::string cmDebuggerPipeConnection_WIN32::GetErrorMessage(DWORD errorCode)
+{
+  LPSTR message = nullptr;
+  DWORD size = FormatMessageA(
+    FORMAT_MESSAGE_ALLOCATE_BUFFER | FORMAT_MESSAGE_FROM_SYSTEM |
+      FORMAT_MESSAGE_IGNORE_INSERTS,
+    nullptr, errorCode, MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
+    (LPSTR)&message, 0, nullptr);
+  std::string errorMessage = "Internal Error with " + this->PipeName + ": " +
+    std::string(message, size);
+  LocalFree(message);
+  return errorMessage;
+}
+
+std::shared_ptr<dap::Reader> cmDebuggerPipeConnection_WIN32::GetReader()
+{
+  return std::static_pointer_cast<dap::Reader>(shared_from_this());
+}
+
+std::shared_ptr<dap::Writer> cmDebuggerPipeConnection_WIN32::GetWriter()
+{
+  return std::static_pointer_cast<dap::Writer>(shared_from_this());
+}
+
+bool cmDebuggerPipeConnection_WIN32::isOpen()
+{
+  return pipes != nullptr;
+}
+
+void cmDebuggerPipeConnection_WIN32::close()
+{
+  CloseConnection();
+}
+
+void cmDebuggerPipeConnection_WIN32::CloseConnection()
+{
+  if (isOpen()) {
+    pipes->close();
+    pipes = nullptr;
+  }
+}
+
+void cmDebuggerPipeConnection_WIN32::WaitForConnection()
+{
+  if (!isOpen()) {
+    return;
+  }
+
+  if (pipes->WaitForConnection()) {
+    return;
+  }
+
+  CloseConnection();
+}
+
+size_t cmDebuggerPipeConnection_WIN32::read(void* buffer, size_t n)
+{
+  size_t result = 0;
+  if (isOpen()) {
+    result = pipes->read(buffer, n);
+    if (result == 0) {
+      CloseConnection();
+    }
+  }
+
+  return result;
+}
+
+bool cmDebuggerPipeConnection_WIN32::write(void const* buffer, size_t n)
+{
+  bool result = false;
+  if (isOpen()) {
+    result = pipes->write(buffer, n);
+    if (!result) {
+      CloseConnection();
+    }
+  }
+
+  return result;
+}
+
+cmDebuggerPipeClient_WIN32::cmDebuggerPipeClient_WIN32(std::string name)
+  : PipeName(std::move(name))
+{
+}
+
+cmDebuggerPipeClient_WIN32::~cmDebuggerPipeClient_WIN32()
+{
+  close();
+}
+
+void cmDebuggerPipeClient_WIN32::WaitForConnection()
+{
+  if (!isOpen()) {
+    auto hPipe = CreateFileA(PipeName.c_str(), GENERIC_READ | GENERIC_WRITE, 0,
+                             NULL, OPEN_EXISTING, FILE_FLAG_OVERLAPPED, NULL);
+    if (hPipe == INVALID_HANDLE_VALUE) {
+      auto err = GetLastError();
+      throw std::runtime_error("CreateFile failed with " + err);
+    }
+
+    pipes = std::make_unique<DuplexPipe_WIN32>(hPipe);
+  }
+}
+
+bool cmDebuggerPipeClient_WIN32::isOpen()
+{
+  return pipes != nullptr;
+}
+
+void cmDebuggerPipeClient_WIN32::close()
+{
+  if (isOpen()) {
+    pipes->close();
+    pipes = nullptr;
+  }
+}
+
+size_t cmDebuggerPipeClient_WIN32::read(void* buffer, size_t n)
+{
+  size_t result = 0;
+  if (isOpen()) {
+    result = pipes->read(buffer, n);
+    if (result == 0) {
+      close();
+    }
+  }
+
+  return result;
+}
+
+bool cmDebuggerPipeClient_WIN32::write(void const* buffer, size_t n)
+{
+  bool result = false;
+  if (isOpen()) {
+    result = pipes->write(buffer, n);
+    if (!result) {
+      close();
+    }
+  }
+
+  return result;
+}
+
+#endif // _WIN32
+
+} // namespace cmDebugger
diff --git a/Source/cmDebuggerWindowsPipeConnection.h b/Source/cmDebuggerWindowsPipeConnection.h
new file mode 100644
index 0000000..88ed1de
--- /dev/null
+++ b/Source/cmDebuggerWindowsPipeConnection.h
@@ -0,0 +1,101 @@
+/* Distributed under the OSI-approved BSD 3-Clause License.  See accompanying
+   file Copyright.txt or https://cmake.org/licensing for details.  */
+#pragma once
+
+#include "cmConfigure.h" // IWYU pragma: keep
+
+#include <condition_variable>
+#include <cstddef>
+#include <future>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <thread>
+
+#include <windows.h>
+
+#include <cm3p/cppdap/io.h>
+
+#include "cmDebuggerAdapter.h"
+
+namespace cmDebugger {
+
+#ifdef _WIN32
+
+class DuplexPipe_WIN32
+{
+public:
+  DuplexPipe_WIN32(HANDLE read);
+  ~DuplexPipe_WIN32();
+
+  void close();
+  size_t read(void* buffer, size_t n);
+  bool write(void const* buffer, size_t n);
+
+  bool WaitForConnection();
+
+private:
+  HANDLE hPipe;
+  OVERLAPPED readOp;
+  OVERLAPPED writeOp;
+};
+
+class cmDebuggerPipeConnection_WIN32
+  : public dap::ReaderWriter
+  , public cmDebuggerConnection
+  , public std::enable_shared_from_this<cmDebuggerPipeConnection_WIN32>
+{
+public:
+  cmDebuggerPipeConnection_WIN32(std::string name);
+  ~cmDebuggerPipeConnection_WIN32() override;
+
+  void WaitForConnection() override;
+
+  bool StartListening(std::string& errorMessage) override;
+  std::shared_ptr<dap::Reader> GetReader() override;
+  std::shared_ptr<dap::Writer> GetWriter() override;
+
+  // dap::ReaderWriter implementation
+
+  bool isOpen() override;
+  void close() override;
+  size_t read(void* buffer, size_t n) override;
+  bool write(void const* buffer, size_t n) override;
+
+  // Used for unit test synchronization
+  std::promise<void> StartedListening;
+
+private:
+  void CloseConnection();
+  std::string GetErrorMessage(DWORD errorCode);
+
+  std::string const PipeName;
+  std::unique_ptr<DuplexPipe_WIN32> pipes;
+};
+
+using cmDebuggerPipeConnection = cmDebuggerPipeConnection_WIN32;
+
+class cmDebuggerPipeClient_WIN32
+  : public dap::ReaderWriter
+  , public std::enable_shared_from_this<cmDebuggerPipeClient_WIN32>
+{
+public:
+  cmDebuggerPipeClient_WIN32(std::string name);
+  ~cmDebuggerPipeClient_WIN32();
+  void WaitForConnection();
+
+  bool isOpen() override;
+  void close() override;
+  size_t read(void* buffer, size_t n) override;
+  bool write(void const* buffer, size_t n) override;
+
+private:
+  std::string const PipeName;
+  std::unique_ptr<DuplexPipe_WIN32> pipes;
+};
+
+using cmDebuggerPipeClient = cmDebuggerPipeClient_WIN32;
+
+#endif // _WIN32
+
+} // namespace cmDebugger
diff --git a/Source/cmake.cxx b/Source/cmake.cxx
index b8ebca5..ab4a7e6 100644
--- a/Source/cmake.cxx
+++ b/Source/cmake.cxx
@@ -40,7 +40,11 @@
 #include "cmCommands.h"
 #ifdef CMake_ENABLE_DEBUGGER
 #  include "cmDebuggerAdapter.h"
-#  include "cmDebuggerPipeConnection.h"
+#  ifdef _WIN32
+#    include "cmDebuggerWindowsPipeConnection.h"
+#  else //!_WIN32
+#    include "cmDebuggerPosixPipeConnection.h"
+#  endif //_WIN32
 #endif
 #include "cmDocumentation.h"
 #include "cmDocumentationEntry.h"
diff --git a/Tests/CMakeLib/testDebuggerAdapterPipe.cxx b/Tests/CMakeLib/testDebuggerAdapterPipe.cxx
index 643661d..c0f2e9b 100644
--- a/Tests/CMakeLib/testDebuggerAdapterPipe.cxx
+++ b/Tests/CMakeLib/testDebuggerAdapterPipe.cxx
@@ -19,13 +19,15 @@
 #include <cm3p/cppdap/types.h>
 
 #include "cmDebuggerAdapter.h"
-#include "cmDebuggerPipeConnection.h"
 #include "cmDebuggerProtocol.h"
 #include "cmVersionConfig.h"
 
 #ifdef _WIN32
 #  include "cmCryptoHash.h"
+#  include "cmDebuggerWindowsPipeConnection.h"
 #  include "cmSystemTools.h"
+#else
+#  include "cmDebuggerPosixPipeConnection.h"
 #endif
 
 #include "testCommon.h"
@@ -128,7 +130,7 @@ bool testProtocolWithPipes()
 
   auto client2Debugger =
     std::make_shared<cmDebugger::cmDebuggerPipeClient>(namedPipe);
-  client2Debugger->Start();
+
   client2Debugger->WaitForConnection();
   client->bind(client2Debugger, client2Debugger);
 
diff --git a/Tests/CMakeLib/testDebuggerNamedPipe.cxx b/Tests/CMakeLib/testDebuggerNamedPipe.cxx
index d2b0728..41761eb 100644
--- a/Tests/CMakeLib/testDebuggerNamedPipe.cxx
+++ b/Tests/CMakeLib/testDebuggerNamedPipe.cxx
@@ -16,7 +16,12 @@
 
 #include "cmsys/RegularExpression.hxx"
 
-#include "cmDebuggerPipeConnection.h"
+#ifdef _WIN32
+#  include "cmDebuggerWindowsPipeConnection.h"
+#else
+#  include "cmDebuggerPosixPipeConnection.h"
+#endif
+
 #include "cmSystemTools.h"
 
 #ifdef _WIN32
@@ -104,7 +109,7 @@ int runTest(int argc, char* argv[])
       attempt++;
       try {
         client = std::make_shared<cmDebugger::cmDebuggerPipeClient>(namedPipe);
-        client->Start();
+
         client->WaitForConnection();
         std::cout << "cmDebuggerPipeClient connected.\n";
         break;