def __create_action(self, main_root):
		action_dir = os.path.join(main_root, 'action')
		safe_mkdir(action_dir)

		actions = set()
		for unit, param in self.__task.data['workflow'].items():
			actions.add(param['type'])
		self.__action_mgr.create_actions(actions, action_dir)
def _prepare_streams(self) -> None:
        if self._write_stream is not None:
            return
        if self.filename:
            filepath = os.path.join(self.dirpath, self._prepare_filename())
            fs = get_filesystem(filepath)
            file = fs.open(filepath, "a")
            self._output_file = file
            self._write_stream = file.write
        else:
            self._write_stream = self._rank_zero_info