X-Git-Url: https://git.decadent.org.uk/gitweb/?a=blobdiff_plain;f=dak%2Fimport_known_changes.py;h=c8d5bf96c921a575bc738b9fec24fc6ee08447f3;hb=63cfa70f44ac9b08b214b30d80cd39b0295a1cd3;hp=3702411d3cbe9908ad034c8fbae3493177241099;hpb=6021a91078574c97968d076c0320182a20c33649;p=dak.git

diff --git a/dak/import_known_changes.py b/dak/import_known_changes.py
index 3702411d..c8d5bf96 100755
--- a/dak/import_known_changes.py
+++ b/dak/import_known_changes.py
@@ -32,7 +32,7 @@ import sys
 import os
 import logging
 import threading
-from daklib.dbconn import DBConn,get_knownchange
+from daklib.dbconn import DBConn, get_dbchange
 from daklib.config import Config
 import apt_pkg
 from daklib.dak_exceptions import DBUpdateError, InvalidDscError, ChangesUnicodeError
@@ -128,37 +128,48 @@ class OneAtATime(object):
     """
     def __init__(self):
         self.next_in_line = None
-        self.next_lock = threading.Condition()
+        self.read_lock = threading.Condition()
+        self.write_lock = threading.Condition()
         self.die = False
 
     def plsDie(self):
         self.die = True
-        self.next_lock.notify()
+        self.write_lock.acquire()
+        self.write_lock.notifyAll()
+        self.write_lock.release()
+
+        self.read_lock.acquire()
+        self.read_lock.notifyAll()
+        self.read_lock.release()
 
     def enqueue(self, next):
-        self.next_lock.acquire()
+        self.write_lock.acquire()
         while self.next_in_line:
             if self.die:
                 return
-            self.next_lock.wait()
+            self.write_lock.wait()
 
         assert( not self.next_in_line )
         self.next_in_line = next
-        self.next_lock.notify()
-        self.next_lock.release()
+        self.write_lock.release()
+        self.read_lock.acquire()
+        self.read_lock.notify()
+        self.read_lock.release()
 
     def dequeue(self):
-        self.next_lock.acquire()
+        self.read_lock.acquire()
         while not self.next_in_line:
             if self.die:
                 return
-            self.next_lock.wait()
+            self.read_lock.wait()
 
         result = self.next_in_line
 
         self.next_in_line = None
-        self.next_lock.notify()
-        self.next_lock.release()
+        self.read_lock.release()
+        self.write_lock.acquire()
+        self.write_lock.notify()
+        self.write_lock.release()
 
         if isinstance(result, EndOfChanges):
             return None
@@ -195,7 +206,7 @@ class ChangesGenerator(threading.Thread):
             if os.path.exists(checkdir):
                 print "Looking into %s" % (checkdir)
 
-                for dirpath, dirnames, filenames in os.walk(checkdir, topdown=False):
+                for dirpath, dirnames, filenames in os.walk(checkdir, topdown=True):
                     if not filenames:
                         # Empty directory (or only subdirectories), next
                         continue
@@ -207,7 +218,7 @@ class ChangesGenerator(threading.Thread):
                                 continue
                             count += 1
 
-                            if not get_knownchange(changesfile, self.session):
+                            if not get_dbchange(changesfile, self.session):
                                 to_import = ChangesToImport(dirpath, changesfile, count)
                                 if self.die:
                                     return
@@ -255,13 +266,12 @@ class ImportThread(threading.Thread):
             except ChangesUnicodeError:
                 warn("found invalid changes file, not properly utf-8 encoded")
 
-
             except KeyboardInterrupt:
                 print("Caught C-c; on ImportThread. terminating.")
                 self.parent.plsDie()
                 sys.exit(1)
+
             except:
-                traceback.print_exc()
                 self.parent.plsDie()
                 sys.exit(1)