Worker should stop task and report failed to master if a mini-batch of the task failed.

[workingloong]

A mini-batch may failed during executing mini-batches of task, but the worker will continue and report err_msg to master, then report task done after completing all mini-batches of the task. The worker should stop execution of the task and report failed to master. Then master resets the task in todo queue.

[QiJune]

Please refer to #1253 . Maybe we could just let it failed. Task status is also no need to keep.

[mhaoli]

Please refer to #1253 . Maybe we could just let it failed. Task status is also no need to keep.

This issue is different from #1253. This issue states a bug: A worker may tell the master that the task T is finished successful even if some mini-batches in task T failed.

This is because that function report_record_done ignores err_msg if the mini-batch is not the last mini-batch in the current task. Please refer to

elasticdl/elasticdl/python/worker/task_data_service.py

Lines 40 to 64 in 5455415

	def report_record_done(self, count, err_msg=""):
	"""
	Report the number of records in the latest processed batch,
	so TaskDataService knows if some pending tasks are finished
	and report_task_result to the master.
	self._pending_tasks_with_counts[0][0] is the first pending task,
	self._pending_tasks_with_counts[0][1] is the number of records
	in this task.
	"""
	self._reported_record_count += count
	if (
	len(self._pending_tasks_with_counts)
	and self._reported_record_count
	>= self._pending_tasks_with_counts[0][1]
	):
	with self._lock:
	while (
	len(self._pending_tasks_with_counts)
	and self._reported_record_count
	>= self._pending_tasks_with_counts[0][1]
	):
	task, _ = self._pending_tasks_with_counts.pop(0)
	self._worker.report_task_result(task.task_id, err_msg)
	if len(self._pending_tasks_with_counts):
	self._current_task = self._pending_tasks_with_counts[0][0]

[chunyang-wen]

@workingloong Any update on this issue? If not, I can try to fix the problem.

[workingloong]

@workingloong Any update on this issue? If not, I can try to fix the problem.

No any.

[chunyang-wen]

After a further investigation into the code base, it seems that the err_msg's length is used to indicate the status of a mini-batch. If any worker reports a non-empty message to the master, it means the task dispatched to the worker fails. And TaskDispatcher will add the task to the todo queue.

TaskDataService of a worker just reports the message back to master.

I will close this issue. If any problem exists, please reopen it and ping me.

[workingloong]

After a further investigation into the code base, it seems that the err_msg's length is used to indicate the status of a mini-batch. If any worker reports a non-empty message to the master, it means the task dispatched to the worker fails. And TaskDispatcher will add the task to the todo queue.

TaskDataService of a worker just reports the message back to master.

I will close this issue. If any problem exists, please reopen it and ping me.

TaskDataService only reports the message when the mini-batch is the last one in the task. So the other mini-batch err_msg will be ignored, because self._reported_record_count < self._pending_tasks_with_counts[0][1].

elasticdl/elasticdl/python/worker/task_data_service.py

Lines 40 to 64 in 5455415

	def report_record_done(self, count, err_msg=""):
	"""
	Report the number of records in the latest processed batch,
	so TaskDataService knows if some pending tasks are finished
	and report_task_result to the master.
	self._pending_tasks_with_counts[0][0] is the first pending task,
	self._pending_tasks_with_counts[0][1] is the number of records
	in this task.
	"""
	self._reported_record_count += count
	if (
	len(self._pending_tasks_with_counts)
	and self._reported_record_count
	>= self._pending_tasks_with_counts[0][1]
	):
	with self._lock:
	while (
	len(self._pending_tasks_with_counts)
	and self._reported_record_count
	>= self._pending_tasks_with_counts[0][1]
	):
	task, _ = self._pending_tasks_with_counts.pop(0)
	self._worker.report_task_result(task.task_id, err_msg)
	if len(self._pending_tasks_with_counts):
	self._current_task = self._pending_tasks_with_counts[0][0]