better wrong input check

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

Author: yicheng

python/ray/tests/BUILD.bazel

@@ -467,6 +467,7 @@ py_test_module_list(
         "test_label_scheduling.py",
         "test_minimal_install.py",
         "test_path_utils.py",
+        "test_pipe.py",
         "test_runtime_env_ray_minimal.py",
     ],
     tags = [

python/ray/tests/test_pipe.py

@@ -81,42 +81,6 @@ def test_sibling_processes_communicate():
     pipe.close()
 
 
-def test_multiple_pipes_to_one_child():
-    """One child writes to multiple pipes created by parent."""
-    pipe1 = Pipe()
-    pipe2 = Pipe()
-    writer_handle1 = pipe1.make_writer_handle()
-    writer_handle2 = pipe2.make_writer_handle()
-
-    code = f"""
-from ray._private.pipe import Pipe
-p1 = Pipe.from_writer_handle({writer_handle1})
-p2 = Pipe.from_writer_handle({writer_handle2})
-p1.write("data_for_pipe1")
-p2.write("data_for_pipe2")
-p1.close()
-p2.close()
-"""
-    proc = subprocess.Popen(
-        [sys.executable, "-c", code],
-        close_fds=False,
-    )
-
-    pipe1.close_writer_handle()
-    pipe2.close_writer_handle()
-
-    data1 = pipe1.read(timeout_s=5)
-    data2 = pipe2.read(timeout_s=5)
-    proc.wait()
-
-    assert data1 == "data_for_pipe1"
-    assert data2 == "data_for_pipe2"
-    assert proc.returncode == 0
-
-    pipe1.close()
-    pipe2.close()
-
-
 def test_read_timeout():
     """Read raises RuntimeError on timeout when no data available."""
     pipe = Pipe()

src/ray/util/tests/BUILD.bazel

@@ -220,6 +220,19 @@ ray_cc_test(
     ],
 )
 
+ray_cc_test(
+    name = "pipe_test",
+    size = "small",
+    srcs = ["pipe_test.cc"],
+    tags = ["team:core"],
+    deps = [
+        "//src/ray/util:pipe",
+        "//src/ray/util:process",
+        "@com_google_absl//absl/strings",
+        "@com_google_googletest//:gtest_main",
+    ],
+)
+
 ray_cc_test(
     name = "pipe_logger_test",
     size = "small",

src/ray/util/tests/pipe_test.cc

@@ -0,0 +1,114 @@
+// Copyright 2017 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/pipe.h"
+
+#include <gtest/gtest.h>
+
+#include <string>
+
+#include "absl/strings/str_cat.h"
+#include "ray/util/process.h"
+
+namespace ray {
+
+TEST(PipeTest, ParentReadChildWrite) {
+  Pipe pipe;
+  intptr_t writer_handle = pipe.MakeWriterHandle();
+
+  std::string python_code = absl::StrCat(
+      "import os; "
+      "os.write(",
+      writer_handle,
+      ", b'hello from child'); "
+      "os.close(",
+      writer_handle,
+      ")");
+
+  auto [proc, ec] = Process::Spawn({"python", "-c", python_code}, /*decouple=*/false);
+  ASSERT_FALSE(ec) << ec.message();
+
+  pipe.CloseWriterHandle();
+
+  auto result = pipe.Read(/*timeout_s=*/5);
+  ASSERT_TRUE(result.ok()) << result.status().message();
+  EXPECT_EQ(*result, "hello from child");
+
+  int exit_code = proc.Wait();
+  EXPECT_EQ(exit_code, 0);
+}
+
+TEST(PipeTest, SiblingProcessesCommunicate) {
+  Pipe pipe;
+
+  // Spawn writer first
+  intptr_t writer_handle = pipe.MakeWriterHandle();
+  std::string writer_code = absl::StrCat(
+      "import os; "
+      "os.write(",
+      writer_handle,
+      ", b'hello from sibling'); "
+      "os.close(",
+      writer_handle,
+      ")");
+
+  auto [writer_proc, writer_ec] =
+      Process::Spawn({"python", "-c", writer_code}, /*decouple=*/false);
+  ASSERT_FALSE(writer_ec) << writer_ec.message();
+  pipe.CloseWriterHandle();
+
+  // Then spawn reader
+  intptr_t reader_handle = pipe.MakeReaderHandle();
+  std::string reader_code = absl::StrCat(
+      "import os, sys; "
+      "data = os.read(",
+      reader_handle,
+      ", 100); "
+      "os.close(",
+      reader_handle,
+      "); "
+      "sys.exit(0 if data == b'hello from sibling' else 1)");
+
+  auto [reader_proc, reader_ec] =
+      Process::Spawn({"python", "-c", reader_code}, /*decouple=*/false);
+  ASSERT_FALSE(reader_ec) << reader_ec.message();
+  pipe.CloseReaderHandle();
+
+  int writer_exit = writer_proc.Wait();
+  EXPECT_EQ(writer_exit, 0);
+
+  int reader_exit = reader_proc.Wait();
+  EXPECT_EQ(reader_exit, 0);
+}
+
+TEST(PipeTest, ReadTimeout) {
+  Pipe pipe;
+  auto result = pipe.Read(/*timeout_s=*/0.1);
+  EXPECT_FALSE(result.ok());
+}
+
+TEST(PipeTest, OperationsAfterClose) {
+  Pipe pipe;
+  pipe.Close();
+
+  auto read_result = pipe.Read();
+  EXPECT_FALSE(read_result.ok());
+
+  Pipe pipe2;
+  pipe2.Close();
+  auto write_result = pipe2.Write("data");
+  EXPECT_FALSE(write_result.ok());
+}
+
+}  // namespace ray