From: Ansgar Burchardt <ansgar@debian.org>
Date: Sat, 26 Mar 2011 17:09:11 +0000 (+0000)
Subject: dakmultiprocessing: close all session
X-Git-Url: https://err.no/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=456b7dde5f862666d82f4e8f48234f08a941fb9a;p=dak

dakmultiprocessing: close all session

Make sure to close all sessions so we do not run out of connections.
Before this exceptions in the workers might leave open transactions
around.

Signed-off-by: Ansgar Burchardt <ansgar@debian.org>
---

diff --git a/daklib/dakmultiprocessing.py b/daklib/dakmultiprocessing.py
index 646f3633..ded81a29 100644
--- a/daklib/dakmultiprocessing.py
+++ b/daklib/dakmultiprocessing.py
@@ -26,6 +26,14 @@ multiprocessing for DAK
 ###############################################################################
 
 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):
@@ -33,7 +41,9 @@ class Pool():
         self.results = []
 
     def apply_async(self, func, args=(), kwds={}, callback=None):
-        self.results.append(self.pool.apply_async(func, args, kwds, callback))
+        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()