Deferred init so that we can pass in previously created workers.

def init(self, aggregators):
        """Deferred init so that we can pass in previously created workers."""

        assert len(aggregators) == self.num_aggregation_workers, aggregators
        if len(self.remote_evaluators) < self.num_aggregation_workers:
            raise ValueError(
                "The number of aggregation workers should not exceed the "
                "number of total evaluation workers ({} vs {})".format(
                    self.num_aggregation_workers, len(self.remote_evaluators)))

        assigned_evaluators = collections.defaultdict(list)
        for i, ev in enumerate(self.remote_evaluators):
            assigned_evaluators[i % self.num_aggregation_workers].append(ev)

        self.workers = aggregators
        for i, worker in enumerate(self.workers):
            worker.init.remote(
                self.broadcasted_weights, assigned_evaluators[i],
                self.max_sample_requests_in_flight_per_worker,
                self.replay_proportion, self.replay_buffer_num_slots,
                self.train_batch_size, self.sample_batch_size)

        self.agg_tasks = TaskPool()
        for agg in self.workers:
            agg.set_weights.remote(self.broadcasted_weights)
            self.agg_tasks.add(agg, agg.get_train_batches.remote())

        self.initialized = True