worker.coordinator.coordinator

TaskState

class ding.worker.coordinator.coordinator.TaskState(task_id: str)

Overview:

State recorder of the task, including task_id and start_time.

Interface:

__init__

__init__(task_id: str) → None

Overview:

Init the task tate according to task_id and the init time.

Coordinator

class ding.worker.coordinator.coordinator.Coordinator(cfg: dict)

Overview:

the coordinator will manage parallel tasks and data

Interface:

__init__, start, close, __del__, state_dict, load_state_dict, deal_with_collector_send_data, deal_with_collector_finish_task, deal_with_learner_get_data, deal_with_learner_send_info, deal_with_learner_finish_task

Property:

system_shutdown_flag

__del__() → None

Overview:

__del__ method will close the coordinator.

__init__(cfg: dict) → None

Overview:

init method of the coordinator

Arguments:

• cfg (dict): the config file to init the coordinator

close() → None

Overview:

Close the coordinator, including closing the interaction thread, the collector learner threads and the

buffers.

deal_with_collector_finish_task(task_id: str, finished_task: dict) → None

Overview:

finish the collector task

Arguments:

• task_id (str): the collector task_id

• finished_task (dict): the finished_task

deal_with_collector_send_data(task_id: str, buffer_id: str, data_id: str, data: dict) → None

Overview:

deal with the data send from collector

Arguments:

• task_id (str): the collector task_id

• buffer_id (str): the buffer_id

• data_id (str): the data_id

• data (str): the data to dealt with

deal_with_learner_finish_task(task_id: str, finished_task: dict) → None

Overview:

finish the learner task, close the corresponding buffer

Arguments:

• task_id (str): the learner task_id

• finished_task (dict): the dict of task to finish

deal_with_learner_get_data(task_id: str, buffer_id: str, batch_size: int, cur_learner_iter: int) → List[dict]

Overview:

learner get the data from buffer

Arguments:

• task_id (str): the learner task_id

• buffer_id (str): the buffer_id

• batch_size (int): the batch_size to sample

• cur_learn_iter (int): the current learner iter num

deal_with_learner_send_info(task_id: str, buffer_id: str, info: dict) → None

Overview:

the learner send the info and update the priority in buffer

Arguments:

• task_id (str): the learner task id

• buffer_id (str): the buffer_id of buffer to add info to

• info (dict): the info to add

error(s: str) → None

Overview:

Return the error

Arguments:

• s (str): the error info to print

info(s: str) → None

Overview:

Return the info

Arguments:

• s (str): the string to print in info

load_state_dict(state_dict: dict) → None

Overview:

Pass when load state_dict.

start() → None

Overview:

Start the coordinator, including lunching the interaction thread and the collector learner threads.

state_dict() → dict

Overview:

Return empty state_dict.

property system_shutdown_flag: bool

Overview:

Return whether the system is shutdown

Returns:

• system_shutdown_flag (bool): whether the system is shutdown