[pull] master from ray-project:master

.buildkite/forge.rayci.yml

@@ -0,0 +1,4 @@
+group: forge
+steps:
+  - name: forge
+    wanda: ci/docker/forge.wanda.yaml

.buildkite/pipeline.ml.yml

@@ -53,6 +53,7 @@
     - ./ci/env/env_info.sh
     - bazel test --config=ci $(./ci/run/bazel_export_options) --build_tests_only --test_tag_filters=tune,-gpu_only,-ray_air,-gpu,-doctest python/ray/train/...
 
+
 - label: ":brain: RLlib: Benchmarks (Torch 2.x)"
   conditions: ["NO_WHEELS_REQUIRED", "RAY_CI_RLLIB_AFFECTED"]
   instance_size: medium
@@ -314,6 +315,7 @@
       --test_env=AIR_VERBOSITY=1
       python/ray/tune/...
 
+
 - label: ":octopus: :brain: Tune tests and examples {using RLlib}"
   conditions: ["NO_WHEELS_REQUIRED", "RAY_CI_TUNE_AFFECTED", "RAY_CI_RLLIB_AFFECTED"]
   instance_size: large
@@ -335,6 +337,71 @@
     - bazel test --config=ci $(./ci/run/bazel_export_options) --build_tests_only python/ray/tests/horovod/...
 
 
+
+##### STORAGE REFACTOR
+
+- label: ":steam_locomotive: :floppy_disk: New persistence mode: Train tests and examples"
+  conditions: ["NO_WHEELS_REQUIRED", "RAY_CI_TRAIN_AFFECTED"]
+  instance_size: large
+  parallelism: 4
+  commands:
+    - cleanup() { if [ "${BUILDKITE_PULL_REQUEST}" = "false" ]; then ./ci/build/upload_build_info.sh; fi }; trap cleanup EXIT
+    # Todo (krfricke): Move mosaicml to train-test-requirements.txt
+    - pip install "mosaicml==0.12.1"
+    - TRAIN_TESTING=1 DATA_PROCESSING_TESTING=1 INSTALL_HOROVOD=1 ./ci/env/install-dependencies.sh
+    - ./ci/env/env_info.sh
+    - ./ci/run/run_bazel_test_with_sharding.sh
+      --config=ci $(./ci/run/bazel_export_options)
+      --test_tag_filters=-gpu_only,-gpu,-minimal,-tune,-needs_credentials,-doctest,-no_new_storage
+      --test_env=RAY_AIR_NEW_PERSISTENCE_MODE=1
+      python/ray/train/...
+
+- label: ":steam_locomotive: :octopus: :floppy_disk: New persistence mode: Train + Tune tests and examples"
+  conditions: ["NO_WHEELS_REQUIRED", "RAY_CI_TRAIN_AFFECTED"]
+  instance_size: medium
+  commands:
+    - cleanup() { if [ "${BUILDKITE_PULL_REQUEST}" = "false" ]; then ./ci/build/upload_build_info.sh; fi }; trap cleanup EXIT
+    - TRAIN_TESTING=1 TUNE_TESTING=1 ./ci/env/install-dependencies.sh
+    - ./ci/env/env_info.sh
+    - bazel test --config=ci $(./ci/run/bazel_export_options) --build_tests_only 
+      --test_tag_filters=tune,-gpu_only,-ray_air,-gpu,-doctest,-no_new_storage
+      --test_env=RAY_AIR_NEW_PERSISTENCE_MODE=1
+      python/ray/train/...
+
+
+- label: ":octopus: :floppy_disk: New persistence mode: Tune tests and examples (small)"
+  conditions: ["NO_WHEELS_REQUIRED", "RAY_CI_TUNE_AFFECTED"]
+  instance_size: small
+  parallelism: 3
+  commands:
+    - cleanup() { if [ "${BUILDKITE_PULL_REQUEST}" = "false" ]; then ./ci/build/upload_build_info.sh; fi }; trap cleanup EXIT
+    - TUNE_TESTING=1 ./ci/env/install-dependencies.sh
+    - ./ci/env/env_info.sh
+    - ./ci/run/run_bazel_test_with_sharding.sh
+      --config=ci $(./ci/run/bazel_export_options) --build_tests_only
+      --test_tag_filters=-medium_instance,-soft_imports,-gpu_only,-rllib,-multinode,-no_new_storage
+      --test_env=RAY_AIR_NEW_PERSISTENCE_MODE=1
+      python/ray/tune/...
+
+- label: ":octopus: :floppy_disk: New persistence mode: Tune tests and examples (medium)"
+  conditions: ["NO_WHEELS_REQUIRED", "RAY_CI_TUNE_AFFECTED"]
+  instance_size: medium
+  commands:
+    - cleanup() { if [ "${BUILDKITE_PULL_REQUEST}" = "false" ]; then ./ci/build/upload_build_info.sh; fi }; trap cleanup EXIT
+    - TUNE_TESTING=1 DATA_PROCESSING_TESTING=1 ./ci/env/install-dependencies.sh
+    - ./ci/env/env_info.sh
+    - bazel test --config=ci $(./ci/run/bazel_export_options) --build_tests_only
+      --test_tag_filters=medium_instance,-soft_imports,-gpu_only,-rllib,-multinode,-no_new_storage
+      --test_env=RAY_AIR_NEW_PERSISTENCE_MODE=1
+      python/ray/tune/...
+
+
+###### END STORAGE REFACTOR
+
+
+
+
+
 # TODO(amogkam): Re-enable Ludwig tests after Ludwig supports Ray 2.0
 #- label: ":octopus: Ludwig tests and examples. Python 3.7"
 #  conditions: ["NO_WHEELS_REQUIRED", "RAY_CI_TUNE_AFFECTED"]

.github/CODEOWNERS

@@ -54,11 +54,11 @@
 #!/python/ray/rllib/ @ray-project/ray-core
 
 # Java worker.
-/java/dependencies.bzl @kfstorm @raulchen @ericl @iycheng @WangTaoTheTonic @SongGuyang @jovany-wang
-/java/pom.xml @kfstorm @raulchen @ericl @iycheng @WangTaoTheTonic @SongGuyang @jovany-wang
-/java/pom_template.xml @kfstorm @raulchen @ericl @iycheng @WangTaoTheTonic @SongGuyang @jovany-wang
-/java/*/pom_template.xml @kfstorm @raulchen @ericl @iycheng @WangTaoTheTonic @SongGuyang @jovany-wang
-/java/api/ @kfstorm @raulchen @ericl @iycheng @WangTaoTheTonic @SongGuyang @jovany-wang
+/java/dependencies.bzl @kfstorm @raulchen @ericl @iycheng @WangTaoTheTonic @SongGuyang
+/java/pom.xml @kfstorm @raulchen @ericl @iycheng @WangTaoTheTonic @SongGuyang
+/java/pom_template.xml @kfstorm @raulchen @ericl @iycheng @WangTaoTheTonic @SongGuyang
+/java/*/pom_template.xml @kfstorm @raulchen @ericl @iycheng @WangTaoTheTonic @SongGuyang
+/java/api/ @kfstorm @raulchen @ericl @iycheng @WangTaoTheTonic @SongGuyang
 
 # C++ worker
 /cpp/include/ray @SongGuyang @raulchen @kfstorm @ray-project/ray-core

ci/docker/forge.wanda.yaml

@@ -0,0 +1,4 @@
+name: "forge"
+froms:
+  - ubuntu:20.04
+dockerfile: ci/docker/forge.Dockerfile

cpp/include/ray/api/msgpack_adaptor.h

@@ -0,0 +1,88 @@
+// Copyright 2023 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 <any>
+#include <msgpack.hpp>
+
+// TODO(Larry Lian) Adapt on windows
+#ifndef _WIN32
+
+namespace msgpack {
+namespace adaptor {
+
+template <>
+struct pack<std::any> {
+  template <typename Stream>
+  msgpack::packer<Stream> &operator()(msgpack::packer<Stream> &o,
+                                      const std::any &v) const {
+    const auto &any_type = v.type();
+    if (any_type == typeid(msgpack::type::nil_t)) {
+      o.pack(std::any_cast<msgpack::type::nil_t>(v));
+    } else if (any_type == typeid(bool)) {
+      o.pack(std::any_cast<bool>(v));
+    } else if (any_type == typeid(uint64_t)) {
+      o.pack(std::any_cast<uint64_t>(v));
+    } else if (any_type == typeid(int64_t)) {
+      o.pack(std::any_cast<int64_t>(v));
+    } else if (any_type == typeid(double)) {
+      o.pack(std::any_cast<double>(v));
+    } else if (any_type == typeid(std::string)) {
+      o.pack(std::any_cast<std::string>(v));
+    } else if (any_type == typeid(std::vector<char>)) {
+      o.pack(std::any_cast<std::vector<char>>(v));
+    } else {
+      throw msgpack::type_error();
+    }
+    return o;
+  }
+};
+
+template <>
+struct convert<std::any> {
+  msgpack::object const &operator()(msgpack::object const &o, std::any &v) const {
+    switch (o.type) {
+    case type::NIL:
+      v = o.as<msgpack::type::nil_t>();
+      break;
+    case type::BOOLEAN:
+      v = o.as<bool>();
+      break;
+    case type::POSITIVE_INTEGER:
+      v = o.as<uint64_t>();
+      break;
+    case type::NEGATIVE_INTEGER:
+      v = o.as<int64_t>();
+      break;
+    case type::FLOAT32:
+    case type::FLOAT64:
+      v = o.as<double>();
+      break;
+    case type::STR:
+      v = o.as<std::string>();
+      break;
+    case type::BIN:
+      v = o.as<std::vector<char>>();
+      break;
+    default:
+      throw msgpack::type_error();
+    }
+    return o;
+  }
+};
+
+}  // namespace adaptor
+}  // namespace msgpack
+#endif

cpp/include/ray/api/serializer.h

@@ -14,6 +14,7 @@
 
 #pragma once
 
+#include <ray/api/msgpack_adaptor.h>
 #include <ray/api/ray_exception.h>
 #include <ray/api/xlang_function.h>
 

cpp/src/ray/test/cluster/counter.cc

@@ -127,9 +127,11 @@ RAY_REMOTE(RAY_FUNC(Counter::FactoryCreate),
            &Counter::GetCount,
            &Counter::CreateNestedChildActor,
            &Counter::GetBytes,
-           &Counter::echoBytes,
-           &Counter::echoString,
-           &Counter::GetIntByObjectRef);
+           &Counter::EchoBytes,
+           &Counter::EchoString,
+           &Counter::GetIntByObjectRef,
+           &Counter::EchoStrings,
+           &Counter::EchoAnyArray);
 
 RAY_REMOTE(ActorConcurrentCall::FactoryCreate, &ActorConcurrentCall::CountDown);
 

cpp/src/ray/test/cluster/counter.h

@@ -16,6 +16,7 @@
 
 #include <ray/api.h>
 
+#include <cassert>
 #include <condition_variable>
 #include <mutex>
 
@@ -57,9 +58,52 @@ class Counter {
     return bytes;
   }
 
-  std::vector<std::byte> echoBytes(const std::vector<std::byte> &bytes) { return bytes; }
+  std::vector<std::byte> EchoBytes(const std::vector<std::byte> &bytes) { return bytes; }
 
-  std::string echoString(const std::string &str) { return str; }
+  std::string EchoString(const std::string &str) { return str; }
+
+  std::vector<std::string> EchoStrings(const std::vector<std::string> &strings) {
+    return strings;
+  }
+
+  std::vector<std::any> EchoAnyArray(const std::vector<std::any> &anys) {
+    assert(anys.size() == 11);
+    // check bool
+    assert(anys[0].type() == typeid(bool));
+    assert(std::any_cast<bool>(anys[0]) == true);
+    // equal to java Byte.MAX_VALUE
+    assert(anys[1].type() == typeid(uint64_t));
+    assert(std::any_cast<uint64_t>(anys[1]) == 127);
+    // equal to java Short.MAX_VALUE
+    assert(anys[2].type() == typeid(uint64_t));
+    assert(std::any_cast<uint64_t>(anys[2]) == 32767);
+    // equal to java Integer.MAX_VALUE
+    assert(anys[3].type() == typeid(uint64_t));
+    assert(std::any_cast<uint64_t>(anys[3]) == 2147483647);
+    // equal to java Short.MAX_VALUE
+    assert(anys[4].type() == typeid(uint64_t));
+    assert(std::any_cast<uint64_t>(anys[4]) == 9223372036854775807);
+    // equal to java Long.MIN_VALUE
+    assert(anys[5].type() == typeid(int64_t));
+    assert(std::any_cast<int64_t>(anys[5]) == -9223372036854775808);
+    // equal to java BigInteger.valueOf(Long.MAX_VALUE)
+    assert(anys[6].type() == typeid(uint64_t));
+    assert(std::any_cast<uint64_t>(anys[6]) == 9223372036854775807);
+    // equal to java string "Hello World!"
+    assert(anys[7].type() == typeid(std::string));
+    assert(std::any_cast<std::string>(anys[7]) == "Hello World!");
+    // equal to java float 1.234f
+    assert(anys[8].type() == typeid(double));
+    assert(std::any_cast<double>(anys[8]) == 1.234);
+    // equal to java double 1.234
+    assert(anys[9].type() == typeid(double));
+    assert(std::any_cast<double>(anys[9]) == 1.234);
+    // equal to java byte[]
+    std::vector<char> bytes = {'b', 'i', 'n', 'a', 'r', 'y'};
+    assert(anys[10].type() == typeid(std::vector<std::byte>));
+    assert(std::any_cast<std::vector<char>>(anys[10]) == bytes);
+    return anys;
+  }
 
   int GetIntVal(ray::ObjectRef<ray::ObjectRef<int>> obj) {
     auto val = *obj.Get();

cpp/src/ray/test/serialization_test.cc

@@ -15,9 +15,38 @@
 #include <gtest/gtest.h>
 #include <ray/api.h>
 
+bool IsAnyEqual(const std::any &lhs, const std::any &rhs) {
+  if (lhs.type() != rhs.type()) {
+    return false;
+  }
+  if (lhs.type() == typeid(uint64_t)) {
+    return std::any_cast<uint64_t>(lhs) == std::any_cast<uint64_t>(rhs);
+  } else if (lhs.type() == typeid(int64_t)) {
+    return std::any_cast<int64_t>(lhs) == std::any_cast<int64_t>(rhs);
+  } else if (lhs.type() == typeid(bool)) {
+    return std::any_cast<bool>(lhs) == std::any_cast<bool>(rhs);
+  } else if (lhs.type() == typeid(float)) {
+    return std::any_cast<float>(lhs) == std::any_cast<float>(rhs);
+  } else if (lhs.type() == typeid(double)) {
+    return std::any_cast<double>(lhs) == std::any_cast<double>(rhs);
+  } else if (lhs.type() == typeid(std::string)) {
+    return std::any_cast<std::string>(lhs) == std::any_cast<std::string>(rhs);
+  } else if (lhs.type() == typeid(std::vector<char>)) {
+    return std::any_cast<std::vector<char>>(lhs) == std::any_cast<std::vector<char>>(rhs);
+  } else {
+    return false;
+  }
+}
+
 TEST(SerializationTest, TypeHybridTest) {
   uint32_t in_arg1 = 123456789, out_arg1;
   std::string in_arg2 = "123567ABC", out_arg2;
+  double in_arg3 = 1.1;
+  float in_arg4 = 1.2;
+  uint64_t in_arg5 = 100;
+  int64_t in_arg6 = -100;
+  bool in_arg7 = false;
+  std::vector<char> in_arg8 = {'a', 'b'};
 
   // 1 arg
   // marshall
@@ -40,6 +69,40 @@ TEST(SerializationTest, TypeHybridTest) {
 
   EXPECT_EQ(in_arg1, out_arg1);
   EXPECT_EQ(in_arg2, out_arg2);
+
+// TODO(Larry Lian) Adapt on windows
+#ifndef _WIN32
+  // Test std::any
+  msgpack::sbuffer buffer3 = ray::internal::Serializer::Serialize(std::make_tuple(
+      in_arg1, in_arg2, in_arg3, in_arg4, in_arg5, in_arg6, in_arg7, in_arg8));
+
+  std::vector<std::any> result =
+      ray::internal::Serializer::Deserialize<std::vector<std::any>>(buffer3.data(),
+                                                                    buffer3.size());
+  EXPECT_TRUE(result[0].type() == typeid(uint64_t));
+  EXPECT_EQ(std::any_cast<uint64_t>(result[0]), in_arg1);
+  EXPECT_TRUE(result[1].type() == typeid(std::string));
+  EXPECT_EQ(std::any_cast<std::string>(result[1]), in_arg2);
+  EXPECT_TRUE(result[2].type() == typeid(double));
+  EXPECT_EQ(std::any_cast<double>(result[2]), in_arg3);
+  EXPECT_TRUE(result[3].type() == typeid(double));
+  EXPECT_EQ(std::any_cast<double>(result[3]), in_arg4);
+  EXPECT_TRUE(result[4].type() == typeid(uint64_t));
+  EXPECT_EQ(std::any_cast<uint64_t>(result[4]), in_arg5);
+  EXPECT_TRUE(result[5].type() == typeid(int64_t));
+  EXPECT_EQ(std::any_cast<int64_t>(result[5]), in_arg6);
+  EXPECT_TRUE(result[6].type() == typeid(bool));
+  EXPECT_EQ(std::any_cast<bool>(result[6]), in_arg7);
+  EXPECT_TRUE(result[7].type() == typeid(std::vector<char>));
+  EXPECT_EQ(std::any_cast<std::vector<char>>(result[7]), in_arg8);
+
+  msgpack::sbuffer buffer4 = ray::internal::Serializer::Serialize(result);
+
+  std::vector<std::any> result_2 =
+      ray::internal::Serializer::Deserialize<std::vector<std::any>>(buffer4.data(),
+                                                                    buffer4.size());
+  EXPECT_TRUE(std::equal(result.begin(), result.end(), result_2.begin(), IsAnyEqual));
+#endif
 }
 
 TEST(SerializationTest, BoundaryValueTest) {

doc/source/_static/js/top-navigation.js

@@ -22,7 +22,6 @@ is_examples = window.location.href.endsWith("ray-overview/examples.html")
 is_get_started = window.location.href.endsWith("ray-overview/getting-started.html")
 is_use_cases = window.location.href.endsWith("ray-overview/use-cases.html")
 is_libraries = window.location.href.includes("/ray-core/") ||
-    window.location.href.includes("/ray-air/") ||
     window.location.href.includes("/data/") ||
     window.location.href.includes("/train/") ||
     window.location.href.includes("/tune/") ||
@@ -89,7 +88,6 @@ librariesMenu.setAttribute("class", "menu")
 librariesMenu.innerHTML = "<a href='#'>Libraries" + downCaret + "</a>"
 librariesList = document.createElement("ul")
 librariesList.innerHTML += "<li><a href='" + getNavURL("ray-core/walkthrough.html") + "'><span class='primary'>Ray Core</span><span class='secondary'>Scale general Python applications</span></a></li>"
-librariesList.innerHTML += "<li><a href='" + getNavURL("ray-air/getting-started.html") + "'><span class='primary'>Ray AIR</span><span class='secondary'>Scale AI applications</span></a></li>"
 librariesList.innerHTML += "<li><a href='" + getNavURL("data/data.html") + "'><span class='primary'>Ray Data</span><span class='secondary'>Scale data ingest and preprocessing</span></a></li>"
 librariesList.innerHTML += "<li><a href='" + getNavURL("train/train.html") + "'><span class='primary'>Ray Train</span><span class='secondary'>Scale machine learning training</span></a></li>"
 librariesList.innerHTML += "<li><a href='" + getNavURL("tune/index.html") + "'><span class='primary'>Ray Tune</span><span class='secondary'>Scale hyperparameter tuning</span></a></li>"