[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

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

src/ray/util/dup2_wrapper.h

@@ -0,0 +1,43 @@
+// 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 {
+
+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 oldfd, MEMFD_TYPE_NON_UNIQUE restorefd)
+      : oldfd_(oldfd), restorefd_(restorefd) {}
+
+  MEMFD_TYPE_NON_UNIQUE oldfd_;
+  MEMFD_TYPE_NON_UNIQUE restorefd_;
+};
+
+}  // namespace ray

src/ray/util/dup2_wrapper_posix.cc

@@ -0,0 +1,43 @@
+// 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/dup2_wrapper.h"
+#include "ray/util/logging.h"
+
+namespace ray {
+
+/*static*/ std::unique_ptr<ScopedDup2Wrapper> ScopedDup2Wrapper::New(int oldfd,
+                                                                     int newfd) {
+  const int restorefd = dup(oldfd);
+  RAY_CHECK_NE(restorefd, -1) << "Fails to duplicate oldfd " << oldfd << " 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(oldfd, restorefd));
+}
+
+ScopedDup2Wrapper::~ScopedDup2Wrapper() {
+  const int ret = dup2(oldfd_, restorefd_);
+  RAY_CHECK_NE(ret, -1) << "Fails to duplicate oldfd " << oldfd_ << " to " << restorefd_
+                        << " because " << strerror(errno);
+}
+
+}  // namespace ray

src/ray/util/dup2_wrapper_windows.cc

@@ -0,0 +1,54 @@
+// 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/dup2_wrapper.h"
+
+namespace ray {
+
+/*static*/ std::unique_ptr<ScopedDup2Wrapper> ScopedDup2Wrapper::New(HANDLE oldfd,
+                                                                     HANLE newfd) {
+  HANDLE restorefd = NULL;
+  BOOL success = DuplicateHandle(GetCurrentProcess(),
+                                 oldfd,
+                                 GetCurrentProcess(),
+                                 &restorefd,
+                                 0,
+                                 FALSE,
+                                 DUPLICATE_SAME_ACCESS);
+  RAY_CHECK(success);
+
+  success = DuplicateHandle(GetCurrentProcess(),
+                            oldfd,
+                            GetCurrentProcess(),
+                            &newfd,
+                            0,
+                            FALSE,
+                            DUPLICATE_SAME_ACCESS);
+  RAY_CHECK(success);
+
+  return std::unique_ptr<ScopedDup2Wrapper>(new ScopedDup2Wrapper(oldfd, restorefd));
+}
+
+ScopedDup2Wrapper::~ScopedDup2Wrapper() {
+  BOOL success = DuplicateHandle(GetCurrentProcess(),
+                                 oldfd_,
+                                 GetCurrentProcess(),
+                                 &restorefd_,
+                                 0,
+                                 FALSE,
+                                 DUPLICATE_SAME_ACCESS);
+  RAY_CHECK(success);
+}
+
+}  // namespace ray

src/ray/util/tests/BUILD

@@ -323,3 +323,18 @@ ray_cc_test(
     size = "small",
     tags = ["team:core"],
 )
+
+ray_cc_test(
+    name = "dup2_wrapper_test",
+    srcs = ["dup2_wrapper_test.cc"],
+    deps = [
+        "//src/ray/common/test:testing",
+        "//src/ray/util:compat",
+        "//src/ray/util:dup2_wrapper",
+        "//src/ray/util:filesystem",
+        "//src/ray/util:temporary_directory",
+        "@com_google_googletest//:gtest_main",
+    ],
+    size = "small",
+    tags = ["team:core"],
+)

src/ray/util/tests/dup2_wrapper_test.cc

@@ -0,0 +1,84 @@
+// 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/dup2_wrapper.h"
+
+#include <gtest/gtest.h>
+
+#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"
+
+#if defined(__APPLE__) || defined(__linux__)
+#include <fcntl.h>
+#include <unistd.h>
+#elif defined(_WIN32)
+#include <windows.h>
+#endif
+
+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();
+
+#if defined(__APPLE__) || defined(__linux__)
+  int fd = open(path_string.data(), O_RDWR | O_CREAT, S_IRUSR | S_IWUSR);
+  ASSERT_NE(fd, -1);
+#elif defined(_WIN32)
+  HANDLE fd = CreateFile(path_string,                   // File name
+                         GENERIC_READ | GENERIC_WRITE,  // Access mode: read/write
+                         0,                             // No sharing
+                         NULL,                          // Default security attributes
+                         OPEN_ALWAYS,  // Open file if it exists, create if it doesn't
+                         FILE_ATTRIBUTE_NORMAL,  // File attributes
+                         NULL);                  // No template file
+
+  // Check if the file was successfully opened or created
+  ASSERT_NE(fd, INVALID_HANDLE_VALUE);
+#endif
+
+  {
+    auto dup2_wrapper = ScopedDup2Wrapper::New(/*oldfd=*/fd, /*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