Merge lp:~stub/launchpad/librarian-gc into lp:launchpad

[Stuart Bishop]

[Stuart Bishop (stub)]

Address Bug #290837.

The Librarian garbage collector currently drives the system into swap because it is doing its set logic in RAM. We should stop doing this.

[Aaron Bentley (abentley)]

-----BEGIN PGP SIGNED MESSAGE-----
Hash: SHA1

 review needs-fixing

Stuart Bishop wrote:
> Stuart Bishop has proposed merging lp:~stub/launchpad/librarian-gc into lp:~launchpad-pqm/launchpad/devel.
>
> Requested reviews:
> Canonical Launchpad Engineering (launchpad)
>
> Address Bug #290837.
>
> The Librarian garbage collector currently drives the system into swap because it is doing its set logic in RAM. We should stop doing this.
>
> === modified file 'lib/canonical/librarian/ftests/test_gc.py'
> --- lib/canonical/librarian/ftests/test_gc.py 2009-01-06 03:40:57 +0000
> +++ lib/canonical/librarian/ftests/test_gc.py 2009-07-09 12:41:44 +0000
> @@ -1,4 +1,4 @@
> -# Copyright 2004-2007 Canonical Ltd. All rights reserved.
> +# Copyright 2004-2009 Canonical Ltd. All rights reserved.
> """Librarian garbage collection tests"""
>
> __metaclass__ = type
> @@ -119,6 +119,8 @@
> past = past.replace(tzinfo=utc)
> f1.last_accessed = past
> f2.last_accessed = past
> + f1.date_created = past
> + f2.date_created = past
>
> del f1, f2
>
> @@ -159,7 +161,7 @@
> # This should have committed
> self.ztm.begin()
>
> - # Confirm that the LibaryFileContents are still there
> + # Confirm that the LibaryFileContents are still there.
> c1 = LibraryFileContent.get(c1_id)
> c2 = LibraryFileContent.get(c2_id)
>
>
> === modified file 'lib/canonical/librarian/librariangc.py'
> --- lib/canonical/librarian/librariangc.py 2009-01-20 08:15:41 +0000
> +++ lib/canonical/librarian/librariangc.py 2009-07-09 12:40:21 +0000
> @@ -1,4 +1,4 @@
> -# Copyright 2004-2008 Canonical Ltd. All rights reserved.
> +# Copyright 2004-2009 Canonical Ltd. All rights reserved.
> """Librarian garbage collection routines"""
>
> __metaclass__ = type
> @@ -8,13 +8,15 @@
> from time import time
> import os
>
> +from zope.interface import implements
> +
> from canonical.config import config
> +from canonical.launchpad.interfaces.looptuner import ITunableLoop
> +from canonical.launchpad.utilities.looptuner import DBLoopTuner
> from canonical.librarian.storage import _relFileLocation as relative_file_path
> from canonical.librarian.storage import _sameFile
> from canonical.database.postgresql import listReferences
>
> -BATCH_SIZE = 1
> -
> log = None
> debug = False
>
> @@ -176,8 +178,11 @@
> con.commit()
>
>
> -def delete_unreferenced_aliases(con):
> - """Delete unreferenced LibraryFileAliases and their LibraryFileContent
> +class UnreferencedLibraryFileContentPruner:
> + """Delete unreferenced LibraryFileAliases.
> +
> + The LibraryFileContent records are left untouched for the code that
> + knows how to delete them and the corresponding files on disk.
>
> This is the second step in a full garbage collection sweep. We determine
> which LibraryFileContent entries are not being referenced by other objects
> @@ -190,92 +195,124 @@
> must be unreferenced for them to be deleted - a single reference will keep
> the whole set alive.
> """
> - log.info("Deleting unreferenced LibraryFileAliases")
> -
> - # Generate ...

review: Needs Fixing

[Stuart Bishop (stub)]

On Mon, Jul 13, 2009 at 10:04 PM, Aaron Bentley<email address hidden> wrote:

>  review needs-fixing
>
> Stuart Bishop wrote:
>> Stuart Bishop has proposed merging lp:~stub/launchpad/librarian-gc into lp:~launchpad-pqm/launchpad/devel.
>>
>> Requested reviews:
>>     Canonical Launchpad Engineering (launchpad)
>>
>> Address Bug #290837.
>>
>> The Librarian garbage collector currently drives the system into swap because it is doing its set logic in RAM. We should stop doing this.

>> === modified file 'lib/canonical/librarian/librariangc.py'
>> +    implements(ITunableLoop)
>> +
>> +    index = 1 # Index of current batch of rows to remove.
>> +
>> +    total_deleted = 0 # Running total
>> +
>> +    con = None # Database connection to use
>
> "index", "total_deleted" and "con" are instance variables.  It would not
> make sense to check the value of
> UnreferencedLibraryFileContentPruner.total_deleted, ever.  You would
> only check it on an instance of UnreferencedLibraryFileContentPruner.
> Therefore, they should be initialized in
> UnreferencedLibraryFileContentPruner.__init__, instead.

Ok. These are all being initialized in __init__ and the class attributes removed here and in the other two tunable loops.

>> +        # Start with a list of all unexpired and recently accessed content
>> +        # - we don't remove them even if they are unlinked.
>> +        # We currently don't remove stuff until it has been expired for
>> +        # more than one week, but we will change this if disk space becomes
>> +        # short and it actually will make a noticeable difference.
>> +        # Note that ReferencedLibraryFileContent will contain duplicates -
>> +        # duplicates are the exception so we are better off filtering them
>
> Perhaps "duplicates are unusual", "duplicates are exceptional", or else
> "duplicates are the exception, not the rule".

Fixed. "duplicates are unusual".

>> +        # once at the end rather than when we load the data into the temporary
>> +        # file.
>> +        cur.execute("DROP TABLE IF EXISTS ReferencedLibraryFileContent")
>> +        cur.execute("""
>> +            SELECT LibraryFileAlias.content
>> +            INTO TEMPORARY TABLE ReferencedLibraryFileContent
>>              FROM LibraryFileAlias
>> -            WHERE content in (%(in_content_ids)s)
>> -                AND (
>> -                    expires + '1 week'::interval
>> -                        > CURRENT_TIMESTAMP AT TIME ZONE 'UTC'
>> -                    OR last_accessed + '1 week'::interval
>> -                        > CURRENT_TIMESTAMP AT TIME ZONE 'UTC'
>> -                    )
>> -            """ % vars())
>> -        assert cur.fetchone()[0] == 0, "Logic error - sanity check failed"
>> +            WHERE
>> +                expires >
>> +                    CURRENT_TIMESTAMP AT TIME ZONE 'UTC' - interval '1 week'
>> +                OR last_accessed >
>> +                    CURRENT_TIMESTAMP AT TIME ZONE 'UTC' - interval '1 week'
>> +                OR date_created >
>> +                    CURRENT_TIMESTAMP AT TIME ZONE 'UTC' - interval '1 week'
>> +            """...

[Aaron Bentley (abentley)]

-----BEGIN PGP SIGNED MESSAGE-----
Hash: SHA1

 status approved

Please add a comment to the global log variable, as noted below. The
rest is chatter.

Stuart Bishop wrote:
>>> + log.debug("Getting references from %s.%s." % (table, column))
>>> + cur.execute("""
>>> + INSERT INTO ReferencedLibraryFileContent
>>> + SELECT LibraryFileAlias.content
>>> + FROM LibraryFileAlias, %(table)s
>>> + WHERE LibraryFileAlias.id = %(table)s.%(column)s
>>> + """ % {'table': table, 'column': column})
>> Wouldn't it be better practise to use SQL substitution here, like
>> ...WHERE LibraryFileAlias.id = ?.?", table, column)
>
> We can't use bound variables here because SQL identifiers use different quoting rules to SQL strings (" vs ').
>
>> I doubt any of our table names are SQL keywords, but some of our columns
>> might be named "from", for example.
>
> Yes - it is fragile. I've fixed this by doing the quoting myself using canonical.database.postgresql.quoteIdentifier()

Cool.

>> Is it actually performant to generate a table of *referenced*
>> LibraryFileContent? I would have thought you'd want to generate a table
>> of unreferenced content directly.
>
> I don't really know which approach would perform better. I don't think it is worth benchmarking unless performance turns out to be an issue. I suspect this approach is faster, as if we started with a table containing all LFC and deleted rows when we find references, we end up working with a table that is mostly holes.

Oh, I meant that it seemed like putting it all into one query that only
returned the rows we wanted to play with would allow the DB engine to do
less work, ideally scaling with the number of rows returned instead of
the total number of rows. As you say, if it's not a performance issue,
let's not worry about it.

>>> + cur.execute("DROP TABLE ReferencedLibraryFileContent")
>>> + cur.execute(
>>> + "SELECT COALESCE(max(id),0) FROM UnreferencedLibraryFileContent")
>>> + self.max_id = cur.fetchone()[0]
>>> + log.debug("%d unferenced LibraryFileContent to remove." % self.max_id)
>>> con.commit()
>> Is there any risk that some LibraryFileContent might become referenced
>> after you commit?
>
> There is a risk. I mitigate it by not deleting anything younger than a day - if it hasn't been linked by then it probably isn't going to be linked. I could try catching the exception, aborting the database transaction and possibly retry the batch without the failed entry but I doubt it will ever actually happen and I'd rather not write the tests for this extra complexity :) At the moment, if this condition is actually triggered the foreign key constraints will cause an exception to be raised. I vote we worry about that if it happens more than once.

I guess I'm really wondering why not do it all in one transaction to
avoid that race?

>>> + def __call__(self, chunksize):
>>> + chunksize = int(chunksize)
>> It seems odd that this might not be an int.
>
> The loop tuner is probably too generic. You actually get a float in case your chunksize represe...

review: Approve

=== modified file 'lib/canonical/librarian/ftests/test_gc.py'
--- lib/canonical/librarian/ftests/test_gc.py	2009-01-06 03:40:57 +0000
+++ lib/canonical/librarian/ftests/test_gc.py	2009-07-16 02:56:16 +0000
@@ -1,4 +1,4 @@
-# Copyright 2004-2007 Canonical Ltd.  All rights reserved.
+# Copyright 2004-2009 Canonical Ltd.  All rights reserved.
 """Librarian garbage collection tests"""
 
 __metaclass__ = type
@@ -119,6 +119,8 @@
         past = past.replace(tzinfo=utc)
         f1.last_accessed = past
         f2.last_accessed = past
+        f1.date_created = past
+        f2.date_created = past
 
         del f1, f2
 
@@ -159,7 +161,7 @@
         # This should have committed
         self.ztm.begin()
 
-        # Confirm that the LibaryFileContents are still there
+        # Confirm that the LibaryFileContents are still there.
         c1 = LibraryFileContent.get(c1_id)
         c2 = LibraryFileContent.get(c2_id)
 

=== modified file 'lib/canonical/librarian/librariangc.py'
--- lib/canonical/librarian/librariangc.py	2009-01-20 08:15:41 +0000
+++ lib/canonical/librarian/librariangc.py	2009-07-16 02:56:16 +0000
@@ -1,20 +1,24 @@
-# Copyright 2004-2008 Canonical Ltd.  All rights reserved.
+# Copyright 2004-2009 Canonical Ltd.  All rights reserved.
 """Librarian garbage collection routines"""
 
 __metaclass__ = type
 
 from datetime import datetime, timedelta
+import errno
 import sys
 from time import time
 import os
 
+from zope.interface import implements
+
 from canonical.config import config
+from canonical.database.postgresql import quoteIdentifier
+from canonical.launchpad.interfaces.looptuner import ITunableLoop
+from canonical.launchpad.utilities.looptuner import DBLoopTuner
 from canonical.librarian.storage import _relFileLocation as relative_file_path
 from canonical.librarian.storage import _sameFile
 from canonical.database.postgresql import listReferences
 
-BATCH_SIZE = 1
-
 log = None
 debug = False
 
@@ -176,8 +180,11 @@
         con.commit()
 
 
-def delete_unreferenced_aliases(con):
-    """Delete unreferenced LibraryFileAliases and their LibraryFileContent
+class UnreferencedLibraryFileContentPruner:
+    """Delete unreferenced LibraryFileAliases.
+
+    The LibraryFileContent records are left untouched for the code that
+    knows how to delete them and the corresponding files on disk.
 
     This is the second step in a full garbage collection sweep. We determine
     which LibraryFileContent entries are not being referenced by other objects
@@ -190,92 +197,122 @@
     must be unreferenced for them to be deleted - a single reference will keep
     the whole set alive.
     """
-    log.info("Deleting unreferenced LibraryFileAliases")
-
-    # Generate a set of all our LibraryFileContent ids, except for ones
-    # with expiry dates not yet reached (these lurk in the database until
-    # expired) and those that have been accessed in the last week (currently
-    # in use, so leave them lurking a while longer)
-    cur = con.cursor()
-    cur.execute("""
-        SELECT c.id
-        FROM LibraryFileContent AS c, LibraryFileAlias AS a
-        WHERE c.id = a.content
-        GROUP BY c.id
-        HAVING (max(expires) IS NULL OR max(expires)
-                < CURRENT_TIMESTAMP AT TIME ZONE 'UTC' - '1 week'::interval
-                )
-            AND (max(last_accessed) IS NULL OR max(last_accessed)
-                < CURRENT_TIMESTAMP AT TIME ZONE 'UTC' - '1 week'::interval
-                )
-        """)
-    content_ids = set(row[0] for row in cur.fetchall())
-    log.info(
-        "Found %d LibraryFileContent entries possibly unreferenced",
-        len(content_ids)
-        )
-
-    # Determine what columns link to LibraryFileAlias
-    # references = [(table, column), ...]
-    references = [
-        tuple(ref[:2])
-        for ref in listReferences(cur, 'libraryfilealias', 'id')
-        if ref[0] != 'libraryfiledownloadcount'
-        ]
-    assert len(references) > 10, 'Database introspection returned nonsense'
-    log.info("Found %d columns referencing LibraryFileAlias", len(references))
-
-    # Remove all referenced LibraryFileContent ids from content_ids
-    for table, column in references:
-        cur.execute("""
-            SELECT DISTINCT LibraryFileContent.id
-            FROM LibraryFileContent, LibraryFileAlias, %(table)s
-            WHERE LibraryFileContent.id = LibraryFileAlias.content
-                AND LibraryFileAlias.id = %(table)s.%(column)s
-            """ % vars())
-        referenced_ids = set(row[0] for row in cur.fetchall())
-        log.info(
-                "Found %d distinct LibraryFileAlias references in %s(%s)",
-                len(referenced_ids), table, column
-                )
-        content_ids.difference_update(referenced_ids)
-        log.debug(
-                "Now only %d LibraryFileContents possibly unreferenced",
-                len(content_ids)
-                )
-
-    # Delete unreferenced LibraryFileAliases. Note that this will raise a
-    # database exception if we screwed up and attempt to delete an alias that
-    # is still referenced.
-    content_ids = list(content_ids)
-    for i in range(0, len(content_ids), BATCH_SIZE):
-        in_content_ids = ','.join(
-            (str(content_id) for content_id in content_ids[i:i+BATCH_SIZE]))
-        # First a sanity check to ensure we aren't removing anything we
-        # shouldn't be.
-        cur.execute("""
-            SELECT COUNT(*)
+    implements(ITunableLoop)
+
+    def __init__(self, con):
+        self.con = con # Database connection to use
+        self.total_deleted = 0 # Running total
+        self.index = 1
+
+        log.info("Deleting unreferenced LibraryFileAliases")
+
+        cur = con.cursor()
+
+        # Start with a list of all unexpired and recently accessed
+        # content - we don't remove them even if they are unlinked. We
+        # currently don't remove stuff until it has been expired for
+        # more than one week, but we will change this if disk space
+        # becomes short and it actually will make a noticeable
+        # difference. Note that ReferencedLibraryFileContent will
+        # contain duplicates - duplicates are unusual so we are better
+        # off filtering them once at the end rather than when we load
+        # the data into the temporary file.
+        cur.execute("DROP TABLE IF EXISTS ReferencedLibraryFileContent")
+        cur.execute("""
+            SELECT LibraryFileAlias.content
+            INTO TEMPORARY TABLE ReferencedLibraryFileContent
             FROM LibraryFileAlias
-            WHERE content in (%(in_content_ids)s)
-                AND (
-                    expires + '1 week'::interval
-                        > CURRENT_TIMESTAMP AT TIME ZONE 'UTC'
-                    OR last_accessed + '1 week'::interval
-                        > CURRENT_TIMESTAMP AT TIME ZONE 'UTC'
-                    )
-            """ % vars())
-        assert cur.fetchone()[0] == 0, "Logic error - sanity check failed"
+            WHERE
+                expires >
+                    CURRENT_TIMESTAMP AT TIME ZONE 'UTC' - interval '1 week'
+                OR last_accessed >
+                    CURRENT_TIMESTAMP AT TIME ZONE 'UTC' - interval '1 week'
+                OR date_created >
+                    CURRENT_TIMESTAMP AT TIME ZONE 'UTC' - interval '1 week'
+            """)
+        con.commit()
+
+        # Determine what columns link to LibraryFileAlias
+        # references = [(table, column), ...]
+        references = [
+            tuple(ref[:2])
+            for ref in listReferences(cur, 'libraryfilealias', 'id')
+            if ref[0] != 'libraryfiledownloadcount'
+            ]
+        assert len(references) > 10, (
+            'Database introspection returned nonsense')
         log.debug(
-                "Deleting all LibraryFileAlias references to "
-                "LibraryFileContents %s", in_content_ids
-                )
-        cur.execute("""
-            DELETE FROM LibraryFileAlias WHERE content IN (%(in_content_ids)s)
-            """ % vars())
+            "Found %d columns referencing LibraryFileAlias", len(references))
+
+        # Find all relevant LibraryFileAlias references and fill in
+        # ReferencedLibraryFileContent
+        for table, column in references:
+            log.debug("Getting references from %s.%s." % (table, column))
+            cur.execute("""
+                INSERT INTO ReferencedLibraryFileContent
+                SELECT LibraryFileAlias.content
+                FROM LibraryFileAlias, %(table)s
+                WHERE LibraryFileAlias.id = %(table)s.%(column)s
+                """ % {
+                    'table': quoteIdentifier(table),
+                    'column': quoteIdentifier(column)})
+            con.commit()
+
+        log.debug("Calculating expired unreferenced LibraryFileContent set.")
+        cur.execute("DROP TABLE IF EXISTS UnreferencedLibraryFileContent")
+        cur.execute("""
+            CREATE TEMPORARY TABLE UnreferencedLibraryFileContent (
+                id serial PRIMARY KEY,
+                content integer UNIQUE)
+            """)
+        cur.execute("""
+            INSERT INTO UnreferencedLibraryFileContent (content)
+            SELECT id AS content FROM LibraryFileContent
+            EXCEPT
+            SELECT content FROM ReferencedLibraryFileContent
+            """)
+        cur.execute("DROP TABLE ReferencedLibraryFileContent")
+        cur.execute(
+            "SELECT COALESCE(max(id),0) FROM UnreferencedLibraryFileContent")
+        self.max_id = cur.fetchone()[0]
+        log.debug("%d unferenced LibraryFileContent to remove." % self.max_id)
         con.commit()
 
-
-def delete_unreferenced_content(con):
+    def isDone(self):
+        if self.index > self.max_id:
+            log.info(
+                "Deleted %d LibraryFileAlias records." % self.total_deleted)
+            return True
+        else:
+            return False
+
+    def __call__(self, chunksize):
+        chunksize = int(chunksize)
+        cur = self.con.cursor()
+        cur.execute("""
+            DELETE FROM LibraryFileAlias
+            USING (
+                SELECT content FROM UnreferencedLibraryFileContent
+                WHERE id BETWEEN %s AND %s
+                ) AS UnreferencedLibraryFileContent
+            WHERE LibraryFileAlias.content
+                = UnreferencedLibraryFileContent.content
+            """, (self.index, self.index + chunksize - 1))
+        deleted_rows = cur.rowcount
+        self.total_deleted += deleted_rows
+        log.debug("Deleted %d LibraryFileAlias records." % deleted_rows)
+        self.con.commit()
+        self.index += chunksize
+
+
+def delete_unreferenced_aliases(con):
+    "Run the UnreferencedLibraryFileContentPruner."
+    loop_tuner = DBLoopTuner(
+        UnreferencedLibraryFileContentPruner(con), 5, log=log)
+    loop_tuner.run()
+
+
+class UnreferencedContentPruner:
     """Delete LibraryFileContent entries and their disk files that are
     not referenced by any LibraryFileAlias entries.
 
@@ -283,87 +320,159 @@
     LibraryFileAlias, so all entries in this state are garbage no matter
     what their expires flag says.
     """
-    cur = con.cursor()
-    cur.execute("""
-        SELECT LibraryFileContent.id
-        FROM LibraryFileContent
-        LEFT OUTER JOIN LibraryFileAlias
-            ON LibraryFileContent.id = LibraryFileAlias.content
-        WHERE LibraryFileAlias.content IS NULL
+    implements(ITunableLoop)
+
+    def __init__(self, con):
+        self.con = con
+        self.index = 1
+        self.total_deleted = 0
+        cur = con.cursor()
+        cur.execute("DROP TABLE IF EXISTS UnreferencedLibraryFileContent")
+        cur.execute("""
+            CREATE TEMPORARY TABLE UnreferencedLibraryFileContent (
+                id serial PRIMARY KEY,
+                content integer UNIQUE)
+            """)
+        cur.execute("""
+            INSERT INTO UnreferencedLibraryFileContent (content)
+            SELECT DISTINCT LibraryFileContent.id
+            FROM LibraryFileContent
+            LEFT OUTER JOIN LibraryFileAlias
+                ON LibraryFileContent.id = LibraryFileAlias.content
+            WHERE LibraryFileAlias.content IS NULL
         """)
-    garbage_ids = [row[0] for row in cur.fetchall()]
-
-    for i in range(0, len(garbage_ids), BATCH_SIZE):
-        in_garbage_ids = ','.join(
-            (str(garbage_id) for garbage_id in garbage_ids[i:i+BATCH_SIZE])
-            )
-
-        # Delete old LibraryFileContent entries. Note that this will fail
-        # if we screwed up and still have LibraryFileAlias entries referencing
-        # it.
-        log.debug("Deleting LibraryFileContents %s", in_garbage_ids)
-        cur.execute("""
-            DELETE FROM LibraryFileContent WHERE id in (%s)
-            """ % in_garbage_ids)
-
-        for garbage_id in garbage_ids[i:i+BATCH_SIZE]:
+        cur.execute("""
+            SELECT COALESCE(max(id), 0) FROM UnreferencedLibraryFileContent
+            """)
+        self.max_id = cur.fetchone()[0]
+        log.debug("%d unreferenced LibraryFileContent rows to remove.")
+
+    def isDone(self):
+        if self.index > self.max_id:
+            log.info("Deleted %d unreferenced files." % self.total_deleted)
+            return True
+        else:
+            return False
+
+    def __call__(self, chunksize):
+        chunksize = int(chunksize)
+
+        cur = self.con.cursor()
+
+        # Delete unreferenced LibraryFileContent entries.
+        cur.execute("""
+            DELETE FROM LibraryFileContent
+            USING (
+                SELECT content FROM UnreferencedLibraryFileContent
+                WHERE id BETWEEN %s AND %s) AS UnreferencedLibraryFileContent
+            WHERE
+                LibraryFileContent.id = UnreferencedLibraryFileContent.content
+            """, (self.index, self.index + chunksize - 1))
+        rows_deleted = cur.rowcount
+        self.total_deleted += rows_deleted
+        self.con.commit()
+
+        # Remove files from disk. We do this outside the transaction,
+        # as the garbage collector happily deals with files that exist
+        # on disk but not in the DB.
+        cur.execute("""
+            SELECT content FROM UnreferencedLibraryFileContent
+            WHERE id BETWEEN %s AND %s
+            """, (self.index, self.index + chunksize - 1))
+        for content_id in (row[0] for row in cur.fetchall()):
             # Remove the file from disk, if it hasn't already been
-            path = get_file_path(garbage_id)
-            if os.path.exists(path):
-                log.debug("Deleting %s", path)
+            path = get_file_path(content_id)
+            try:
                 os.unlink(path)
-            else:
-                log.info("%s already deleted", path)
-
-        # And commit the database changes. It may be possible for this to
-        # fail in rare cases, leaving a record in the DB with no corresponding
-        # file on disk, but that is OK as it will all be tidied up next run,
-        # and the file is unreachable anyway so nothing will attempt to
-        # access it between now and the next garbage collection run.
-        con.commit()
-
-
-def flag_expired_files(connection):
+                log.debug("Deleted %s", path)
+            except OSError, e:
+                if e.errno != errno.ENOENT:
+                    raise
+                if config.librarian_server.upstream_host is None:
+                    # It is normal to have files in the database that
+                    # are not on disk if the Librarian has an upstream
+                    # Librarian, such as on staging. Don't annoy the
+                    # operator with noise in this case.
+                    log.info("%s already deleted", path)
+
+        self.index += chunksize
+
+
+def delete_unreferenced_content(con):
+    """Invoke UnreferencedContentPruner."""
+    loop_tuner = DBLoopTuner(UnreferencedContentPruner(con), 5, log=log)
+    loop_tuner.run()
+
+
+class FlagExpiredFiles:
     """Flag files past their expiry date as 'deleted' in the database.
 
     Actual removal from disk is not performed here - that is deferred to
     delete_unwanted_files().
     """
-    cur = connection.cursor()
-
-    # First get the list of all LibraryFileContent.
-    cur.execute("SELECT id FROM LibraryFileContent WHERE deleted IS FALSE")
-    all_ids = set(row[0] for row in cur.fetchall())
-
-    # Now the list of unexpired content. May contain some ids not in the
-    # all_ids set if uploads are currently in progress.
-    cur.execute("""
-        SELECT DISTINCT content
-        FROM LibraryFileAlias
-        WHERE expires IS NULL
-            OR expires >= CURRENT_TIMESTAMP AT TIME ZONE 'UTC'
-        """)
-    unexpired_ids = set(row[0] for row in cur.fetchall())
-
-    # Destroy our all_ids set to create the set of expired ids.
-    # We do it this way, as we are dealing with large sets and need to
-    # be careful of RAM usage on the production server.
-    all_ids -= unexpired_ids
-    expired_ids = all_ids
-    del all_ids
-    del unexpired_ids
-
-    for commit_counter, content_id in enumerate(expired_ids):
-        log.debug("%d is expired." % content_id)
-        cur = connection.cursor()
+    implements(ITunableLoop)
+
+    def __init__(self, con):
+        self.con = con
+        self.index = 1
+        self.total_flagged = 0
+        cur = con.cursor()
+
+        log.debug("Creating set of expired LibraryFileContent.")
+        cur.execute("DROP TABLE IF EXISTS ExpiredLibraryFileContent")
+        cur.execute("""
+            CREATE TEMPORARY TABLE ExpiredLibraryFileContent
+            (id serial PRIMARY KEY, content integer UNIQUE)
+            """)
+        cur.execute("""
+            INSERT INTO ExpiredLibraryFileContent (content)
+            SELECT id FROM LibraryFileContent WHERE deleted IS FALSE
+            EXCEPT ALL
+            SELECT DISTINCT content
+            FROM LibraryFileAlias
+            WHERE expires IS NULL
+                OR expires >= CURRENT_TIMESTAMP AT TIME ZONE 'UTC'
+            """)
+        cur.execute(
+            "SELECT COALESCE(max(id),0) FROM ExpiredLibraryFileContent")
+        self.max_id = cur.fetchone()[0]
+        log.debug(
+            "%d expired LibraryFileContent to flag for removal."
+            % self.max_id)
+
+    def isDone(self):
+        if self.index > self.max_id:
+            log.info(
+                "Flagged %d expired files for removal."
+                % self.total_flagged)
+            return True
+        else:
+            return False
+
+    def __call__(self, chunksize):
+        chunksize = int(chunksize)
+        cur = self.con.cursor()
         cur.execute("""
             UPDATE LibraryFileContent SET deleted=TRUE
-            WHERE id = %d
-            """ % content_id)
-        if commit_counter % 100 == 0:
-            connection.commit()
-    connection.commit()
-    log.info("Flagged %d expired files for removal." % len(expired_ids))
+            FROM (
+                SELECT content FROM ExpiredLibraryFileContent
+                WHERE id BETWEEN %s AND %s
+                ) AS ExpiredLibraryFileContent
+            WHERE LibraryFileContent.id = ExpiredLibraryFileContent.content
+            """, (self.index, self.index + chunksize - 1))
+        flagged_rows = cur.rowcount
+        log.debug(
+            "Flagged %d expired LibraryFileContent for removal."
+            % flagged_rows)
+        self.total_flagged += flagged_rows
+        self.index += chunksize
+        self.con.commit()
+
+
+def flag_expired_files(connection):
+    """Invoke FlagExpiredFiles."""
+    loop_tuner = DBLoopTuner(FlagExpiredFiles(connection), 5, log=log)
+    loop_tuner.run()
 
 
 def delete_unwanted_files(con):
@@ -380,38 +489,72 @@
     cur.execute("SELECT max(id) from LibraryFileContent")
     max_id = cur.fetchone()[0]
 
-    # Build a set containing all stored LibraryFileContent ids
-    # that we want to keep.
+    # Calculate all stored LibraryFileContent ids that we want to keep.
+    # Results are ordered so we don't have to suck them all in at once.
     cur.execute("""
         SELECT id FROM LibraryFileContent
         WHERE deleted IS FALSE OR datecreated
             > CURRENT_TIMESTAMP AT TIME ZONE 'UTC' - '1 day'::interval
+        ORDER BY id
         """)
-    all_ids = set(row[0] for row in cur.fetchall())
+
+    def get_next_wanted_content_id():
+        result = cur.fetchone()
+        if result is None:
+            return None
+        else:
+            return result[0]
 
     count = 0
+    next_wanted_content_id = get_next_wanted_content_id()
+
     for content_id in range(1, max_id+1):
-        if content_id in all_ids:
-            continue # Linked in the db - do nothing
+        while (next_wanted_content_id is not None
+                and content_id > next_wanted_content_id):
+            next_wanted_content_id = get_next_wanted_content_id()
+
+        file_wanted = (
+                next_wanted_content_id is not None
+                and next_wanted_content_id == content_id)
+
         path = get_file_path(content_id)
 
-        if not os.path.exists(path):
-            continue # Exists neither on disk nor in the database - do nothing
-
-        one_day = 24 * 60 * 60
-        if time() - os.path.getctime(path) < one_day:
-            log.debug(
-                "File %d not removed - created too recently" % content_id)
-            continue # File created too recently - do nothing
-
-        # File uploaded a while ago but not in the database - remove it
-        log.debug("Deleting %s" % path)
-        os.remove(path)
-        count += 1
+        if file_wanted:
+            if (config.librarian_server.upstream_host is None
+                and not os.path.exists(path)):
+                # It is normal to have files in the database that are
+                # not on disk if the Librarian has an upstream
+                # Librarian, such as on staging. Don't spam in this
+                # case.
+                log.error(
+                    "LibraryFileContent %d exists in the db but not at %s"
+                    % (content_id, path))
+
+        else:
+            try:
+                one_day = 24 * 60 * 60
+                if time() - os.path.getctime(path) < one_day:
+                    log.debug(
+                        "File %d not removed - created too recently"
+                        % content_id)
+                else:
+                    # File uploaded a while ago but no longer wanted.
+                    os.unlink(path)
+                    log.debug("Deleted %s" % path)
+                    count += 1
+            except OSError, e:
+                if e.errno != errno.ENOENT:
+                    raise
+                if config.librarian_server.upstream_host is None:
+                    # It is normal to have files in the database that
+                    # are not on disk if the Librarian has an upstream
+                    # Librarian, such as on staging. Don't annoy the
+                    # operator with noise in this case.
+                    log.info("%s already deleted", path)
 
     log.info(
-            "Removed %d from disk that where no longer referenced in the db"
-            % count
+            "Deleted %d files from disk that where no longer referenced "
+            "in the db" % count
             )