'Run a replication pass'

def replicate(self, override_devices=None, override_partitions=None, override_policies=None):

    self.start = time.time()
    self.suffix_count = 0
    self.suffix_sync = 0
    self.suffix_hash = 0
    self.replication_count = 0
    self.last_replication_count = (-1)
    self.replication_cycle = ((self.replication_cycle + 1) % 10)
    self.partition_times = []
    self.my_replication_ips = self._get_my_replication_ips()
    self.all_devs_info = set()
    self.handoffs_remaining = 0
    stats = eventlet.spawn(self.heartbeat)
    lockup_detector = eventlet.spawn(self.detect_lockups)
    eventlet.sleep()
    current_nodes = None
    try:
        self.run_pool = GreenPool(size=self.concurrency)
        jobs = self.collect_jobs(override_devices=override_devices, override_partitions=override_partitions, override_policies=override_policies)
        for job in jobs:
            current_nodes = job['nodes']
            if (override_devices and (job['device'] not in override_devices)):
                continue
            if (override_partitions and (job['partition'] not in override_partitions)):
                continue
            dev_path = join(self.devices_dir, job['device'])
            if (self.mount_check and (not ismount(dev_path))):
                self._add_failure_stats([(failure_dev['replication_ip'], failure_dev['device']) for failure_dev in job['nodes']])
                self.logger.warning(_('%s    is    not    mounted'), job['device'])
                continue
            if (self.handoffs_first and (not job['delete'])):
                if self.handoffs_remaining:
                    self.logger.warning(_('Handoffs    first    mode    still    has    handoffs    remaining.        Aborting    current    replication    pass.'))
                    break
            if (not self.check_ring(job['policy'].object_ring)):
                self.logger.info(_('Ring    change    detected.    Aborting    current    replication    pass.'))
                return
            try:
                if isfile(job['path']):
                    self.logger.warning('Removing    partition    directory    which    was    a    file:    %s', job['path'])
                    os.remove(job['path'])
                    continue
            except OSError:
                continue
            if job['delete']:
                self.run_pool.spawn(self.update_deleted, job)
            else:
                self.run_pool.spawn(self.update, job)
        current_nodes = None
        with Timeout(self.lockup_timeout):
            self.run_pool.waitall()
    except (Exception, Timeout):
        if current_nodes:
            self._add_failure_stats([(failure_dev['replication_ip'], failure_dev['device']) for failure_dev in current_nodes])
        else:
            self._add_failure_stats(self.all_devs_info)
        self.logger.exception(_('Exception    in    top-level    replication    loop'))
        self.kill_coros()
    finally:
        stats.kill()
        lockup_detector.kill()
        self.stats_line()
        self.stats['attempted'] = self.replication_count
