[core] Implement dup2 wrapper

src/ray/common/status.h

@@ -313,6 +313,8 @@ class RAY_EXPORT Status {
   // Returns the string "OK" for success.
   std::string ToString() const;
 
+  std::string StatusString() const { return ToString(); }
+
   // Return a string representation of the status code, without the message
   // text or posix code information.
   std::string CodeAsString() const;

src/ray/common/status_or.h

@@ -151,6 +151,8 @@ class StatusOr {
 
   ABSL_MUST_USE_RESULT std::string message() const { return status_.message(); }
 
+  std::string StatusString() const { return status_.StatusString(); }
+
   // Returns a reference to the current `ray::Status` contained within the
   // `ray::StatusOr<T>`. If `ray::StatusOr<T>` contains a `T`, then this
   // function returns `ray::Ok()`.

src/ray/common/test/testing.h

@@ -16,6 +16,6 @@
 
 #pragma once
 
-#define RAY_EXPECT_OK(s) EXPECT_TRUE((s).ok())
+#define RAY_EXPECT_OK(s) EXPECT_TRUE((s).ok()) << s.StatusString()
 
-#define RAY_ASSERT_OK(s) ASSERT_TRUE((s).ok())
+#define RAY_ASSERT_OK(s) ASSERT_TRUE((s).ok()) << s.StatusString()

src/ray/util/BUILD

@@ -336,3 +336,16 @@ ray_cc_library(
         "@com_google_absl//absl/synchronization",
     ],
 )
+
+ray_cc_library(
+    name = "scoped_dup2_wrapper",
+    hdrs = ["scoped_dup2_wrapper.h"],
+    srcs = select({
+        "@platforms//os:windows": ["scoped_dup2_wrapper_windows.cc"],
+        "//conditions:default": ["scoped_dup2_wrapper_posix.cc"],
+    }),
+    deps = [
+        ":compat",
+        ":logging",
+    ],
+)

src/ray/util/scoped_dup2_wrapper.h

@@ -0,0 +1,46 @@
+// Copyright 2025 The Ray Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//  http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include <memory>
+
+#include "ray/util/compat.h"
+
+namespace ray {
+
+// A RAII-style wrapper for `dup2` syscall, which
+// - Works on cross-platform;
+// - Recovers file descriptor at destruction.
+class ScopedDup2Wrapper {
+ public:
+  // Duplicate [oldfd] to [newfd], same semantics with syscall `dup2`.
+  static std::unique_ptr<ScopedDup2Wrapper> New(MEMFD_TYPE_NON_UNIQUE oldfd,
+                                                MEMFD_TYPE_NON_UNIQUE newfd);
+
+  ScopedDup2Wrapper(const ScopedDup2Wrapper &) = delete;
+  ScopedDup2Wrapper &operator=(const ScopedDup2Wrapper &) = delete;
+
+  // Restore oldfd.
+  ~ScopedDup2Wrapper();
+
+ private:
+  ScopedDup2Wrapper(MEMFD_TYPE_NON_UNIQUE newfd, MEMFD_TYPE_NON_UNIQUE restorefd)
+      : newfd_(newfd), restorefd_(restorefd) {}
+
+  MEMFD_TYPE_NON_UNIQUE newfd_;
+  MEMFD_TYPE_NON_UNIQUE restorefd_;
+};
+
+}  // namespace ray

src/ray/util/scoped_dup2_wrapper_posix.cc

@@ -0,0 +1,45 @@
+// Copyright 2025 The Ray Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//  http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include <unistd.h>
+
+#include <cstring>
+
+#include "ray/util/logging.h"
+#include "ray/util/scoped_dup2_wrapper.h"
+
+namespace ray {
+
+/*static*/ std::unique_ptr<ScopedDup2Wrapper> ScopedDup2Wrapper::New(int oldfd,
+                                                                     int newfd) {
+  const int restorefd = dup(newfd);
+  RAY_CHECK_NE(restorefd, -1) << "Fails to duplicate newfd " << newfd << " because "
+                              << strerror(errno);
+
+  const int ret = dup2(oldfd, newfd);
+  RAY_CHECK_NE(ret, -1) << "Fails to duplicate oldfd " << oldfd << " to " << newfd
+                        << " because " << strerror(errno);
+
+  return std::unique_ptr<ScopedDup2Wrapper>(new ScopedDup2Wrapper(newfd, restorefd));
+}
+
+ScopedDup2Wrapper::~ScopedDup2Wrapper() {
+  int ret = dup2(restorefd_, newfd_);
+  RAY_CHECK_NE(ret, -1) << "Fails to duplicate restorefd " << restorefd_ << " to "
+                        << newfd_ << " because " << strerror(errno);
+
+  RAY_CHECK_OK(Close(restorefd_));
+}
+
+}  // namespace ray

src/ray/util/scoped_dup2_wrapper_windows.cc

@@ -0,0 +1,50 @@
+// Copyright 2025 The Ray Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//  http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include <fcntl.h>  // For _O_WTEXT
+#include <io.h>     // For _open_osfhandle
+#include <windows.h>
+
+#include "ray/util/scoped_dup2_wrapper.h"
+
+namespace ray {
+
+/*static*/ std::unique_ptr<ScopedDup2Wrapper> ScopedDup2Wrapper::New(HANDLE oldfd,
+                                                                     HANDLE newfd) {
+  HANDLE restorefd = NULL;
+  BOOL success = DuplicateHandle(GetCurrentProcess(),
+                                 newfd,
+                                 GetCurrentProcess(),
+                                 &restorefd,
+                                 0,
+                                 FALSE,
+                                 DUPLICATE_SAME_ACCESS);
+  RAY_CHECK(success);
+
+  int old_win_fd = _open_osfhandle(reinterpret_cast<intptr_t>(oldfd), _O_WRONLY);
+  int new_win_fd = _open_osfhandle(reinterpret_cast<intptr_t>(newfd), _O_WRONLY);
+  RAY_CHECK_NE(_dup2(old_win_fd, new_win_fd), -1) << "Fails to duplicate file descriptor";
+
+  return std::unique_ptr<ScopedDup2Wrapper>(new ScopedDup2Wrapper(newfd, restorefd));
+}
+
+ScopedDup2Wrapper::~ScopedDup2Wrapper() {
+  int restore_win_fd = _open_osfhandle(reinterpret_cast<intptr_t>(restorefd_), _O_WRONLY);
+  int new_win_fd = _open_osfhandle(reinterpret_cast<intptr_t>(newfd_), _O_WRONLY);
+  RAY_CHECK_NE(_dup2(restore_win_fd, new_win_fd), -1)
+      << "Fails to duplicate file descriptor";
+  RAY_CHECK_OK(Close(restorefd_));
+}
+
+}  // namespace ray

src/ray/util/tests/BUILD

@@ -345,3 +345,19 @@ ray_cc_test(
     size = "small",
     tags = ["team:core"],
 )
+
+ray_cc_test(
+    name = "scoped_dup2_wrapper_test",
+    srcs = ["scoped_dup2_wrapper_test.cc"],
+    deps = [
+        "//src/ray/common/test:testing",
+        "//src/ray/util:compat",
+        "//src/ray/util:filesystem",
+        "//src/ray/util:scoped_dup2_wrapper",
+        "//src/ray/util:temporary_directory",
+        "@boost//:iostreams",
+        "@com_google_googletest//:gtest_main",
+    ],
+    size = "small",
+    tags = ["team:core"],
+)

src/ray/util/tests/scoped_dup2_wrapper_test.cc

@@ -0,0 +1,66 @@
+// Copyright 2025 The Ray Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//  http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "ray/util/scoped_dup2_wrapper.h"
+
+#include <gtest/gtest.h>
+
+#include <boost/iostreams/device/file_descriptor.hpp>
+#include <iostream>
+#include <string>
+#include <string_view>
+
+#include "ray/common/test/testing.h"
+#include "ray/util/compat.h"
+#include "ray/util/filesystem.h"
+#include "ray/util/temporary_directory.h"
+
+namespace ray {
+
+namespace {
+
+constexpr std::string_view kContent = "helloworld\n";
+
+TEST(ScopedDup2WrapperTest, BasicTest) {
+  ScopedTemporaryDirectory temp_dir;
+  const auto dir = temp_dir.GetDirectory();
+  const auto path = dir / "test_file";
+  const std::string path_string = path.string();
+  boost::iostreams::file_descriptor_sink fd_sink{path_string, std::ios_base::out};
+
+  {
+    auto dup2_wrapper =
+        ScopedDup2Wrapper::New(/*oldfd=*/fd_sink.handle(), /*newfd=*/GetStderrHandle());
+
+    // Write to stdout should appear in file.
+    std::cerr << kContent << std::flush;
+    const auto actual_content = ReadEntireFile(path_string);
+    RAY_ASSERT_OK(actual_content);
+    EXPECT_EQ(*actual_content, kContent);
+  }
+
+  testing::internal::CaptureStderr();
+  std::cerr << kContent << std::flush;
+  const std::string stderr_content = testing::internal::GetCapturedStderr();
+  EXPECT_EQ(stderr_content, kContent);
+
+  // Not changed since last write.
+  const auto actual_content = ReadEntireFile(path_string);
+  RAY_ASSERT_OK(actual_content);
+  EXPECT_EQ(*actual_content, kContent);
+}
+
+}  // namespace
+
+}  // namespace ray