Merge remote branch 'origin/master' into p-s-from-db

diff --git a/dak/generate_filelist.py b/dak/generate_filelist.py
index 2a6d218badf2e6e12a6d2b9a43151a4400d4906c..2d2e9499e932581acb2db1c95e76d741b10c4343 100755 (executable)
--- a/dak/generate_filelist.py
+++ b/dak/generate_filelist.py
@@ -39,7 +39,7 @@ Generate file lists for apt-ftparchive.
 from daklib.dbconn import *
 from daklib.config import Config
 from daklib import utils, daklog
-from multiprocessing import Pool
+from daklib.dakmultiprocessing import Pool
 import apt_pkg, os, stat, sys
 
 from daklib.lists import getSources, getBinaries, getArchAll
@@ -72,11 +72,13 @@ def writeSourceList(suite_id, component_id, incremental_mode):
     (file, timestamp) = listPath(suite, component,
             incremental_mode = incremental_mode)
 
+    message = "sources list for %s %s" % (suite.suite_name, component.component_name)
+
     for _, filename in getSources(suite, component, session, timestamp):
         file.write(filename + '\n')
-    session.close()
+    session.rollback()
     file.close()
-    return "sources list for %s %s" % (suite.suite_name, component.component_name)
+    return message
 
 def writeAllList(suite_id, component_id, architecture_id, type, incremental_mode):
     session = DBConn().session()
@@ -86,12 +88,14 @@ def writeAllList(suite_id, component_id, architecture_id, type, incremental_mode
     (file, timestamp) = listPath(suite, component, architecture, type,
             incremental_mode)
 
+    message = "all list for %s %s (arch=%s, type=%s)" % (suite.suite_name, component.component_name, architecture.arch_string, type)
+
     for _, filename in getArchAll(suite, component, architecture, type,
             session, timestamp):
         file.write(filename + '\n')
-    session.close()
+    session.rollback()
     file.close()
-    return "all list for %s %s (arch=%s, type=%s)" % (suite.suite_name, component.component_name, architecture.arch_string, type)
+    return message
 
 def writeBinaryList(suite_id, component_id, architecture_id, type, incremental_mode):
     session = DBConn().session()
@@ -101,12 +105,14 @@ def writeBinaryList(suite_id, component_id, architecture_id, type, incremental_m
     (file, timestamp) = listPath(suite, component, architecture, type,
             incremental_mode)
 
+    message = "binary list for %s %s (arch=%s, type=%s)" % (suite.suite_name, component.component_name, architecture.arch_string, type)
+
     for _, filename in getBinaries(suite, component, architecture, type,
             session, timestamp):
         file.write(filename + '\n')
-    session.close()
+    session.rollback()
     file.close()
-    return "binary list for %s %s (arch=%s, type=%s)" % (suite.suite_name, component.component_name, architecture.arch_string, type)
+    return message
 
 def usage():
     print """Usage: dak generate_filelist [OPTIONS]

diff --git a/dak/generate_packages_sources2.py b/dak/generate_packages_sources2.py
index 8d8a8aefa3f2d51103a6324480726c3101d2e6ee..ea3c0e53b21866ab40dcff27f75171121d6fb341 100755 (executable)
--- a/dak/generate_packages_sources2.py
+++ b/dak/generate_packages_sources2.py
@@ -31,7 +31,9 @@ Generate Packages/Sources files
 from daklib.dbconn import *
 from daklib.config import Config
 from daklib import utils, daklog
-from multiprocessing import Pool
+from daklib.dakmultiprocessing import Pool
+from daklib.filewriter import PackagesFileWriter, SourcesFileWriter
+
 import apt_pkg, os, stat, sys
 
 def usage():
@@ -93,23 +95,6 @@ ORDER BY
 s.source, s.version
 """
 
-def open_sources(suite, component):
-    cnf = Config()
-    dest = os.path.join(cnf['Dir::Root'], 'dists', suite.suite_name, component.component_name, 'source', 'Sources')
-
-    # create queue if it does not exist yet
-    if os.path.exists(dest) and os.path.isdir(dest):
-        dest_dir = dest
-    else:
-        dest_dir = os.path.dirname(dest)
-    if not os.path.exists(dest_dir):
-        umask = os.umask(00000)
-        os.makedirs(dest_dir, 02775)
-        os.umask(umask)
-
-    f = open(dest, 'w')
-    return f
-
 def generate_sources(suite_id, component_id):
     global _sources_query
 
@@ -119,7 +104,8 @@ def generate_sources(suite_id, component_id):
     suite = session.query(Suite).get(suite_id)
     component = session.query(Component).get(component_id)
 
-    output = open_sources(suite, component)
+    writer = SourcesFileWriter(suite=suite.suite_name, component=component.component_name)
+    output = writer.open()
 
     # run query and write Sources
     r = session.execute(_sources_query, {"suite": suite_id, "component": component_id, "dsc_type": dsc_type})
@@ -127,7 +113,11 @@ def generate_sources(suite_id, component_id):
         print >>output, stanza
         print >>output, ""
 
-    return ["generate sources", suite.suite_name, component.component_name]
+    writer.close()
+
+    message = ["generate sources", suite.suite_name, component.component_name]
+    session.rollback()
+    return message
 
 #############################################################################
 
@@ -198,26 +188,6 @@ WHERE
 ORDER BY tmp.package, tmp.version
 """
 
-def open_packages(suite, component, architecture, type_name):
-    cnf = Config()
-    if type_name == 'udeb':
-        dest = os.path.join(cnf['Dir::Root'], 'dists', suite.suite_name, component.component_name, 'debian-installer', 'binary-%s' % architecture.arch_string, 'Packages')
-    else:
-        dest = os.path.join(cnf['Dir::Root'], 'dists', suite.suite_name, component.component_name, 'binary-%s' % architecture.arch_string, 'Packages')
-
-    # create queue if it does not exist yet
-    if os.path.exists(dest) and os.path.isdir(dest):
-        dest_dir = dest
-    else:
-        dest_dir = os.path.dirname(dest)
-    if not os.path.exists(dest_dir):
-        umask = os.umask(00000)
-        os.makedirs(dest_dir, 02775)
-        os.umask(umask)
-
-    f = open(dest, 'w')
-    return f
-
 def generate_packages(suite_id, component_id, architecture_id, type_name):
     global _packages_query
 
@@ -229,7 +199,9 @@ def generate_packages(suite_id, component_id, architecture_id, type_name):
     component = session.query(Component).get(component_id)
     architecture = session.query(Architecture).get(architecture_id)
 
-    output = open_packages(suite, component, architecture, type_name)
+    writer = PackagesFileWriter(suite=suite.suite_name, component=component.component_name,
+            architecture=architecture.arch_string, debtype=type_name)
+    output = writer.open()
 
     r = session.execute(_packages_query, {"suite": suite_id, "component": component_id,
         "arch": architecture_id, "type_id": type_id, "type_name": type_name, "arch_all": arch_all_id})
@@ -237,9 +209,11 @@ def generate_packages(suite_id, component_id, architecture_id, type_name):
         print >>output, stanza
         print >>output, ""
 
-    session.close()
+    writer.close()
 
-    return ["generate-packages", suite.suite_name, component.component_name, architecture.arch_string]
+    message = ["generate-packages", suite.suite_name, component.component_name, architecture.arch_string]
+    session.rollback()
+    return message
 
 #############################################################################
 
@@ -289,9 +263,8 @@ def main():
         for c in component_ids:
             pool.apply_async(generate_sources, [s.suite_id, c], callback=log)
             for a in s.architectures:
-                #pool.apply_async(generate_packages, [s.suite_id, c, a.arch_id, 'deb'], callback=log)
-                apply(generate_packages, [s.suite_id, c, a.arch_id, 'deb'])
-                #pool.apply_async(generate_packages, [s.suite_id, c, a.arch_id, 'udeb'], callback=log)
+                pool.apply_async(generate_packages, [s.suite_id, c, a.arch_id, 'deb'], callback=log)
+                pool.apply_async(generate_packages, [s.suite_id, c, a.arch_id, 'udeb'], callback=log)
 
     pool.close()
     pool.join()

diff --git a/dak/show_new.py b/dak/show_new.py
index 8405aeef9f0fe2391d9e969169b944db7abf179b..84b45077b1d11733b55cfd89cdc789e727092cac 100755 (executable)
--- a/dak/show_new.py
+++ b/dak/show_new.py
@@ -37,7 +37,7 @@ from daklib.regexes import re_source_ext
 from daklib.config import Config
 from daklib import daklog
 from daklib.changesutils import *
-from daklib.threadpool import ThreadPool
+from daklib.dakmultiprocessing import Pool
 
 # Globals
 Cnf = None
@@ -250,14 +250,15 @@ def main():
 
     examine_package.use_html=1
 
-    threadpool = ThreadPool()
+    pool = Pool()
     for changes_file in changes_files:
         changes_file = utils.validate_changes_file_arg(changes_file, 0)
         if not changes_file:
             continue
         print "\n" + changes_file
-        threadpool.queueTask(do_pkg, changes_file)
-    threadpool.joinAll()
+        pool.apply_async(do_pkg, (changes_file,))
+    pool.close()
+    pool.join()
 
     files = set(os.listdir(cnf["Show-New::HTMLPath"]))
     to_delete = filter(lambda x: x.endswith(".html"), files.difference(sources))

diff --git a/daklib/dakmultiprocessing.py b/daklib/dakmultiprocessing.py
new file mode 100644 (file)
index 0000000..ded81a2
--- /dev/null
+++ b/daklib/dakmultiprocessing.py
@@ -0,0 +1,56 @@
+#!/usr/bin/env python
+# vim:set et sw=4:
+
+"""
+multiprocessing for DAK
+
+@contact: Debian FTP Master <ftpmaster@debian.org>
+@copyright: 2011  Ansgar Burchardt <ansgar@debian.org>
+@license: GNU General Public License version 2 or later
+"""
+
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+
+###############################################################################
+
+import multiprocessing
+import sqlalchemy.orm.session
+
+def _func_wrapper(func, *args, **kwds):
+    try:
+        return func(*args, **kwds)
+    finally:
+        # Make sure connections are closed. We might die otherwise.
+        sqlalchemy.orm.session.Session.close_all()
+
+class Pool():
+    def __init__(self, *args, **kwds):
+        self.pool = multiprocessing.Pool(*args, **kwds)
+        self.results = []
+
+    def apply_async(self, func, args=(), kwds={}, callback=None):
+        wrapper_args = list(args)
+        wrapper_args.insert(0, func)
+        self.results.append(self.pool.apply_async(_func_wrapper, wrapper_args, kwds, callback))
+
+    def close(self):
+        self.pool.close()
+
+    def join(self):
+        self.pool.join()
+        for r in self.results:
+            # return values were already handled in the callbacks, but asking
+            # for them might raise exceptions which would otherwise be lost
+            r.get()