Revert "[Streaming] fix streaming ci (#9675)"

BUILD.bazel

@@ -367,29 +367,6 @@ cc_library(
     ],
 )
 
-# This header is used to warp some streaming code so we can reduce suspicious
-# symbols export.
-cc_library(
-    name = "exported_streaming_internal",
-    srcs = glob(
-        [
-            "src/ray/streaming/streaming.cc",
-        ],
-    ),
-    hdrs = glob(
-        [
-            "src/ray/streaming/streaming.h",
-        ],
-    ),
-    copts = COPTS,
-    strip_include_prefix = "src",
-    visibility = ["//visibility:public"],
-    deps = [
-        ":core_worker_lib",
-    ],
-    alwayslink = 1,
-)
-
 cc_binary(
     name = "raylet",
     srcs = ["src/ray/raylet/main.cc"],
@@ -1725,7 +1702,6 @@ pyx_library(
     ),
     deps = [
         "//:core_worker_lib",
-        "//:exported_streaming_internal",
         "//:global_state_accessor_lib",
         "//:ray_util",
         "//:raylet_lib",
@@ -1780,7 +1756,6 @@ cc_binary(
     visibility = ["//java:__subpackages__"],
     deps = [
         "//:core_worker_lib",
-        "//:exported_streaming_internal",
         "//:global_state_accessor_lib",
         "//:src/ray/ray_exported_symbols.lds",
         "//:src/ray/ray_version_script.lds",

src/ray/common/id.h

@@ -75,8 +75,7 @@ class BaseID {
  protected:
   BaseID(const std::string &binary) {
     RAY_CHECK(binary.size() == Size() || binary.size() == 0)
-        << "expected size is " << Size() << ", but got data " << binary << " of size "
-        << binary.size();
+        << "expected size is " << Size() << ", but got " << binary.size();
     std::memcpy(const_cast<uint8_t *>(this->Data()), binary.data(), binary.size());
   }
   // All IDs are immutable for hash evaluations. MutableData is only allow to use
@@ -342,25 +341,24 @@ std::ostream &operator<<(std::ostream &os, const TaskID &id);
 std::ostream &operator<<(std::ostream &os, const ObjectID &id);
 std::ostream &operator<<(std::ostream &os, const PlacementGroupID &id);
 
-#define DEFINE_UNIQUE_ID(type)                                                           \
-  class RAY_EXPORT type : public UniqueID {                                              \
-   public:                                                                               \
-    explicit type(const UniqueID &from) {                                                \
-      std::memcpy(&id_, from.Data(), kUniqueIDSize);                                     \
-    }                                                                                    \
-    type() : UniqueID() {}                                                               \
-    static type FromRandom() { return type(UniqueID::FromRandom()); }                    \
-    static type FromBinary(const std::string &binary) { return type(binary); }           \
-    static type Nil() { return type(UniqueID::Nil()); }                                  \
-    static size_t Size() { return kUniqueIDSize; }                                       \
-                                                                                         \
-   private:                                                                              \
-    explicit type(const std::string &binary) {                                           \
-      RAY_CHECK(binary.size() == Size() || binary.size() == 0)                           \
-          << "expected size is " << Size() << ", but got data " << binary << " of size " \
-          << binary.size();                                                              \
-      std::memcpy(&id_, binary.data(), binary.size());                                   \
-    }                                                                                    \
+#define DEFINE_UNIQUE_ID(type)                                                 \
+  class RAY_EXPORT type : public UniqueID {                                    \
+   public:                                                                     \
+    explicit type(const UniqueID &from) {                                      \
+      std::memcpy(&id_, from.Data(), kUniqueIDSize);                           \
+    }                                                                          \
+    type() : UniqueID() {}                                                     \
+    static type FromRandom() { return type(UniqueID::FromRandom()); }          \
+    static type FromBinary(const std::string &binary) { return type(binary); } \
+    static type Nil() { return type(UniqueID::Nil()); }                        \
+    static size_t Size() { return kUniqueIDSize; }                             \
+                                                                               \
+   private:                                                                    \
+    explicit type(const std::string &binary) {                                 \
+      RAY_CHECK(binary.size() == Size() || binary.size() == 0)                 \
+          << "expected size is " << Size() << ", but got " << binary.size();   \
+      std::memcpy(&id_, binary.data(), binary.size());                         \
+    }                                                                          \
   };
 
 #include "ray/common/id_def.h"
@@ -387,8 +385,7 @@ T BaseID<T>::FromRandom() {
 template <typename T>
 T BaseID<T>::FromBinary(const std::string &binary) {
   RAY_CHECK(binary.size() == T::Size() || binary.size() == 0)
-      << "expected size is " << T::Size() << ", but got data " << binary << " of size "
-      << binary.size();
+      << "expected size is " << T::Size() << ", but got " << binary.size();
   T t;
   std::memcpy(t.MutableData(), binary.data(), binary.size());
   return t;

src/ray/common/task/task_spec.cc

@@ -120,7 +120,7 @@ ObjectID TaskSpecification::ReturnId(size_t return_index) const {
 }
 
 bool TaskSpecification::ArgByRef(size_t arg_index) const {
-  return message_->args(arg_index).has_object_ref();
+  return message_->args(arg_index).object_ref().object_id() != "";
 }
 
 ObjectID TaskSpecification::ArgId(size_t arg_index) const {

src/ray/common/task/task_util.h

@@ -41,7 +41,7 @@ class TaskArgByValue : public TaskArg {
   ///
   /// \param[in] value Value of the argument.
   /// \return The task argument.
-  explicit TaskArgByValue(const std::shared_ptr<RayObject> &value) : value_(value) {
+  TaskArgByValue(const std::shared_ptr<RayObject> &value) : value_(value) {
     RAY_CHECK(value) << "Value can't be null.";
   }
 

src/ray/core_worker/transport/dependency_resolver.cc

@@ -46,7 +46,7 @@ void InlineDependencies(
         if (!it->second->IsInPlasmaError()) {
           // The object has not been promoted to plasma. Inline the object by
           // clearing the reference and replacing it with the raw value.
-          mutable_arg->clear_object_ref();
+          mutable_arg->mutable_object_ref()->Clear();
           if (it->second->HasData()) {
             const auto &data = it->second->GetData();
             mutable_arg->set_data(data->Data(), data->Size());

src/ray/ray_exported_symbols.lds

@@ -21,11 +21,9 @@
 *ray*TaskID*
 *ray*ActorID*
 *ray*ObjectID*
-*ray*ObjectReference*
 # Others
 *ray*CoreWorker*
 *PyInit*
 *init_raylet*
 *Java*
 *JNI_*
-*ray*streaming*

src/ray/ray_version_script.lds

@@ -23,13 +23,11 @@ VERSION_1.0 {
         *ray*TaskID*;
         *ray*ActorID*;
         *ray*ObjectID*;
-        *ray*ObjectReference*;
         # Others
         *ray*CoreWorker*;
         *PyInit*;
         *init_raylet*;
         *Java*;
         *JNI_*;
-        *ray*streaming*;
     local: *;
 };

streaming/BUILD.bazel

@@ -67,13 +67,6 @@ cc_binary(
     deps = ["//:core_worker_lib"],
 )
 
-cc_binary(
-    name = "exported_streaming_internal.so",
-    copts = COPTS,
-    linkshared = 1,
-    deps = ["//:exported_streaming_internal"],
-)
-
 cc_library(
     name = "streaming_util",
     srcs = glob([
@@ -153,7 +146,6 @@ cc_library(
         ],
         "//conditions:default": [
             "core_worker_lib.so",
-            "exported_streaming_internal.so",
         ],
     }),
 )
@@ -241,7 +233,6 @@ cc_library(
 )
 
 test_common_deps = [
-    "//:exported_streaming_internal",
     ":streaming_lib",
     "//:ray_common",
     "//:ray_util",
@@ -357,7 +348,6 @@ genrule(
         GENERATED_DIR="streaming/python/generated"
         mkdir -p "$$GENERATED_DIR"
         touch "$$GENERATED_DIR/__init__.py"
-        # Use this `sed` command to change the import path in the generated file.
         sed -i -E 's/from streaming.src.protobuf/from ./' "$$GENERATED_DIR/remote_call_pb2.py"
         sed -i -E 's/from protobuf/from ./' "$$GENERATED_DIR/remote_call_pb2.py"
         date > $@

streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/util/EnvUtil.java

@@ -5,8 +5,6 @@
 import java.lang.management.ManagementFactory;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
-import java.util.List;
-import java.util.concurrent.TimeUnit;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -39,25 +37,4 @@ public static void loadNativeLibraries() {
     JniUtils.loadLibrary("streaming_java");
   }
 
-  /**
-   * Execute an external command.
-   *
-   * @return Whether the command succeeded.
-   */
-  public static boolean executeCommand(List<String> command, int waitTimeoutSeconds) {
-    try {
-      ProcessBuilder processBuilder = new ProcessBuilder(command)
-          .redirectOutput(ProcessBuilder.Redirect.INHERIT)
-          .redirectError(ProcessBuilder.Redirect.INHERIT);
-      Process process = processBuilder.start();
-      boolean exit = process.waitFor(waitTimeoutSeconds, TimeUnit.SECONDS);
-      if (!exit) {
-        process.destroyForcibly();
-      }
-      return process.exitValue() == 0;
-    } catch (Exception e) {
-      throw new RuntimeException("Error executing command " + String.join(" ", command), e);
-    }
-  }
-
 }

streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/util/ResourceUtil.java

@@ -1,6 +1,7 @@
 package io.ray.streaming.runtime.util;
 
 import com.sun.management.OperatingSystemMXBean;
+import io.ray.api.id.UniqueId;
 import io.ray.streaming.runtime.core.resource.Container;
 import io.ray.streaming.runtime.core.resource.ContainerId;
 import java.io.BufferedInputStream;

streaming/java/streaming-runtime/src/test/java/io/ray/streaming/runtime/demo/HybridStreamTest.java

@@ -1,14 +1,11 @@
 package io.ray.streaming.runtime.demo;
 
-import com.google.common.base.Preconditions;
-import com.google.common.collect.ImmutableList;
 import io.ray.api.Ray;
 import io.ray.streaming.api.context.StreamingContext;
 import io.ray.streaming.api.function.impl.FilterFunction;
 import io.ray.streaming.api.function.impl.MapFunction;
 import io.ray.streaming.api.function.impl.SinkFunction;
 import io.ray.streaming.api.stream.DataStreamSource;
-import io.ray.streaming.runtime.util.EnvUtil;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Paths;
@@ -45,8 +42,6 @@ public boolean filter(Object value) throws Exception {
   @Test(timeOut = 60000)
   public void testHybridDataStream() throws Exception {
     Ray.shutdown();
-    Preconditions.checkArgument(
-        EnvUtil.executeCommand(ImmutableList.of("ray", "stop"), 5));
     String sinkFileName = "/tmp/testHybridDataStream.txt";
     Files.deleteIfExists(Paths.get(sinkFileName));
 

streaming/python/tests/test_hybrid_stream.py

@@ -1,10 +1,8 @@
 import json
-import os
-import subprocess
-
 import ray
 from ray.streaming import StreamingContext
-from ray.test_utils import wait_for_condition
+import subprocess
+import os
 
 
 def map_func1(x):
@@ -34,9 +32,9 @@ def test_hybrid_stream():
     print("java_worker_options", java_worker_options)
     assert not ray.is_initialized()
     ray.init(
-        _load_code_from_local=True,
-        _include_java=True,
-        _java_worker_options=java_worker_options,
+        load_code_from_local=True,
+        include_java=True,
+        java_worker_options=java_worker_options,
         _system_config={"num_workers_per_process_java": 1})
 
     sink_file = "/tmp/ray_streaming_test_hybrid_stream.txt"
@@ -47,7 +45,6 @@ def sink_func(x):
         print("HybridStreamTest", x)
         with open(sink_file, "a") as f:
             f.write(str(x))
-            f.flush()
 
     ctx = StreamingContext.Builder().build()
     ctx.from_values("a", "b", "c") \
@@ -57,23 +54,14 @@ def sink_func(x):
         .as_python_stream() \
         .sink(sink_func)
     ctx.submit("HybridStreamTest")
-
-    def check_succeed():
-        if os.path.exists(sink_file):
-            import time
-            time.sleep(3)  # Wait all data be written
-            with open(sink_file, "r") as f:
-                result = f.read()
-                assert "a" in result
-                assert "b" not in result
-                assert "c" in result
-            print("Execution succeed")
-            return True
-        return False
-
-    wait_for_condition(check_succeed, timeout=60, retry_interval_ms=1000)
-    print("Execution succeed")
+    import time
+    time.sleep(3)
     ray.shutdown()
+    with open(sink_file, "r") as f:
+        result = f.read()
+        assert "a" in result
+        assert "b" not in result
+        assert "c" in result
 
 
 if __name__ == "__main__":