def start_training(self, train_func: Callable[([], T)]) -> None:
    'Executes a training function on all workers in a separate thread.\n\n        ``finish_training`` should be called after this.\n\n        Args:\n            train_func (Callable): The training function to run on each worker.\n        '

    def initialize_session(world_rank, train_func):
        try:
            init_session(training_func=train_func, world_rank=world_rank)
        except ValueError:
            raise SGDBackendError('Attempting to start training but a previous training run is still ongoing. You must call `finish_training` before calling `start_training` again.')
    futures = []
    for world_rank in range(len(self.worker_group)):
        futures.append(self.worker_group.execute_single_async(world_rank, initialize_session, world_rank=world_rank, train_func=train_func))
    ray.get(futures)

    def train_async():
        session = get_session()
        session.start()
    self.worker_group.execute_async(train_async)
Executes a training function on all workers in a separate thread.

``finish_training`` should be called after this.

Args:
    train_func (Callable): The training function to run on each worker.
def start_training(self, train_func: Callable[([], T)]) -> None:
    'Executes a training function on all workers in a separate thread.\n\n        ``finish_training`` should be called after this.\n\n        Args:\n            train_func (Callable): The training function to run on each worker.\n        '

    def initialize_session(world_rank, train_func):
        try:
            init_session(training_func=train_func, world_rank=world_rank)
        except ValueError:
            raise SGDBackendError('Attempting to start training but a previous training run is still ongoing. You must call `finish_training` before calling `start_training` again.')
    futures = []
    for world_rank in range(len(self.worker_group)):
        futures.append(self.worker_group.execute_single_async(world_rank, initialize_session, world_rank=world_rank, train_func=train_func))
    ray.get(futures)

    def train_async():
        session = get_session()
        session.start()
    self.worker_group.execute_async(train_async)