[core] Implement redirection without tee and rotation on windows

src/ray/util/filesystem.cc

@@ -15,6 +15,7 @@
 #include "ray/util/filesystem.h"
 
 #include <cstdlib>
+#include <fstream>
 
 #include "ray/util/logging.h"
 
@@ -60,4 +61,18 @@ std::string GetUserTempDir() {
   return result;
 }
 
+std::string CompleteReadFile(const std::string &fname) {
+  std::ifstream file(fname);
+  RAY_CHECK(file.good()) << "Fails to open file " << fname;
+
+  std::ostringstream buffer;
+  buffer << file.rdbuf();
+  RAY_CHECK(file.good()) << "Fails to read from file " << fname;
+
+  std::string content = buffer.str();
+  file.close();
+
+  return content;
+}
+
 }  // namespace ray

src/ray/util/filesystem.h

@@ -59,4 +59,12 @@ std::string JoinPaths(std::string base, const Paths &...components) {
   (join(base, std::string_view(components)), ...);
   return base;
 }
+
+// Read the whole content for the given [fname], and return as string.
+// If any error happens, error message will be logged and the program will exit
+// immediately.
+//
+// TODO(hjiang): Use StatusOr as return type in the followup PR.
+std::string CompleteReadFile(const std::string &fname);
+
 }  // namespace ray

src/ray/util/pipe_logger.cc

@@ -138,6 +138,9 @@ void StartStreamDump(ReadFunc read_func,
         last_line.clear();
 
         // We only log non-empty lines.
+        //
+        // TODO(hjiang): Newliners should also appear in the stdout/stderr/log, current
+        // behavior simply ignore everything.
         if (!cur_new_line.empty()) {
           absl::MutexLock lock(&stream_dumper->mu);
           stream_dumper->content.emplace_back(std::move(cur_new_line));
@@ -242,6 +245,33 @@ RedirectionFileHandle OpenFileForRedirection(const std::string &file_path) {
 
   return RedirectionFileHandle{fd, std::move(flush_fn), std::move(close_fn)};
 }
+#elif defined(_WIN32)
+#include <windows.h>
+RedirectionFileHandle OpenFileForRedirection(const std::string &file_path) {
+  HANDLE file_handle = CreateFile(file_path.c_str(),
+                                  GENERIC_WRITE,
+                                  0,                      // No sharing
+                                  NULL,                   // Default security attributes
+                                  CREATE_ALWAYS,          // Always create a new file
+                                  FILE_ATTRIBUTE_NORMAL,  // Normal file attributes
+                                  NULL                    // No template file
+  );
+  RAY_CHECK(file_handle != INVALID_HANDLE_VALUE)
+      << "Fails to open file " << file_path << " with error "
+      << std::to_string(GetLastError());
+
+  auto flush_fn = [file_handle]() {
+    RAY_CHECK(FlushFileBuffers(file_handle))
+        << "Failed to flush data to disk with error: " << std::to_string(GetLastError());
+  };
+  auto close_fn = [file_handle]() {
+    RAY_CHECK(FlushFileBuffers(file_handle))
+        << "Failed to flush data to disk with error: " << std::to_string(GetLastError());
+    RAY_CHECK(CloseHandle(file_handle))
+        << "Failed to close file with error: " << std::to_string(GetLastError());
+  };
+  return RedirectionFileHandle{file_handle, std::move(flush_fn), std::move(close_fn)};
+}
 #endif
 
 }  // namespace
@@ -340,7 +370,9 @@ RedirectionFileHandle CreateRedirectionFileHandle(
 #elif defined(_WIN32)
 RedirectionFileHandle CreateRedirectionFileHandle(
     const StreamRedirectionOption &stream_redirect_opt) {
-  return RedirectionFileHandle{};
+  // TODO(hjiang): For windows, we currently doesn't support redirection with rotation and
+  // tee to stdout/stderr.
+  return OpenFileForRedirection(stream_redirect_opt.file_path);
 }
 #endif
 

src/ray/util/pipe_logger.h

@@ -107,7 +107,7 @@ class RedirectionFileHandle {
     RAY_CHECK_EQ(bytes_written, static_cast<ssize_t>(len));
   }
 #elif defined(_WIN32)
-  void CompleteWrite(char *data, size_t len) {
+  void CompleteWrite(const char *data, size_t len) {
     DWORD bytes_written = 0;
     WriteFile(write_handle_, data, len, &bytes_written, nullptr);
   }

src/ray/util/stream_redirection_utils.cc

@@ -35,9 +35,12 @@ namespace {
 #if defined(__APPLE__) || defined(__linux__)
 int GetStdoutHandle() { return STDOUT_FILENO; }
 int GetStderrHandle() { return STDERR_FILENO; }
+#elif defined(_WIN32)
+int GetStdoutHandle() { return _fileno(stdout); }
+int GetStderrHandle() { return _fileno(stderr); }
 #endif
 
-// TODO(hjiang): Revisit later, should be able to save some heap alllocation with
+// TODO(hjiang): Revisit later, should be able to save some heap allocation with
 // absl::InlinedVector.
 //
 // Maps from original stream file handle (i.e. stdout/stderr) to its stream redirector.
@@ -52,7 +55,6 @@ void SyncOnStreamRedirection() {
   }
 }
 
-#if defined(__APPLE__) || defined(__linux__)
 // Redirect the given [stream_fd] based on the specified option.
 void RedirectStream(int stream_fd, const StreamRedirectionOption &opt) {
   std::call_once(stream_exit_once_flag, []() {
@@ -61,14 +63,21 @@ void RedirectStream(int stream_fd, const StreamRedirectionOption &opt) {
   });
 
   RedirectionFileHandle handle = CreateRedirectionFileHandle(opt);
+
+#if defined(__APPLE__) || defined(__linux__)
   RAY_CHECK_NE(dup2(handle.GetWriteHandle(), stream_fd), -1)
       << "Fails to duplicate file descritor " << strerror(errno);
+#elif defined(_WIN32)
+  int pipe_write_fd =
+      _open_osfhandle(reinterpret_cast<intptr_t>(handle.GetWriteHandle()), _O_WRONLY);
+  RAY_CHECK_NE(_dup2(pipe_write_fd, stream_fd), -1)
+      << "Fails to duplicate file descritor.";
+#endif
 
   const bool is_new =
       redirection_file_handles.emplace(stream_fd, std::move(handle)).second;
   RAY_CHECK(is_new) << "Redirection has been register for stream " << stream_fd;
 }
-#endif
 
 void FlushOnRedirectedStream(int stream_fd) {
   auto iter = redirection_file_handles.find(stream_fd);
@@ -79,7 +88,6 @@ void FlushOnRedirectedStream(int stream_fd) {
 
 }  // namespace
 
-#if defined(__APPLE__) || defined(__linux__)
 void RedirectStdout(const StreamRedirectionOption &opt) {
   RedirectStream(GetStdoutHandle(), opt);
 }
@@ -89,11 +97,4 @@ void RedirectStderr(const StreamRedirectionOption &opt) {
 void FlushOnRedirectedStdout() { FlushOnRedirectedStream(GetStdoutHandle()); }
 void FlushOnRedirectedStderr() { FlushOnRedirectedStream(GetStderrHandle()); }
 
-#elif defined(_WIN32)
-void RedirectStdout(const StreamRedirectionOption &opt) { return; }
-void RedirectStderr(const StreamRedirectionOption &opt) { return; }
-void FlushOnRedirectedStdout() { return; }
-void FlushOnRedirectedStderr() { return; }
-#endif
-
 }  // namespace ray

src/ray/util/tests/BUILD

@@ -212,25 +212,14 @@ ray_cc_test(
     tags = ["team:core"],
 )
 
-ray_cc_library(
-    name = "unix_test_utils",
-    hdrs = ["unix_test_utils.h"],
-    srcs = ["unix_test_utils.cc"],
-    deps = [
-        "//src/ray/util",
-        "@boost//:scope_exit",
-    ],
-    testonly = True,
-)
-
 ray_cc_test(
     name = "pipe_logger_test",
     srcs = ["pipe_logger_test.cc"],
     deps = [
-        ":unix_test_utils",
         "//src/ray/util",
         "//src/ray/util:pipe_logger",
         "@com_google_googletest//:gtest_main",
+        "@com_google_absl//absl/cleanup",
     ],
     size = "small",
     tags = ["team:core"],
@@ -240,9 +229,8 @@ ray_cc_test(
     name = "stream_redirection_utils_test",
     srcs = ["stream_redirection_utils_test.cc"],
     deps = [
-        "//src/ray/util:stream_redirection_utils",
-        ":unix_test_utils",
         "//src/ray/util",
+        "//src/ray/util:stream_redirection_utils",
         "@com_google_googletest//:gtest_main",
     ],
     size = "small",