[Core] Remove actor_register_async config (#46998)

Signed-off-by: Jiajun Yao <jeromeyjj@gmail.com>

src/ray/common/ray_config_def.h

@@ -790,13 +790,6 @@ RAY_CONFIG(bool, event_log_reporter_enabled, true)
 /// This has no effect if `event_log_reporter_enabled` is false.
 RAY_CONFIG(bool, emit_event_to_log_file, false)
 
-/// Whether to enable register actor async.
-/// If it is false, the actor registration to GCS becomes synchronous, i.e.,
-/// core worker is blocked until GCS registers the actor and replies to it.
-/// If it is true, the actor registration is async, but actor handles cannot
-/// be passed to other worker until it is registered to GCS.
-RAY_CONFIG(bool, actor_register_async, true)
-
 /// Event severity threshold value
 RAY_CONFIG(std::string, event_level, "warning")
 

src/ray/core_worker/actor_creator.h

@@ -80,25 +80,20 @@ class DefaultActorCreator : public ActorCreatorInterface {
 
   Status AsyncRegisterActor(const TaskSpecification &task_spec,
                             gcs::StatusCallback callback) override {
-    if (::RayConfig::instance().actor_register_async()) {
-      auto actor_id = task_spec.ActorCreationId();
-      (*registering_actors_)[actor_id] = {};
-      if (callback != nullptr) {
-        (*registering_actors_)[actor_id].emplace_back(std::move(callback));
-      }
-      return gcs_client_->Actors().AsyncRegisterActor(
-          task_spec, [actor_id, this](Status status) {
-            std::vector<ray::gcs::StatusCallback> cbs;
-            cbs = std::move((*registering_actors_)[actor_id]);
-            registering_actors_->erase(actor_id);
-            for (auto &cb : cbs) {
-              cb(status);
-            }
-          });
-    } else {
-      callback(RegisterActor(task_spec));
-      return Status::OK();
+    auto actor_id = task_spec.ActorCreationId();
+    (*registering_actors_)[actor_id] = {};
+    if (callback != nullptr) {
+      (*registering_actors_)[actor_id].emplace_back(std::move(callback));
     }
+    return gcs_client_->Actors().AsyncRegisterActor(
+        task_spec, [actor_id, this](Status status) {
+          std::vector<ray::gcs::StatusCallback> cbs;
+          cbs = std::move((*registering_actors_)[actor_id]);
+          registering_actors_->erase(actor_id);
+          for (auto &cb : cbs) {
+            cb(status);
+          }
+        });
   }
 
   bool IsActorInRegistering(const ActorID &actor_id) const override {