text: # Copyright (c) 2010-2012 OpenStack Foundation
  #
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  #
  #    http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  # implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  
  import os
  import errno
  from os.path import isdir, isfile, join, dirname
  import random
  import shutil
  import time
  import itertools
  from six import viewkeys
  import six.moves.cPickle as pickle
  from swift import gettext_ as _
  
  import eventlet
  from eventlet import GreenPool, tpool, Timeout, sleep
  from eventlet.green import subprocess
  from eventlet.support.greenlets import GreenletExit
  
  from swift.common.constraints import check_drive
  from swift.common.ring.utils import is_local_device
  from swift.common.utils import whataremyips, unlink_older_than, \
      compute_eta, get_logger, dump_recon_cache, \
      rsync_module_interpolation, mkdirs, config_true_value, list_from_csv, \
      tpool_reraise, config_auto_int_value, storage_directory
  from swift.common.bufferedhttp import http_connect
  from swift.common.daemon import Daemon
  from swift.common.http import HTTP_OK, HTTP_INSUFFICIENT_STORAGE
  from swift.obj import ssync_sender
  from swift.obj.diskfile import get_data_dir, get_tmp_dir, DiskFileRouter
  from swift.common.storage_policy import POLICIES, REPL_POLICY
  
  DEFAULT_RSYNC_TIMEOUT = 900
  
  
  def _do_listdir(partition, replication_cycle):
      return (((partition + replication_cycle) % 10) == 0)
  
  
  class ObjectReplicator(Daemon):
      """
      Replicate objects.
  
      Encapsulates most logic and data needed by the object replication process.
      Each call to .replicate() performs one replication pass.  It's up to the
      caller to do this in a loop.
      """
  
      def __init__(self, conf, logger=None):
          """
          :param conf: configuration object obtained from ConfigParser
          :param logger: logging object
          """
          self.conf = conf
          self.logger = logger or get_logger(conf, log_route='object-replicator')
          self.devices_dir = conf.get('devices', '/srv/node')
          self.mount_check = config_true_value(conf.get('mount_check', 'true'))
          self.swift_dir = conf.get('swift_dir', '/etc/swift')
          self.bind_ip = conf.get('bind_ip', '0.0.0.0')
          self.servers_per_port = int(conf.get('servers_per_port', '0') or 0)
          self.port = None if self.servers_per_port else \
              int(conf.get('bind_port', 6200))
          self.concurrency = int(conf.get('concurrency', 1))
          self.stats_interval = int(conf.get('stats_interval', '300'))
          self.ring_check_interval = int(conf.get('ring_check_interval', 15))
          self.next_check = time.time() + self.ring_check_interval
          self.replication_cycle = random.randint(0, 9)
          self.partition_times = []
          self.interval = int(conf.get('interval') or
                              conf.get('run_pause') or 30)
          self.rsync_timeout = int(conf.get('rsync_timeout',
                                            DEFAULT_RSYNC_TIMEOUT))
          self.rsync_io_timeout = conf.get('rsync_io_timeout', '30')
          self.rsync_bwlimit = conf.get('rsync_bwlimit', '0')
          self.rsync_compress = config_true_value(
              conf.get('rsync_compress', 'no'))
          self.rsync_module = conf.get('rsync_module', '').rstrip('/')
          if not self.rsync_module:
              self.rsync_module = '{replication_ip}::object'
          self.http_timeout = int(conf.get('http_timeout', 60))
          self.lockup_timeout = int(conf.get('lockup_timeout', 1800))
          self.recon_cache_path = conf.get('recon_cache_path',
                                           '/var/cache/swift')
          self.rcache = os.path.join(self.recon_cache_path, "object.recon")
          self.conn_timeout = float(conf.get('conn_timeout', 0.5))
          self.node_timeout = float(conf.get('node_timeout', 10))
          self.sync_method = getattr(self, conf.get('sync_method') or 'rsync')
          self.network_chunk_size = int(conf.get('network_chunk_size', 65536))
          self.default_headers = {
              'Content-Length': '0',
              'user-agent': 'object-replicator %s' % os.getpid()}
          self.rsync_error_log_line_length = \
              int(conf.get('rsync_error_log_line_length', 0))
          self.handoffs_first = config_true_value(conf.get('handoffs_first',
                                                           False))
          self.handoff_delete = config_auto_int_value(
              conf.get('handoff_delete', 'auto'), 0)
          if any((self.handoff_delete, self.handoffs_first)):
              self.logger.warning('Handoff only mode is not intended for normal '
                                  'operation, please disable handoffs_first and '
                                  'handoff_delete before the next '
                                  'normal rebalance')
          self._df_router = DiskFileRouter(conf, self.logger)
  
      def _zero_stats(self):
          """Zero out the stats."""
          self.stats = {'attempted': 0, 'success': 0, 'failure': 0,
                        'hashmatch': 0, 'rsync': 0, 'remove': 0,
                        'start': time.time(), 'failure_nodes': {}}
  
      def _add_failure_stats(self, failure_devs_info):
          for node, dev in failure_devs_info:
              self.stats['failure'] += 1
              failure_devs = self.stats['failure_nodes'].setdefault(node, {})
              failure_devs.setdefault(dev, 0)
              failure_devs[dev] += 1
  
      def _get_my_replication_ips(self):
          my_replication_ips = set()
          ips = whataremyips()
          for policy in POLICIES:
              self.load_object_ring(policy)
              for local_dev in [dev for dev in policy.object_ring.devs
                                if dev and dev['replication_ip'] in ips and
                                dev['replication_port'] == self.port]:
                  my_replication_ips.add(local_dev['replication_ip'])
          return list(my_replication_ips)
  
      # Just exists for doc anchor point
      def sync(self, node, job, suffixes, *args, **kwargs):
          """
          Synchronize local suffix directories from a partition with a remote
          node.
  
          :param node: the "dev" entry for the remote node to sync with
          :param job: information about the partition being synced
          :param suffixes: a list of suffixes which need to be pushed
  
          :returns: boolean and dictionary, boolean indicating success or failure
          """
          return self.sync_method(node, job, suffixes, *args, **kwargs)
  
      def load_object_ring(self, policy):
          """
          Make sure the policy's rings are loaded.
  
          :param policy: the StoragePolicy instance
          :returns: appropriate ring object
          """
          policy.load_ring(self.swift_dir)
          return policy.object_ring
  
      def _rsync(self, args):
          """
          Execute the rsync binary to replicate a partition.
  
          :returns: return code of rsync process. 0 is successful
          """
          start_time = time.time()
          ret_val = None
          try:
              with Timeout(self.rsync_timeout):
                  proc = subprocess.Popen(args,
                                          stdout=subprocess.PIPE,
                                          stderr=subprocess.STDOUT)
                  results = proc.stdout.read()
                  ret_val = proc.wait()
          except Timeout:
              self.logger.error(_("Killing long-running rsync: %s"), str(args))
              proc.kill()
              return 1  # failure response code
          total_time = time.time() - start_time
          for result in results.split('\n'):
              if result == '':
                  continue
              if result.startswith('cd+'):
                  continue
              if not ret_val:
                  self.logger.info(result)
              else:
                  self.logger.error(result)
          if ret_val:
              error_line = _('Bad rsync return code: %(ret)d <- %(args)s') % \
                  {'args': str(args), 'ret': ret_val}
              if self.rsync_error_log_line_length:
                  error_line = error_line[:self.rsync_error_log_line_length]
              self.logger.error(error_line)
          else:
              log_method = self.logger.info if results else self.logger.debug
              log_method(
                  _("Successful rsync of %(src)s at %(dst)s (%(time).03f)"),
                  {'src': args[-2], 'dst': args[-1], 'time': total_time})
          return ret_val
  
      def rsync(self, node, job, suffixes):
          """
          Uses rsync to implement the sync method. This was the first
          sync method in Swift.
          """
          if not os.path.exists(job['path']):
              return False, {}
          args = [
              'rsync',
              '--recursive',
              '--whole-file',
              '--human-readable',
              '--xattrs',
              '--itemize-changes',
              '--ignore-existing',
              '--timeout=%s' % self.rsync_io_timeout,
              '--contimeout=%s' % self.rsync_io_timeout,
              '--bwlimit=%s' % self.rsync_bwlimit,
              '--exclude=.*.%s' % ''.join('[0-9a-zA-Z]' for i in range(6))
          ]
          if self.rsync_compress and \
                  job['region'] != node['region']:
              # Allow for compression, but only if the remote node is in
              # a different region than the local one.
              args.append('--compress')
          rsync_module = rsync_module_interpolation(self.rsync_module, node)
          had_any = False
          for suffix in suffixes:
              spath = join(job['path'], suffix)
              if os.path.exists(spath):
                  args.append(spath)
                  had_any = True
          if not had_any:
              return False, {}
          data_dir = get_data_dir(job['policy'])
          args.append(join(rsync_module, node['device'],
                      data_dir, job['partition']))
          return self._rsync(args) == 0, {}
  
      def ssync(self, node, job, suffixes, remote_check_objs=None):
          return ssync_sender.Sender(
              self, node, job, suffixes, remote_check_objs)()
  
      def check_ring(self, object_ring):
          """
          Check to see if the ring has been updated
          :param object_ring: the ring to check
  
          :returns: boolean indicating whether or not the ring has changed
          """
          if time.time() > self.next_check:
              self.next_check = time.time() + self.ring_check_interval
              if object_ring.has_changed():
                  return False
          return True
  
      def update_deleted(self, job):
          """
          High-level method that replicates a single partition that doesn't
          belong on this node.
  
          :param job: a dict containing info about the partition to be replicated
          """
  
          def tpool_get_suffixes(path):
              return [suff for suff in os.listdir(path)
                      if len(suff) == 3 and isdir(join(path, suff))]
          self.replication_count += 1
          self.logger.increment('partition.delete.count.%s' % (job['device'],))
          headers = dict(self.default_headers)
          headers['X-Backend-Storage-Policy-Index'] = int(job['policy'])
          failure_devs_info = set()
          begin = time.time()
          handoff_partition_deleted = False
          try:
              responses = []
              suffixes = tpool.execute(tpool_get_suffixes, job['path'])
              synced_remote_regions = {}
              delete_objs = None
              if suffixes:
                  for node in job['nodes']:
                      self.stats['rsync'] += 1
                      kwargs = {}
                      if node['region'] in synced_remote_regions and \
                              self.conf.get('sync_method', 'rsync') == 'ssync':
                          kwargs['remote_check_objs'] = \
                              synced_remote_regions[node['region']]
                      # candidates is a dict(hash=>timestamp) of objects
                      # for deletion
                      success, candidates = self.sync(
                          node, job, suffixes, **kwargs)
                      if success:
                          with Timeout(self.http_timeout):
                              conn = http_connect(
                                  node['replication_ip'],
                                  node['replication_port'],
                                  node['device'], job['partition'], 'REPLICATE',
                                  '/' + '-'.join(suffixes), headers=headers)
                              conn.getresponse().read()
                          if node['region'] != job['region']:
                              synced_remote_regions[node['region']] = viewkeys(
                                  candidates)
                      else:
                          failure_devs_info.add((node['replication_ip'],
                                                 node['device']))
                      responses.append(success)
                  for cand_objs in synced_remote_regions.values():
                      if delete_objs is None:
                          delete_objs = cand_objs
                      else:
                          delete_objs = delete_objs & cand_objs
  
              if self.handoff_delete:
                  # delete handoff if we have had handoff_delete successes
                  delete_handoff = len([resp for resp in responses if resp]) >= \
                      self.handoff_delete
              else:
                  # delete handoff if all syncs were successful
                  delete_handoff = len(responses) == len(job['nodes']) and \
                      all(responses)
              if delete_handoff:
                  self.stats['remove'] += 1
                  if (self.conf.get('sync_method', 'rsync') == 'ssync' and
                          delete_objs is not None):
                      self.logger.info(_("Removing %s objects"),
                                       len(delete_objs))
                      _junk, error_paths = self.delete_handoff_objs(
                          job, delete_objs)
                      # if replication works for a hand-off device and it failed,
                      # the remote devices which are target of the replication
                      # from the hand-off device will be marked. Because cleanup
                      # after replication failed means replicator needs to
                      # replicate again with the same info.
                      if error_paths:
                          failure_devs_info.update(
                              [(failure_dev['replication_ip'],
                                failure_dev['device'])
                               for failure_dev in job['nodes']])
                  else:
                      self.delete_partition(job['path'])
                      handoff_partition_deleted = True
              elif not suffixes:
                  self.delete_partition(job['path'])
                  handoff_partition_deleted = True
          except (Exception, Timeout):
              self.logger.exception(_("Error syncing handoff partition"))
              self._add_failure_stats(failure_devs_info)
          finally:
              target_devs_info = set([(target_dev['replication_ip'],
                                       target_dev['device'])
                                      for target_dev in job['nodes']])
              self.stats['success'] += len(target_devs_info - failure_devs_info)
              if not handoff_partition_deleted:
                  self.handoffs_remaining += 1
              self.partition_times.append(time.time() - begin)
              self.logger.timing_since('partition.delete.timing', begin)
  
      def delete_partition(self, path):
          self.logger.info(_("Removing partition: %s"), path)
          tpool.execute(shutil.rmtree, path)
  
      def delete_handoff_objs(self, job, delete_objs):
          success_paths = []
          error_paths = []
          for object_hash in delete_objs:
              object_path = storage_directory(job['obj_path'], job['partition'],
                                              object_hash)
              tpool.execute(shutil.rmtree, object_path, ignore_errors=True)
              suffix_dir = dirname(object_path)
              try:
                  os.rmdir(suffix_dir)
                  success_paths.append(object_path)
              except OSError as e:
                  if e.errno not in (errno.ENOENT, errno.ENOTEMPTY):
                      error_paths.append(object_path)
                      self.logger.exception(
                          "Unexpected error trying to cleanup suffix dir:%r",
                          suffix_dir)
          return success_paths, error_paths
  
      def update(self, job):
          """
          High-level method that replicates a single partition.
  
          :param job: a dict containing info about the partition to be replicated
          """
          self.replication_count += 1
          self.logger.increment('partition.update.count.%s' % (job['device'],))
          headers = dict(self.default_headers)
          headers['X-Backend-Storage-Policy-Index'] = int(job['policy'])
          target_devs_info = set()
          failure_devs_info = set()
          begin = time.time()
          df_mgr = self._df_router[job['policy']]
          try:
              hashed, local_hash = tpool_reraise(
                  df_mgr._get_hashes, job['device'],
                  job['partition'], job['policy'],
                  do_listdir=_do_listdir(
                      int(job['partition']),
                      self.replication_cycle))
              self.suffix_hash += hashed
              self.logger.update_stats('suffix.hashes', hashed)
              attempts_left = len(job['nodes'])
              synced_remote_regions = set()
              random.shuffle(job['nodes'])
              nodes = itertools.chain(
                  job['nodes'],
                  job['policy'].object_ring.get_more_nodes(
                      int(job['partition'])))
              while attempts_left > 0:
                  # If this throws StopIteration it will be caught way below
                  node = next(nodes)
                  target_devs_info.add((node['replication_ip'], node['device']))
                  attempts_left -= 1
                  # if we have already synced to this remote region,
                  # don't sync again on this replication pass
                  if node['region'] in synced_remote_regions:
                      continue
                  try:
                      with Timeout(self.http_timeout):
                          resp = http_connect(
                              node['replication_ip'], node['replication_port'],
                              node['device'], job['partition'], 'REPLICATE',
                              '', headers=headers).getresponse()
                          if resp.status == HTTP_INSUFFICIENT_STORAGE:
                              self.logger.error(
                                  _('%(replication_ip)s/%(device)s '
                                    'responded as unmounted'), node)
                              attempts_left += 1
                              failure_devs_info.add((node['replication_ip'],
                                                     node['device']))
                              continue
                          if resp.status != HTTP_OK:
                              self.logger.error(_("Invalid response %(resp)s "
                                                  "from %(ip)s"),
                                                {'resp': resp.status,
                                                 'ip': node['replication_ip']})
                              failure_devs_info.add((node['replication_ip'],
                                                     node['device']))
                              continue
                          remote_hash = pickle.loads(resp.read())
                          del resp
                      suffixes = [suffix for suffix in local_hash if
                                  local_hash[suffix] !=
                                  remote_hash.get(suffix, -1)]
                      if not suffixes:
                          self.stats['hashmatch'] += 1
                          continue
                      hashed, recalc_hash = tpool_reraise(
                          df_mgr._get_hashes,
                          job['device'], job['partition'], job['policy'],
                          recalculate=suffixes)
                      self.logger.update_stats('suffix.hashes', hashed)
                      local_hash = recalc_hash
                      suffixes = [suffix for suffix in local_hash if
                                  local_hash[suffix] !=
                                  remote_hash.get(suffix, -1)]
                      self.stats['rsync'] += 1
                      success, _junk = self.sync(node, job, suffixes)
                      with Timeout(self.http_timeout):
                          conn = http_connect(
                              node['replication_ip'], node['replication_port'],
                              node['device'], job['partition'], 'REPLICATE',
                              '/' + '-'.join(suffixes),
                              headers=headers)
                          conn.getresponse().read()
                      if not success:
                          failure_devs_info.add((node['replication_ip'],
                                                 node['device']))
                      # add only remote region when replicate succeeded
                      if success and node['region'] != job['region']:
                          synced_remote_regions.add(node['region'])
                      self.suffix_sync += len(suffixes)
                      self.logger.update_stats('suffix.syncs', len(suffixes))
                  except (Exception, Timeout):
                      failure_devs_info.add((node['replication_ip'],
                                             node['device']))
                      self.logger.exception(_("Error syncing with node: %s") %
                                            node)
              self.suffix_count += len(local_hash)
          except (Exception, Timeout):
              failure_devs_info.update(target_devs_info)
              self._add_failure_stats(failure_devs_info)
              self.logger.exception(_("Error syncing partition"))
          finally:
              self.stats['success'] += len(target_devs_info - failure_devs_info)
              self.partition_times.append(time.time() - begin)
              self.logger.timing_since('partition.update.timing', begin)
  
      def stats_line(self):
          """
          Logs various stats for the currently running replication pass.
          """
          if self.replication_count:
              elapsed = (time.time() - self.start) or 0.000001
              rate = self.replication_count / elapsed
              self.logger.info(
                  _("%(replicated)d/%(total)d (%(percentage).2f%%)"
                    " partitions replicated in %(time).2fs (%(rate).2f/sec, "
                    "%(remaining)s remaining)"),
                  {'replicated': self.replication_count, 'total': self.job_count,
                   'percentage': self.replication_count * 100.0 / self.job_count,
                   'time': time.time() - self.start, 'rate': rate,
                   'remaining': '%d%s' % compute_eta(self.start,
                                                     self.replication_count,
                                                     self.job_count)})
              self.logger.info(_('%(success)s successes, %(failure)s failures')
                               % self.stats)
  
              if self.suffix_count:
                  self.logger.info(
                      _("%(checked)d suffixes checked - "
                        "%(hashed).2f%% hashed, %(synced).2f%% synced"),
                      {'checked': self.suffix_count,
                       'hashed': (self.suffix_hash * 100.0) / self.suffix_count,
                       'synced': (self.suffix_sync * 100.0) / self.suffix_count})
                  self.partition_times.sort()
                  self.logger.info(
                      _("Partition times: max %(max).4fs, "
                        "min %(min).4fs, med %(med).4fs"),
                      {'max': self.partition_times[-1],
                       'min': self.partition_times[0],
                       'med': self.partition_times[
                           len(self.partition_times) // 2]})
          else:
              self.logger.info(
                  _("Nothing replicated for %s seconds."),
                  (time.time() - self.start))
  
      def kill_coros(self):
          """Utility function that kills all coroutines currently running."""
          for coro in list(self.run_pool.coroutines_running):
              try:
                  coro.kill(GreenletExit)
              except GreenletExit:
                  pass
  
      def heartbeat(self):
          """
          Loop that runs in the background during replication.  It periodically
          logs progress.
          """
          while True:
              eventlet.sleep(self.stats_interval)
              self.stats_line()
  
      def detect_lockups(self):
          """
          In testing, the pool.waitall() call very occasionally failed to return.
          This is an attempt to make sure the replicator finishes its replication
          pass in some eventuality.
          """
          while True:
              eventlet.sleep(self.lockup_timeout)
              if self.replication_count == self.last_replication_count:
                  self.logger.error(_("Lockup detected.. killing live coros."))
                  self.kill_coros()
              self.last_replication_count = self.replication_count
  
      def build_replication_jobs(self, policy, ips, override_devices=None,
                                 override_partitions=None):
          """
          Helper function for collect_jobs to build jobs for replication
          using replication style storage policy
          """
          jobs = []
          df_mgr = self._df_router[policy]
          self.all_devs_info.update(
              [(dev['replication_ip'], dev['device'])
               for dev in policy.object_ring.devs if dev])
          data_dir = get_data_dir(policy)
          found_local = False
          for local_dev in [dev for dev in policy.object_ring.devs
                            if (dev
                                and is_local_device(ips,
                                                    self.port,
                                                    dev['replication_ip'],
                                                    dev['replication_port'])
                                and (override_devices is None
                                     or dev['device'] in override_devices))]:
              found_local = True
              dev_path = check_drive(self.devices_dir, local_dev['device'],
                                     self.mount_check)
              if not dev_path:
                  self._add_failure_stats(
                      [(failure_dev['replication_ip'],
                        failure_dev['device'])
                       for failure_dev in policy.object_ring.devs
                       if failure_dev])
                  self.logger.warning(
                      _('%s is not mounted'), local_dev['device'])
                  continue
              obj_path = join(dev_path, data_dir)
              tmp_path = join(dev_path, get_tmp_dir(policy))
              unlink_older_than(tmp_path, time.time() -
                                df_mgr.reclaim_age)
              if not os.path.exists(obj_path):
                  try:
                      mkdirs(obj_path)
                  except Exception:
                      self.logger.exception('ERROR creating %s' % obj_path)
                  continue
              for partition in os.listdir(obj_path):
                  if (override_partitions is not None
                          and partition not in override_partitions):
                      continue
  
                  if (partition.startswith('auditor_status_') and
                          partition.endswith('.json')):
                      # ignore auditor status files
                      continue
  
                  part_nodes = None
                  try:
                      job_path = join(obj_path, partition)
                      part_nodes = policy.object_ring.get_part_nodes(
                          int(partition))
                      nodes = [node for node in part_nodes
                               if node['id'] != local_dev['id']]
                      jobs.append(
                          dict(path=job_path,
                               device=local_dev['device'],
                               obj_path=obj_path,
                               nodes=nodes,
                               delete=len(nodes) > len(part_nodes) - 1,
                               policy=policy,
                               partition=partition,
                               region=local_dev['region']))
                  except ValueError:
                      if part_nodes:
                          self._add_failure_stats(
                              [(failure_dev['replication_ip'],
                                failure_dev['device'])
                               for failure_dev in nodes])
                      else:
                          self._add_failure_stats(
                              [(failure_dev['replication_ip'],
                                failure_dev['device'])
                               for failure_dev in policy.object_ring.devs
                               if failure_dev])
                      continue
          if not found_local:
              self.logger.error("Can't find itself in policy with index %d with"
                                " ips %s and with port %s in ring file, not"
                                " replicating",
                                int(policy), ", ".join(ips), self.port)
          return jobs
  
      def collect_jobs(self, override_devices=None, override_partitions=None,
                       override_policies=None):
          """
          Returns a sorted list of jobs (dictionaries) that specify the
          partitions, nodes, etc to be rsynced.
  
          :param override_devices: if set, only jobs on these devices
              will be returned
          :param override_partitions: if set, only jobs on these partitions
              will be returned
          :param override_policies: if set, only jobs in these storage
              policies will be returned
          """
          jobs = []
          ips = whataremyips(self.bind_ip)
          for policy in POLICIES:
              # Skip replication if next_part_power is set. In this case
              # every object is hard-linked twice, but the replicator can't
              # detect them and would create a second copy of the file if not
              # yet existing - and this might double the actual transferred
              # and stored data
              next_part_power = getattr(
                  policy.object_ring, 'next_part_power', None)
              if next_part_power is not None:
                  self.logger.warning(
                      _("next_part_power set in policy '%s'. Skipping"),
                      policy.name)
                  continue
  
              if policy.policy_type == REPL_POLICY:
                  if (override_policies is not None and
                          str(policy.idx) not in override_policies):
                      continue
                  # ensure rings are loaded for policy
                  self.load_object_ring(policy)
                  jobs += self.build_replication_jobs(
                      policy, ips, override_devices=override_devices,
                      override_partitions=override_partitions)
          random.shuffle(jobs)
          if self.handoffs_first:
              # Move the handoff parts to the front of the list
              jobs.sort(key=lambda job: not job['delete'])
          self.job_count = len(jobs)
          return jobs
  
      def replicate(self, override_devices=None, override_partitions=None,
                    override_policies=None):
          """Run a replication pass"""
          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()  # Give spawns a cycle
  
          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 = check_drive(self.devices_dir, job['device'],
                                         self.mount_check)
                  if not 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']:
                      # in handoffs first mode, we won't process primary
                      # partitions until rebalance was successful!
                      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']):
                          # Clean up any (probably zero-byte) files where a
                          # partition should be.
                          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
  
      def run_once(self, *args, **kwargs):
          self._zero_stats()
          self.logger.info(_("Running object replicator in script mode."))
  
          override_devices = list_from_csv(kwargs.get('devices'))
          override_partitions = list_from_csv(kwargs.get('partitions'))
          override_policies = list_from_csv(kwargs.get('policies'))
          if not override_devices:
              override_devices = None
          if not override_partitions:
              override_partitions = None
          if not override_policies:
              override_policies = None
  
          self.replicate(
              override_devices=override_devices,
              override_partitions=override_partitions,
              override_policies=override_policies)
          total = (time.time() - self.stats['start']) / 60
          self.logger.info(
              _("Object replication complete (once). (%.02f minutes)"), total)
          if not (override_partitions or override_devices):
              replication_last = time.time()
              dump_recon_cache({'replication_stats': self.stats,
                                'replication_time': total,
                                'replication_last': replication_last,
                                'object_replication_time': total,
                                'object_replication_last': replication_last},
                               self.rcache, self.logger)
  
      def run_forever(self, *args, **kwargs):
          self.logger.info(_("Starting object replicator in daemon mode."))
          # Run the replicator continually
          while True:
              self._zero_stats()
              self.logger.info(_("Starting object replication pass."))
              # Run the replicator
              self.replicate()
              total = (time.time() - self.stats['start']) / 60
              self.logger.info(
                  _("Object replication complete. (%.02f minutes)"), total)
              replication_last = time.time()
              dump_recon_cache({'replication_stats': self.stats,
                                'replication_time': total,
                                'replication_last': replication_last,
                                'object_replication_time': total,
                                'object_replication_last': replication_last},
                               self.rcache, self.logger)
              self.logger.debug('Replication sleeping for %s seconds.',
                                self.interval)
              sleep(self.interval)
  