create internal module for thridparty system compatible building

BUILD.bazel

@@ -772,6 +772,28 @@ ray_cc_library(
     ],
 )
 
+# This header is used to wrap some internal codes so we can reduce suspicious
+# symbols export.
+ray_cc_library(
+    name = "exported_internal",
+    srcs = glob(
+        [
+            "src/ray/internal/internal.cc",
+        ],
+    ),
+    hdrs = glob(
+        [
+            "src/ray/internal/internal.h",
+        ],
+    ),
+    copts = COPTS,
+    strip_include_prefix = "src",
+    deps = [
+        ":core_worker_lib",
+    ],
+    alwayslink = 1,
+)
+
 ray_cc_test(
     name = "core_worker_resubmit_queue_test",
     size = "small",

src/ray/internal/internal.cc

@@ -0,0 +1,80 @@
+// Copyright 2020 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/internal/internal.h"
+
+#include "ray/core_worker/core_worker.h"
+
+namespace ray {
+namespace internal {
+
+using ray::core::CoreWorkerProcess;
+using ray::core::TaskOptions;
+
+std::vector<rpc::ObjectReference> SendInternal(
+    const ActorID &peer_actor_id,
+    std::shared_ptr<LocalMemoryBuffer> buffer,
+    RayFunction &function,
+    int return_num,
+    int max_retries,
+    bool retry_exceptions,
+    std::string serialized_retry_exception_allowlist) {
+  std::unordered_map<std::string, double> resources;
+  std::string name = function.GetFunctionDescriptor()->DefaultTaskName();
+  TaskOptions options{name, return_num, resources};
+
+  char meta_data[3] = {'R', 'A', 'W'};
+  std::shared_ptr<LocalMemoryBuffer> meta =
+      std::make_shared<LocalMemoryBuffer>((uint8_t *)meta_data, 3, true);
+
+  std::vector<std::unique_ptr<TaskArg>> args;
+  if (function.GetLanguage() == Language::PYTHON) {
+    auto dummy = "__RAY_DUMMY__";
+    std::shared_ptr<LocalMemoryBuffer> dummyBuffer =
+        std::make_shared<LocalMemoryBuffer>((uint8_t *)dummy, 13, true);
+    args.emplace_back(new TaskArgByValue(std::make_shared<RayObject>(
+        std::move(dummyBuffer), meta, std::vector<rpc::ObjectReference>(), true)));
+  }
+  args.emplace_back(new TaskArgByValue(std::make_shared<RayObject>(
+      std::move(buffer), meta, std::vector<rpc::ObjectReference>(), true)));
+
+  std::vector<std::shared_ptr<RayObject>> results;
+  std::vector<rpc::ObjectReference> return_refs;
+  auto result = CoreWorkerProcess::GetCoreWorker().SubmitActorTask(
+      peer_actor_id,
+      function,
+      args,
+      options,
+      max_retries,
+      retry_exceptions,
+      serialized_retry_exception_allowlist,
+      return_refs);
+  if (!result.ok()) {
+    RAY_CHECK(false) << "Back pressure should not be enabled.";
+  }
+  return return_refs;
+}
+
+const ray::stats::TagKeyType TagRegister(const std::string tag_name) {
+  return ray::stats::TagKeyType::Register(tag_name);
+}
+
+const ActorID &GetCurrentActorID() {
+  return CoreWorkerProcess::GetCoreWorker().GetWorkerContext().GetCurrentActorID();
+}
+
+bool IsInitialized() { return CoreWorkerProcess::IsInitialized(); }
+
+}  // namespace internal
+}  // namespace ray

src/ray/internal/internal.h

@@ -0,0 +1,53 @@
+// Copyright 2020 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 "ray/common/buffer.h"
+#include "ray/common/id.h"
+#include "ray/core_worker/common.h"
+#include "ray/stats/metric.h"
+
+// This header is used to warp some internal code so we can reduce suspicious
+// symbols export.
+namespace ray {
+namespace internal {
+
+using ray::core::RayFunction;
+
+/// Send buffer internal
+/// \param[in] buffer buffer to be sent.
+/// \param[in] function the function descriptor of peer's function.
+/// \param[in] return_num return value number of the call.
+/// \param[in] max_retirs task retries time.
+/// \param[in] retry_execptions whether retry if execptions found.
+/// \param[in] serialized_retry_exception_allowlist specificed allowed exceptions.
+/// \param[out] return_ids return ids from SubmitActorTask.
+std::vector<rpc::ObjectReference> SendInternal(
+    const ActorID &peer_actor_id,
+    std::shared_ptr<LocalMemoryBuffer> buffer,
+    RayFunction &function,
+    int return_num,
+    int max_retries = -1,
+    bool retry_exceptions = false,
+    std::string serialized_retry_exception_allowlist = "");
+
+const stats::TagKeyType TagRegister(const std::string tag_name);
+
+/// Get current actor id via internal.
+const ActorID &GetCurrentActorID();
+
+/// Get core worker initialization flag via internal.
+bool IsInitialized();
+}  // namespace internal
+}  // namespace ray