def reclaim(self, age_timestamp, sync_timestamp):
        """
        Delete rows from the db_contains_type table that are marked deleted
        and whose created_at timestamp is < age_timestamp.  Also deletes rows
        from incoming_sync and outgoing_sync where the updated_at timestamp is
        < sync_timestamp.

        In addition, this calls the DatabaseBroker's :func:`_reclaim` method.

        :param age_timestamp: max created_at timestamp of object rows to delete
        :param sync_timestamp: max update_at timestamp of sync rows to delete
        """
        if self.db_file != ':memory:' and os.path.exists(self.pending_file):
            with lock_parent_directory(self.pending_file,
                                       self.pending_timeout):
                self._commit_puts()
        with self.get() as conn:
            conn.execute('''
                DELETE FROM %s WHERE deleted = 1 AND %s < ?
            ''' % (self.db_contains_type, self.db_reclaim_timestamp),
                (age_timestamp,))
            try:
                conn.execute('''
                    DELETE FROM outgoing_sync WHERE updated_at < ?
                ''', (sync_timestamp,))
                conn.execute('''
                    DELETE FROM incoming_sync WHERE updated_at < ?
                ''', (sync_timestamp,))
            except sqlite3.OperationalError as err:
                # Old dbs didn't have updated_at in the _sync tables.
                if 'no such column: updated_at' not in str(err):
                    raise
            DatabaseBroker._reclaim(self, conn, age_timestamp)
            conn.commit()

 def start(self):
     """Start up the ring master"""
     self.logger.notice("Ring-Master starting up")
     self.logger.notice("-> Entering ring orchestration loop.")
     while True:
         try:
             self.pause_if_asked()
             if self.in_change_window():
                 for btype in sorted(self.builder_files.keys()):
                     with lock_parent_directory(self.builder_files[btype],
                                                self.lock_timeout):
                         ring_changed = self.orchestration_pass(btype)
                     if ring_changed:
                         sleep(self.recheck_after_change_interval)
                     else:
                         sleep(self.recheck_interval)
             else:
                 self.logger.debug('Not in change window')
                 sleep(60)
         except exceptions.LockTimeout:
             self.logger.exception('Orchestration LockTimeout Encountered')
         except Exception:
             self.logger.exception('Orchestration Error')
             sleep(60)
         sleep(1)

 def _validate_file(self, filename):
     """Validate md5 of file"""
     if self._changed(filename):
         self.logger.debug("updating md5")
         with lock_parent_directory(self.swiftdir, self.lock_timeout):
             self.last_tstamp[filename] = stat(filename).st_mtime
             self.current_md5[filename] = get_md5sum(filename)

    def _commit_puts(self, item_list=None):
        """
        Scan for .pending files and commit the found records by feeding them
        to merge_items().

        :param item_list: A list of items to commit in addition to .pending
        """
        if self.db_file == ':memory:' or not os.path.exists(self.pending_file):
            return
        if item_list is None:
            item_list = []
        with lock_parent_directory(self.pending_file, self.pending_timeout):
            self._preallocate()
            if not os.path.getsize(self.pending_file):
                if item_list:
                    self.merge_items(item_list)
                return
            with open(self.pending_file, 'r+b') as fp:
                for entry in fp.read().split(':'):
                    if entry:
                        try:
                            self._commit_puts_load(item_list, entry)
                        except Exception:
                            self.logger.exception(
                                _('Invalid pending entry %(file)s: %(entry)s'),
                                {'file': self.pending_file, 'entry': entry})
                if item_list:
                    self.merge_items(item_list)
                try:
                    os.ftruncate(fp.fileno(), 0)
                except OSError as err:
                    if err.errno != errno.ENOENT:
                        raise

def main(arguments=None):
    global argv, backup_dir, builder, builder_file, ring_file
    if arguments:
        argv = arguments
    else:
        argv = sys_argv

    if len(argv) < 2:
        print "swift-ring-builder %(MAJOR_VERSION)s.%(MINOR_VERSION)s\n" % \
              globals()
        print Commands.default.__doc__.strip()
        print
        cmds = [c for c, f in Commands.__dict__.iteritems()
                if f.__doc__ and c[0] != '_' and c != 'default']
        cmds.sort()
        for cmd in cmds:
            print Commands.__dict__[cmd].__doc__.strip()
            print
        print parse_search_value.__doc__.strip()
        print
        for line in wrap(' '.join(cmds), 79, initial_indent='Quick list: ',
                         subsequent_indent='            '):
            print line
        print('Exit codes: 0 = operation successful\n'
              '            1 = operation completed with warnings\n'
              '            2 = error')
        exit(EXIT_SUCCESS)

    builder_file, ring_file = parse_builder_ring_filename_args(argv)

    if exists(builder_file):
        builder = RingBuilder.load(builder_file)
    elif len(argv) < 3 or argv[2] not in('create', 'write_builder'):
        print 'Ring Builder file does not exist: %s' % argv[1]
        exit(EXIT_ERROR)

    backup_dir = pathjoin(dirname(argv[1]), 'backups')
    try:
        mkdir(backup_dir)
    except OSError as err:
        if err.errno != EEXIST:
            raise

    if len(argv) == 2:
        command = "default"
    else:
        command = argv[2]
    if argv[0].endswith('-safe'):
        try:
            with lock_parent_directory(abspath(argv[1]), 15):
                Commands.__dict__.get(command, Commands.unknown.im_func)()
        except exceptions.LockTimeout:
            print "Ring/builder dir currently locked."
            exit(2)
    else:
        Commands.__dict__.get(command, Commands.unknown.im_func)()

 def delete_db(self, object_file):
     hash_dir = os.path.dirname(object_file)
     suf_dir = os.path.dirname(hash_dir)
     with lock_parent_directory(object_file):
         shutil.rmtree(hash_dir, True)
     try:
         os.rmdir(suf_dir)
     except OSError, err:
         if err.errno not in (errno.ENOENT, errno.ENOTEMPTY):
             self.logger.exception(_("ERROR while trying to clean up %s") % suf_dir)

 def test_ringmaster_validate_locked_dir(self):
     self._setup_builder_rings()
     rma = RingMasterApp({'swiftdir': self.testdir, 'log_path': self.test_log_path, 'locktimeout': "0.1"})
     for i in rma.current_md5:
         self.assertEquals(rma._changed(i), False)
     self._setup_builder_rings(count=5)
     for i in rma.current_md5:
         t = time.time() - 300
         os.utime(i, (t, t))
     with lock_parent_directory(self.testdir):
         for i in rma.current_md5:
             self.assertRaises(LockTimeout, rma._validate_file, i)

 def _commit_puts_stale_ok(self):
     """
     Catch failures of _commit_puts() if broker is intended for
     reading of stats, and thus does not care for pending updates.
     """
     if self.db_file == ":memory:" or not os.path.exists(self.pending_file):
         return
     try:
         with lock_parent_directory(self.pending_file, self.pending_timeout):
             self._commit_puts()
     except LockTimeout:
         if not self.stale_reads_ok:
             raise

 def _commit_puts_stale_ok(self):
     """
     Catch failures of _commit_puts() if broker is intended for
     reading of stats, and thus does not care for pending updates.
     """
     if self._skip_commit_puts():
         return
     try:
         with lock_parent_directory(self.pending_file,
                                    self.pending_timeout):
             self._commit_puts()
     except (LockTimeout, sqlite3.OperationalError):
         if not self.stale_reads_ok:
             raise

 def delete_db(self, object_file):
     hash_dir = os.path.dirname(object_file)
     suf_dir = os.path.dirname(hash_dir)
     with lock_parent_directory(object_file):
         shutil.rmtree(hash_dir, True)
     try:
         os.rmdir(suf_dir)
     except OSError as err:
         if err.errno not in (errno.ENOENT, errno.ENOTEMPTY):
             self.logger.exception(
                 _('ERROR while trying to clean up %s') % suf_dir)
     self.stats['remove'] += 1
     device_name = self.extract_device(object_file)
     self.logger.increment('removes.' + device_name)

    def put_object(self, name, timestamp, size, content_type, etag, deleted=0, storage_policy_index=0):
        """
        Creates an object in the DB with its metadata.

        :param name: object name to be created
        :param timestamp: timestamp of when the object was created
        :param size: object size
        :param content_type: object content-type
        :param etag: object etag
        :param deleted: if True, marks the object as deleted and sets the
                        deleted_at timestamp to timestamp
        :param storage_policy_index: the storage policy index for the object
        """
        record = {
            "name": name,
            "created_at": timestamp,
            "size": size,
            "content_type": content_type,
            "etag": etag,
            "deleted": deleted,
            "storage_policy_index": storage_policy_index,
        }
        if self.db_file == ":memory:":
            self.merge_items([record])
            return
        if not os.path.exists(self.db_file):
            raise DatabaseConnectionError(self.db_file, "DB doesn't exist")
        pending_size = 0
        try:
            pending_size = os.path.getsize(self.pending_file)
        except OSError as err:
            if err.errno != errno.ENOENT:
                raise
        if pending_size > PENDING_CAP:
            self._commit_puts([record])
        else:
            with lock_parent_directory(self.pending_file, self.pending_timeout):
                with open(self.pending_file, "a+b") as fp:
                    # Colons aren't used in base64 encoding; so they are our
                    # delimiter
                    fp.write(":")
                    fp.write(
                        pickle.dumps(
                            (name, timestamp, size, content_type, etag, deleted, storage_policy_index),
                            protocol=PICKLE_PROTOCOL,
                        ).encode("base64")
                    )
                    fp.flush()

    def put_container(self, name, put_timestamp, delete_timestamp,
                      object_count, bytes_used, storage_policy_index):
        """
        Create a container with the given attributes.

        :param name: name of the container to create
        :param put_timestamp: put_timestamp of the container to create
        :param delete_timestamp: delete_timestamp of the container to create
        :param object_count: number of objects in the container
        :param bytes_used: number of bytes used by the container
        :param storage_policy_index:  the storage policy for this container
        """
        if delete_timestamp > put_timestamp and \
                object_count in (None, '', 0, '0'):
            deleted = 1
        else:
            deleted = 0
        record = {'name': name, 'put_timestamp': put_timestamp,
                  'delete_timestamp': delete_timestamp,
                  'object_count': object_count,
                  'bytes_used': bytes_used,
                  'deleted': deleted,
                  'storage_policy_index': storage_policy_index}
        if self.db_file == ':memory:':
            self.merge_items([record])
            return
        if not os.path.exists(self.db_file):
            raise DatabaseConnectionError(self.db_file, "DB doesn't exist")
        pending_size = 0
        try:
            pending_size = os.path.getsize(self.pending_file)
        except OSError as err:
            if err.errno != errno.ENOENT:
                raise
        if pending_size > PENDING_CAP:
            self._commit_puts([record])
        else:
            with lock_parent_directory(self.pending_file,
                                       self.pending_timeout):
                with open(self.pending_file, 'a+b') as fp:
                    # Colons aren't used in base64 encoding; so they are our
                    # delimiter
                    fp.write(':')
                    fp.write(pickle.dumps(
                        (name, put_timestamp, delete_timestamp, object_count,
                         bytes_used, deleted, storage_policy_index),
                        protocol=PICKLE_PROTOCOL).encode('base64'))
                    fp.flush()

    def put_container(self, name, put_timestamp, delete_timestamp, object_count, bytes_used):
        """
        Create a container with the given attributes.

        :param name: name of the container to create
        :param put_timestamp: put_timestamp of the container to create
        :param delete_timestamp: delete_timestamp of the container to create
        :param object_count: number of objects in the container
        :param bytes_used: number of bytes used by the container
        """
        if delete_timestamp > put_timestamp and object_count in (None, "", 0, "0"):
            deleted = 1
        else:
            deleted = 0
        record = {
            "name": name,
            "put_timestamp": put_timestamp,
            "delete_timestamp": delete_timestamp,
            "object_count": object_count,
            "bytes_used": bytes_used,
            "deleted": deleted,
        }
        if self.db_file == ":memory:":
            self.merge_items([record])
            return
        if not os.path.exists(self.db_file):
            raise DatabaseConnectionError(self.db_file, "DB doesn't exist")
        pending_size = 0
        try:
            pending_size = os.path.getsize(self.pending_file)
        except OSError as err:
            if err.errno != errno.ENOENT:
                raise
        if pending_size > PENDING_CAP:
            self._commit_puts([record])
        else:
            with lock_parent_directory(self.pending_file, self.pending_timeout):
                with open(self.pending_file, "a+b") as fp:
                    # Colons aren't used in base64 encoding; so they are our
                    # delimiter
                    fp.write(":")
                    fp.write(
                        pickle.dumps(
                            (name, put_timestamp, delete_timestamp, object_count, bytes_used, deleted),
                            protocol=PICKLE_PROTOCOL,
                        ).encode("base64")
                    )
                    fp.flush()

    def dump_builder(self):
        """Write out new builder files

        :param builder: The builder to dump
        :builder_file: The builder file to write to
        """
        with lock_parent_directory(self.builder_file, 15):
            bfile, bmd5 = self._make_backup()
            print "Backed up %s to %s (%s)" % (self.builder_file, bfile, bmd5)
            fd, tmppath = mkstemp(
                dir=dirname(self.builder_file), suffix='.tmp.builder')
            pickle.dump(self.builder.to_dict(), fdopen(fd, 'wb'), protocol=2)
            rename(tmppath, self.builder_file)
        print "Success. %s updated. (%s)" % (
            self.builder_file, self.get_md5sum(self.builder_file))
        print "Rebalance still required."

    def put_container(self, name, put_timestamp, delete_timestamp, object_count, bytes_used):
        """
        Create a container with the given attributes.

        :param name: name of the container to create
        :param put_timestamp: put_timestamp of the container to create
        :param delete_timestamp: delete_timestamp of the container to create
        :param object_count: number of objects in the container
        :param bytes_used: number of bytes used by the container
        """
        if delete_timestamp > put_timestamp and object_count in (None, "", 0, "0"):
            deleted = 1
        else:
            deleted = 0
        record = {
            "name": name,
            "put_timestamp": put_timestamp,
            "delete_timestamp": delete_timestamp,
            "object_count": object_count,
            "bytes_used": bytes_used,
            "deleted": deleted,
        }
        if self.db_file == ":memory:":
            self.merge_items([record])
            return
        commit = False
        with lock_parent_directory(self.pending_file, self.pending_timeout):
            with open(self.pending_file, "a+b") as fp:
                # Colons aren't used in base64 encoding; so they are our
                # delimiter
                fp.write(":")
                fp.write(
                    pickle.dumps(
                        (name, put_timestamp, delete_timestamp, object_count, bytes_used, deleted),
                        protocol=PICKLE_PROTOCOL,
                    ).encode("base64")
                )
                fp.flush()
                if fp.tell() > PENDING_CAP:
                    commit = True
        if commit:
            self._commit_puts()

 def _commit_puts(self, item_list=None):
     """Handles commiting rows in .pending files."""
     if self.db_file == ":memory:" or not os.path.exists(self.pending_file):
         return
     if item_list is None:
         item_list = []
     with lock_parent_directory(self.pending_file, self.pending_timeout):
         self._preallocate()
         if not os.path.getsize(self.pending_file):
             if item_list:
                 self.merge_items(item_list)
             return
         with open(self.pending_file, "r+b") as fp:
             for entry in fp.read().split(":"):
                 if entry:
                     try:
                         (name, put_timestamp, delete_timestamp, object_count, bytes_used, deleted) = pickle.loads(
                             entry.decode("base64")
                         )
                         item_list.append(
                             {
                                 "name": name,
                                 "put_timestamp": put_timestamp,
                                 "delete_timestamp": delete_timestamp,
                                 "object_count": object_count,
                                 "bytes_used": bytes_used,
                                 "deleted": deleted,
                             }
                         )
                     except Exception:
                         self.logger.exception(
                             _("Invalid pending entry %(file)s: %(entry)s"),
                             {"file": self.pending_file, "entry": entry},
                         )
             if item_list:
                 self.merge_items(item_list)
             try:
                 os.ftruncate(fp.fileno(), 0)
             except OSError, err:
                 if err.errno != errno.ENOENT:
                     raise

    def put_object(self, name, timestamp, size, content_type, etag,
                   deleted=0, metadata=None):
        """
        Creates an object in the DB with its metadata.

        :param name: object name to be created
        :param timestamp: timestamp of when the object was created
        :param size: object size
        :param content_type: object content-type
        :param etag: object etag
        :param deleted: if True, marks the object as deleted and sets the
                        deteleted_at timestamp to timestamp
        """
        record = {'name': name, 'created_at': timestamp, 'size': size,
                  'content_type': content_type, 'etag': etag,
                  'metadata': metadata, 'deleted': deleted}
        if self.db_file == ':memory:':
            self.merge_items([record])
            return
        if not os.path.exists(self.db_file):
            raise DatabaseConnectionError(self.db_file, "DB doesn't exist")
        pending_size = 0
        try:
            pending_size = os.path.getsize(self.pending_file)
        except OSError as err:
            if err.errno != errno.ENOENT:
                raise
        if pending_size > PENDING_CAP:
            self._commit_puts([record])
        else:
            with lock_parent_directory(self.pending_file,
                                       self.pending_timeout):
                with open(self.pending_file, 'a+b') as fp:
                    # Colons aren't used in base64 encoding; so they are our
                    # delimiter
                    fp.write(':')
                    fp.write(pickle.dumps(
                        (name, timestamp, size, content_type, etag, deleted,
                         metadata),
                        protocol=PICKLE_PROTOCOL).encode('base64'))
                    fp.flush()

    def put_record(self, record):
        """
        Put a record into the DB. If the DB has an associated pending file with
        space then the record is appended to that file and a commit to the DB
        is deferred. If the DB is in-memory or its pending file is full then
        the record will be committed immediately.

        :param record: a record to be added to the DB.
        :raises DatabaseConnectionError: if the DB file does not exist or if
            ``skip_commits`` is True.
        :raises LockTimeout: if a timeout occurs while waiting to take a lock
            to write to the pending file.
        """
        if self._db_file == ':memory:':
            self.merge_items([record])
            return
        if not os.path.exists(self.db_file):
            raise DatabaseConnectionError(self.db_file, "DB doesn't exist")
        if self.skip_commits:
            raise DatabaseConnectionError(self.db_file,
                                          'commits not accepted')
        with lock_parent_directory(self.pending_file, self.pending_timeout):
            pending_size = 0
            try:
                pending_size = os.path.getsize(self.pending_file)
            except OSError as err:
                if err.errno != errno.ENOENT:
                    raise
            if pending_size > PENDING_CAP:
                self._commit_puts([record])
            else:
                with open(self.pending_file, 'a+b') as fp:
                    # Colons aren't used in base64 encoding; so they are our
                    # delimiter
                    fp.write(b':')
                    fp.write(base64.b64encode(pickle.dumps(
                        self.make_tuple_for_pickle(record),
                        protocol=PICKLE_PROTOCOL)))
                    fp.flush()

    def reclaim(self, age_timestamp, sync_timestamp):
        """
        Delete reclaimable rows and metadata from the db.

        By default this method will delete rows from the db_contains_type table
        that are marked deleted and whose created_at timestamp is <
        age_timestamp, and deletes rows from incoming_sync and outgoing_sync
        where the updated_at timestamp is < sync_timestamp. In addition, this
        calls the :meth:`_reclaim_metadata` method.

        Subclasses may reclaim other items by overriding :meth:`_reclaim`.

        :param age_timestamp: max created_at timestamp of object rows to delete
        :param sync_timestamp: max update_at timestamp of sync rows to delete
        """
        if not self._skip_commit_puts():
            with lock_parent_directory(self.pending_file,
                                       self.pending_timeout):
                self._commit_puts()
        with self.get() as conn:
            self._reclaim(conn, age_timestamp, sync_timestamp)
            self._reclaim_metadata(conn, age_timestamp)
            conn.commit()

 def put_record(self, record):
     if self.db_file == ':memory:':
         self.merge_items([record])
         return
     if not os.path.exists(self.db_file):
         raise DatabaseConnectionError(self.db_file, "DB doesn't exist")
     with lock_parent_directory(self.pending_file, self.pending_timeout):
         pending_size = 0
         try:
             pending_size = os.path.getsize(self.pending_file)
         except OSError as err:
             if err.errno != errno.ENOENT:
                 raise
         if pending_size > PENDING_CAP:
             self._commit_puts([record])
         else:
             with open(self.pending_file, 'a+b') as fp:
                 # Colons aren't used in base64 encoding; so they are our
                 # delimiter
                 fp.write(':')
                 fp.write(pickle.dumps(
                     self.make_tuple_for_pickle(record),
                     protocol=PICKLE_PROTOCOL).encode('base64'))
                 fp.flush()