[Transfer]update ray version and use internal core worker api

.github/workflows/ubuntu_building.yml

@@ -28,7 +28,7 @@ jobs:
            python3 -m virtualenv -p python3 py3
            . py3/bin/activate
            pip install pytest
-           pip install https://ray-mobius-us.oss-us-west-1.aliyuncs.com/ci/linux/ubuntu/0bb82f29b65dca348acf5aa516d21ef3f176a3e1/ray-2.0.0.dev0-cp38-cp38-linux_x86_64.whl
+           sh -c "bash scripts/install-ray.sh"
     - name: Streaming python test
       run: sh -c "bash streaming/buildtest.sh --test_categories=streaming_python"
 

scripts/install-ray.sh

@@ -3,7 +3,7 @@
 minor_version=`python -c "import sys;print(sys.version_info[1])"`
 
 if [ "$minor_version" == "8" ] ; then
-	python -m pip install https://ray-mobius-us.oss-us-west-1.aliyuncs.com/ci/linux/ubuntu/0bb82f29b65dca348acf5aa516d21ef3f176a3e1/ray-2.0.0.dev0-cp38-cp38-linux_x86_64.whl
+	python -m pip install https://ray-mobius-us.oss-us-west-1.aliyuncs.com/ci/linux/ubuntu/b7d148815e427ccd47046d1c69ee3ab55bf7db3c/ray-2.0.0.dev0-cp38-cp38-linux_x86_64.whl
 elif [ "$minor_version" == "7" ] ; then
 	python -m pip install https://ray-mobius-us.oss-us-west-1.aliyuncs.com/ci/linux/ubuntu/0bb82f29b65dca348acf5aa516d21ef3f176a3e1/ray-2.0.0.dev0-cp37-cp37m-linux_x86_64.whl
 elif [ "$minor_version" == "6"] ; then

streaming/WORKSPACE

@@ -2,12 +2,12 @@ workspace(name = "com_github_ray_streaming")
 
 # LOAD RAY WORKSPACE
 load("@bazel_tools//tools/build_defs/repo:http.bzl", "http_archive", "http_file")
-ray_version = "0bb82f29b65dca348acf5aa516d21ef3f176a3e1"
+ray_version = "b7d148815e427ccd47046d1c69ee3ab55bf7db3c"
 http_archive(
         name="com_github_ray_project_ray",
         strip_prefix = "ray-{}".format(ray_version),
         urls = ["https://github.com/ray-project/ray/archive/{}.zip".format(ray_version)],
-        sha256 = "56b0bf6f0d5506be4bff694e9a3bedc6c8f3373b0ecd9f20fc2a25aed5efc569",
+        sha256 = "5de3e76a6208c4fc56f7f8a80275705d47b59124947865aa90b1ada96432c522",
 )
 
 http_archive(

streaming/src/event_service.cc

@@ -3,6 +3,8 @@
 #include <chrono>
 #include <unordered_set>
 
+#include "ray/internal/internal.h"
+
 namespace ray {
 namespace streaming {
 
@@ -117,7 +119,7 @@ Event &EventQueue::Front() {
 }
 
 EventService::EventService(uint32_t event_size)
-    : worker_id_(CoreWorkerProcess::IsInitialized()
+    : worker_id_(ray::internal::IsInitialized()
                      ? CoreWorkerProcess::GetCoreWorker().GetWorkerID()
                      : WorkerID::Nil()),
       event_queue_(std::make_shared<EventQueue>(event_size)),
@@ -170,8 +172,8 @@ void EventService::Execute(Event &event) {
 }
 
 void EventService::LoopThreadHandler() {
-  if (CoreWorkerProcess::IsInitialized()) {
-    CoreWorkerProcess::SetCurrentThreadWorkerId(worker_id_);
+  if (ray::internal::IsInitialized()) {
+    ray::internal::SetCurrentThreadWorker(worker_id_);
   }
   while (true) {
     if (stop_flag_) {

streaming/src/queue/queue_client.h

@@ -1,5 +1,7 @@
 #pragma once
 
+#include "ray/internal/internal.h"
+
 #include "queue/queue_handler.h"
 #include "queue/transport.h"
 
@@ -19,7 +21,7 @@ class ReaderClient {
   /// function descriptor to be called by DataWriter, synchronous semantics
   ReaderClient() {
     downstream_handler_ = ray::streaming::DownstreamQueueMessageHandler::CreateService(
-        CoreWorkerProcess::GetCoreWorker().GetWorkerContext().GetCurrentActorID());
+        ray::internal::GetCurrentActorID());
   }
 
   /// Post buffer to downstream queue service, asynchronously.
@@ -38,7 +40,7 @@ class WriterClient {
  public:
   WriterClient() {
     upstream_handler_ = ray::streaming::UpstreamQueueMessageHandler::CreateService(
-        CoreWorkerProcess::GetCoreWorker().GetWorkerContext().GetCurrentActorID());
+        ray::internal::GetCurrentActorID());
   }
 
   void OnWriterMessage(std::shared_ptr<LocalMemoryBuffer> buffer);