[Train] Update run status and actor status for train runs.

[woshiyyya]

Why are these changes needed?

Added two entries for TrainRunInfo:

• TrainRunInfo.run_status: STARTED, FINISHED, ERRORED
• TrainRunInfo.controller_actor_status: ALIVE, DEAD
• WorkerInfo.status: ALIVE, DEAD

Update Train run status when training finished or errored.

Update controller_actor_status and worker status on demand while calling TrainHead endpoint.

Related issue number

Checks

• I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
• I've run scripts/format.sh to lint the changes in this PR.
• I've included any doc changes needed for https://docs.ray.io/en/master/.
  • I've added any new APIs to the API Reference. For example, if I added a
    method in Tune, I've added it in doc/source/tune/api/ under the
    corresponding .rst file.
• I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
• Testing Strategy
  • Unit tests
  • Release tests
  • This PR is not tested :(

[matthewdeng]

Do we need this in this method since it's already checked at the start of get_train_runs? Also if this ever does happen it'll just error out when trying to use these imports later in the code.

[woshiyyya]

Removed

[matthewdeng]

nit: Feels a bit messy to update the run status here. If we keep this here I'd at least rename the method to _add_actor_and_update_run_status or add a docstring that documents this behavior.

[woshiyyya]

This one is doing some post processing to handle some abnormally terminated cases. Let me update the function name here.

[matthewdeng]

Should there be a status_detail for this one?

[matthewdeng]

What is the difference between ERRORED and ABORTED?

[woshiyyya]

We only mark it as ERRORED when Trainer detected an error and actively report it to StateActor.

We mark it as ABORTED when the TrainRun abnormally failed (e.g. due to node failure) and we passively set it in TrainHead.

[matthewdeng]

What's the purpose of this? We only ever have one "User Error" message right now.

[woshiyyya]

This is for tracking the error reason for now, and can be extend to track the details of current run status in the future(e.g. scaling up / down/ recovering when doing elastic training.

[matthewdeng]

This can cause problems if not careful in the future e.g. if somewhere we call update_train_run_info with a different id. One way to limit this is to restrict the public API and use a private method instead.

Something like this:

Suggested change

	self.train_run_info_dict = dict(
	id=run_id,
	job_id=job_id,
	name=run_name,
	controller_actor_id=controller_actor_id,
	workers=worker_info_list,
	datasets=dataset_info_list,
	start_time_ms=start_time_ms,
	run_status=run_status,
	status_detail=status_detail,
	)
	train_run_info = TrainRunInfo(**self.train_run_info_dict)
	ray.get(self.state_actor.register_train_run.remote(train_run_info))
	def update_train_run_info(self, updates: Dict[str, Any]) -> None:
	"""Update specific fields of a registered TrainRunInfo instance."""
	self.train_run_info_dict.update(updates)
	train_run_info = TrainRunInfo(**self.train_run_info_dict)
	ray.get(self.state_actor.register_train_run.remote(train_run_info))
	updates = dict(
	id=run_id,
	job_id=job_id,
	name=run_name,
	controller_actor_id=controller_actor_id,
	workers=worker_info_list,
	datasets=dataset_info_list,
	start_time_ms=start_time_ms,
	run_status=run_status,
	status_detail=status_detail,
	)
	# you can first assert that the info dict is empty here I guess
	_update_train_run_info(updates)
	def end_train_run(self, run_status, status_detail, end_time_ms):
	updates=dict(
	run_status=run_status,
	status_detail=status_detail,
	end_time_ms=end_time_ms,
	)
	_update_train_run_info(updates)
	def _update_train_run_info(self, updates: Dict[str, Any]) -> None:
	"""Update specific fields of a registered TrainRunInfo instance."""
	self.train_run_info_dict.update(updates)
	train_run_info = TrainRunInfo(**self.train_run_info_dict)
	ray.get(self.state_actor.register_train_run.remote(train_run_info))

[woshiyyya]

This is a good point! Let me update this

[matthewdeng]

Will this populate the dictionary on the first call? Since the first line in _update_train_run_info will check if the run_id is present.

[woshiyyya]

Yes it will populate the dict. But yeah I think it's a bit implicit, let me update it to

self.train_run_info_dict[run_id] = {}