Repository: ray-project/ray
File: src/ray/core_worker/core_worker.cc (C++)

Code before:
      func_name, rpc::TaskStatus::GETTING_AND_PINNING_ARGS, is_retry);
  --num_get_pin_args_in_flight_;
  if (!pin_args_request_status.ok()) {
    ++num_failed_get_pin_args_;
    // If this has happened, it's because we are unable to talk to our local raylet.
    // This very likely means that the raylet has shutdown before this worker
    // unexpectedly. In whic case we'll trigger shut down.
    Exit(rpc::WorkerExitType::SYSTEM_ERROR,
         absl::StrCat("Worker failed to get and pin task arguments! Error message: ",
                      pin_args_request_status.message()),
         nullptr);
    return pin_args_request_status;
  }

  task_queue_length_ -= 1;
  num_executed_tasks_ += 1;

  // Modify the worker's per function counters.
  std::string actor_repr_name;
  {
    absl::MutexLock lock(&mutex_);
    actor_repr_name = actor_repr_name_;
  }
  if (!options_.is_local_mode) {
    task_counter_.MovePendingToRunning(func_name, is_retry);

    const auto update =
        (task_spec.IsActorTask() && !actor_repr_name.empty())
            ? worker::TaskStatusEvent::TaskStateUpdate(actor_repr_name, pid_)
            : worker::TaskStatusEvent::TaskStateUpdate(pid_);
    RAY_UNUSED(
        task_event_buffer_->RecordTaskStatusEventIfNeeded(task_spec.TaskId(),
                                                          task_spec.JobId(),
                                                          task_spec.AttemptNumber(),
                                                          task_spec,
                                                          rpc::TaskStatus::RUNNING,
                                                          /*include_task_info=*/false,
                                                          update));

    worker_context_->SetCurrentTask(task_spec);
    SetCurrentTaskId(task_spec.TaskId(), task_spec.AttemptNumber(), task_spec.GetName());
  }
  {
    absl::MutexLock lock(&mutex_);
    running_tasks_.emplace(task_spec.TaskId(), task_spec);
    if (resource_ids.has_value()) {
      resource_ids_ = std::move(*resource_ids);
    }
  }

  RayFunction func{task_spec.GetLanguage(), task_spec.FunctionDescriptor()};

Review comment (performance):
LGTM!

Previously, `actor_repr_name` was set and store in both `CoreWorker::actor_repr_name_` and `TaskReceiver::actor_repr_name_` member variables.

After this PR, both CoreWorker and TaskReceiver receive the value via an output parameter. CoreWorker still caches it in member variable (for task events), while TaskReceiver simply passes it to the `PushTaskReply` for GCS (used by observability features like the Dashboard). This makes TaskReceiver stateless.

The only nit is I think we don't need to acquire the lock here since all are executed serially. I also noticed that in this PR, when CoreWorker caches `actor_repr_name` in the member variable, there's no lock either — so we might want to be consistent.



Code after:
      GetAndPinArgsForExecutor(task_spec, &args, &arg_refs, &borrowed_ids);
  task_counter_.UnsetMetricStatus(
      func_name, rpc::TaskStatus::GETTING_AND_PINNING_ARGS, is_retry);
  --num_get_pin_args_in_flight_;
  if (!pin_args_request_status.ok()) {
    ++num_failed_get_pin_args_;
    // If this has happened, it's because we are unable to talk to our local raylet.
    // This very likely means that the raylet has shutdown before this worker
    // unexpectedly. In whic case we'll trigger shut down.
    Exit(rpc::WorkerExitType::SYSTEM_ERROR,
         absl::StrCat("Worker failed to get and pin task arguments! Error message: ",
                      pin_args_request_status.message()),
         nullptr);
    return pin_args_request_status;
  }

  task_queue_length_ -= 1;
  num_executed_tasks_ += 1;

  if (!options_.is_local_mode) {
    // Modify the worker's per-function counters.
    task_counter_.MovePendingToRunning(func_name, is_retry);

    worker::TaskStatusEvent::TaskStateUpdate update;
    {
      absl::MutexLock lock(&mutex_);
      update = (task_spec.IsActorTask() && !actor_repr_name_.empty())
                   ? worker::TaskStatusEvent::TaskStateUpdate(actor_repr_name_, pid_)
                   : worker::TaskStatusEvent::TaskStateUpdate(pid_);
    }

    RAY_UNUSED(
        task_event_buffer_->RecordTaskStatusEventIfNeeded(task_spec.TaskId(),
                                                          task_spec.JobId(),
                                                          task_spec.AttemptNumber(),
                                                          task_spec,
                                                          rpc::TaskStatus::RUNNING,
                                                          /*include_task_info=*/false,
                                                          update));

    worker_context_->SetCurrentTask(task_spec);
    SetCurrentTaskId(task_spec.TaskId(), task_spec.AttemptNumber(), task_spec.GetName());
  }

  {
    absl::MutexLock lock(&mutex_);
    running_tasks_.emplace(task_spec.TaskId(), task_spec);
    if (resource_ids.has_value()) {
      resource_ids_ = std::move(*resource_ids);
    }
  }