[core] Use cross-platform syscall implementation (ray-project#50209)

Signed-off-by: dentiny <[email protected]>

Author: dentiny

src/ray/util/pipe_logger.h

@@ -30,23 +30,9 @@
 #include "ray/util/util.h"
 #include "spdlog/logger.h"
 
-#if defined(__APPLE__) || defined(__linux__)
-#include <unistd.h>
-#elif defined(_WIN32)
-#include <windows.h>
-#endif
-
 namespace ray {
 
-// Environmenr variable, which indicates the pipe size of read.
-//
-// TODO(hjiang): Should document the env variable after end-to-end integration has
-// finished.
-inline constexpr std::string_view kPipeLogReadBufSizeEnv = "RAY_PIPE_LOG_READ_BUF_SIZE";
-
 // File handle requires active destruction via owner calling [Close].
-//
-// TODO(hjiang): Wrap fd with spdlog sink to manage stream flush and close.
 class RedirectionFileHandle {
  public:
   RedirectionFileHandle() = default;
@@ -100,17 +86,8 @@ class RedirectionFileHandle {
   MEMFD_TYPE_NON_UNIQUE GetWriteHandle() const { return write_handle_; }
 
   // Write the given data into redirection handle; currently only for testing usage.
-  //
-  // TODO(hjiang): Use platform compatible API, see
-  // https://github.com/ray-project/ray/pull/50170
   void CompleteWrite(const char *data, size_t len) {
-#if defined(__APPLE__) || defined(__linux__)
-    [[maybe_unused]] auto x = write(write_handle_, data, len);
-#elif defined(_WIN32)
-    DWORD bytes_written;
-    [[maybe_unused]] auto x =
-        WriteFile(write_handle_, data, (DWORD)len, &bytes_written, NULL);
-#endif
+    RAY_CHECK_OK(::ray::CompleteWrite(write_handle_, data, len));
   }
 
  private:

src/ray/util/spdlog_fd_sink.h

@@ -19,12 +19,6 @@
 #include "ray/util/compat.h"
 #include "ray/util/util.h"
 
-#if defined(__APPLE__) || defined(__linux__)
-#include <unistd.h>
-#elif defined(_WIN32)
-#include <windows.h>
-#endif
-
 namespace ray {
 
 // A sink which logs to the file descriptor.
@@ -39,26 +33,9 @@ class non_owned_fd_sink final : public spdlog::sinks::base_sink<Mutex> {
   void sink_it_(const spdlog::details::log_msg &msg) override {
     spdlog::memory_buf_t formatted;
     spdlog::sinks::base_sink<Mutex>::formatter_->format(msg, formatted);
-
-#if defined(__APPLE__) || defined(__linux__)
-    RAY_CHECK_EQ(write(fd_, formatted.data(), formatted.size()),
-                 static_cast<ssize_t>(formatted.size()))
-        << "Fails to write because " << strerror(errno);
-#elif defined(_WIN32)
-    DWORD bytes_written;
-    BOOL success =
-        WriteFile(fd_, formatted.data(), (DWORD)formatted.size(), &bytes_written, NULL);
-    RAY_CHECK(success);
-    RAY_CHECK_EQ((DWORD)formatted.size(), bytes_written);
-#endif
-  }
-  void flush_() override {
-#if defined(__APPLE__) || defined(__linux__)
-    RAY_CHECK_EQ(fdatasync(fd_), 0) << "Fails to flush file because " << strerror(errno);
-#elif defined(_WIN32)
-    RAY_CHECK(FlushFileBuffers(fd_));
-#endif
+    RAY_CHECK_OK(CompleteWrite(fd_, formatted.data(), formatted.size()));
   }
+  void flush_() override { RAY_CHECK_OK(Flush(fd_)); }
 
  private:
   MEMFD_TYPE_NON_UNIQUE fd_;