[master][PATCH] Threaded Koji RPM lookups and downloads

Martin Kolman mkolman at redhat.com
Wed Jul 31 19:46:50 UTC 2013


As the Koji RPM search takes quite some time, it is now threaded,
one thread per package. The thread handles both the lookup and package
download.
This should provide the biggest possible speedup and basically
cuts the whole operation down to the duration of the longest of the
Koji lookups.
As urlgrabber turned out to be thread-unsafe, it has been replaced by
a urllib.urlretrieve() call and the live download progress has been
replaced by simple per-file/thread "download done" notification

Signed-off-by: Martin Kolman <mkolman at redhat.com>
---
 scripts/makeupdates | 165 ++++++++++++++++++++++++++++++++++------------------
 1 file changed, 108 insertions(+), 57 deletions(-)

diff --git a/scripts/makeupdates b/scripts/makeupdates
index 6368ec2..0b0e0b2 100755
--- a/scripts/makeupdates
+++ b/scripts/makeupdates
@@ -28,8 +28,8 @@ import subprocess
 import sys
 import re
 import glob
-import urlgrabber
-import urlgrabber.progress
+import urllib
+import threading
 
 RPM_FOLDER_NAME = "~/.anaconda_updates_rpm_cache"
 KOJI_BASE_URL = "http://kojipkgs.fedoraproject.org//packages/" \
@@ -112,15 +112,19 @@ def doGitContentDiff(tag, args=[]):
     lines = proc[0].split('\n')
     return lines
 
-def download_with_progress(url, path):
+def download_to_file(url, path):
+    """Download a file to the given path,
+    return the storage path if successful,
+    or None if the download fails for some reason
+    """
     try:
-        prog = urlgrabber.progress.text_progress_meter()
-        urlgrabber.urlgrab(url, filename=path, progress_obj=prog)
-        return True
-    except Exception:
+        result = urllib.urlretrieve(url, path)
+        # return the storage path
+        return result[0]
+    except Exception, e:
+        print("download of %s to %s failed with exception: %s" % (url, path, e))
         return None
 
-
 def create_RPM_cache_folder():
     """Create RPM package cache folder if it does not already exist"""
     if not os.path.exists(RPM_FOLDER_NAME):
@@ -376,15 +380,10 @@ def check_for_new_packages(tag, arch):
     fedora_number = get_fedora_version()
     # remove available packages from the list
     new_packages, include_rpms = remove_local_packages(new_packages, arch)
-    # generate URLs for packages that are not locally available
-    urls = get_koji_URLS(new_packages, fedora_number, arch)
-    # download the Koji URLs
-    for url, filename in urls:
-        # append the RPM cache folder to the filename
-        download_path = os.path.join(RPM_FOLDER_NAME, filename)
-        if download_with_progress(url, download_path):
-            # add successfully downloads to the RPM inclusion list
-            include_rpms.append(download_path)
+    # if some packages are not locally available, download them from Koji
+    if new_packages:
+        include_rpms.extend(get_RPMs_from_koji(new_packages, fedora_number, arch))
+    # return absolute paths for the packages
     return map(lambda x: os.path.abspath(x), include_rpms)
 
 def remove_local_packages(packages, arch):
@@ -421,7 +420,9 @@ def remove_local_packages(packages, arch):
                 del packages[name]
 
     # return only those packages that are not locally available
-    if include_rpms:
+    if include_rpms and not packages:
+        print("all %d RPMs found locally:" % len(include_rpms))
+    elif include_rpms:
         print("%d RPMs found locally:" % len(include_rpms))
     else:
         print("no packages found locally")
@@ -429,50 +430,100 @@ def remove_local_packages(packages, arch):
         print(os.path.basename(rpm))
     return packages, include_rpms
 
-def get_koji_URLS(packages, fedora_number, arch):
+def get_RPMs_from_koji(packages, fedora_number, arch):
     """Get RPM download URLs for given packages and Fedora version,
     return URLS and RPM filenames
     """
-    url_filename = []
+    threads = []
+    rpm_paths = []
+    # the print lock is used to make sure only one
+    # thread is printing to stdout at a time
+    print_lock = threading.Lock()
+
+    index = 1
+    print("starting %d worker threads" % len(packages))
+    for package, version in packages.iteritems():
+        package_version = (package, version)
+        thread = threading.Thread(name=index, target=get_rpm_from_Koji_thread,
+                                  args=(package_version, fedora_number,
+                                        arch, rpm_paths, print_lock))
+        thread.start()
+        threads.append(thread)
+        index+=1
+    # wait for all threads to finish
+    for thread in threads:
+        thread.join()
+
+    print("%d RPMs have been downloaded" % len(rpm_paths))
+
+    # return the list of paths for the downloaded RPMs
+    return rpm_paths
+
+def get_rpm_from_Koji_thread(package, fedora_number, arch,
+                             rpm_paths, print_lock):
+    """Download the given package from Koji and if successful,
+    append the path to the downloaded file to the rpm_paths list
+    """
+    # just to be sure, create a separate session for each query,
+    # as the individual lookups will run in different threads
     import koji
     kojiclient = koji.ClientSession('http://koji.fedoraproject.org/kojihub', {})
-    for name, version in packages.iteritems():
-        if not version:
-            version = "*"
-        # check if version contains build number or not
-        if len(version.split("-")) == 1:
-            version = "%s-*" % version
-        package_glob = "%s-%s.fc%d.*.rpm" % (name, version, fedora_number)
-        print("searching for %s in Koji" % package_glob)
-        results = kojiclient.search(package_glob, "rpm", "glob")
-        # leave only results that are either noarch or are built
-        # for the current architecture
-        allowed = ("noarch.rpm", "%s.rpm" % arch)
-        results = [x for x in results if x['name'].endswith(allowed)]
-        if results:  # any packages left ?
-            package_metadata = {}
-            rpm_name = results[0]['name']
-            package_metadata['rpm_name'] = rpm_name
-            print("RPM found: %s" % rpm_name)
-            rpm_id = results[0]['id']
-
-            # get info about the RPM to
-            # get the arch and build_id
-            result = kojiclient.getRPM(rpm_id)
-            package_metadata['arch'] = result['arch']
-            package_metadata['release'] = result['release']
-            build_id = result['build_id']
-
-            # so we can get the toplevel package name and version
-            result = kojiclient.getBuild(build_id)
-            package_metadata['toplevel_name'] = result['package_name']
-            package_metadata['toplevel_version'] = result['version']
-
-            # and use the information to build the URL
-            url = KOJI_BASE_URL % package_metadata
-            # simple, isn't it ? :)
-            url_filename.append((url, rpm_name))
-    return url_filename
+    name, version = package
+    if not version:
+        version = "*"
+    # check if version contains build number or not
+    if len(version.split("-")) == 1:
+        version = "%s-*" % version
+    package_glob = "%s-%s.fc%d.*.rpm" % (name, version, fedora_number)
+
+    # get the current thread, so output can be prefixed by thread number
+    prefix = "thread %s:" % threading.current_thread().name
+    with print_lock:
+        print("%s searching for: %s in Koji" % (prefix, package_glob))
+    # call the Koji API
+    results = kojiclient.search(package_glob, "rpm", "glob")
+    # leave only results that are either noarch
+    # or are built for the current architecture
+    allowed = ("noarch.rpm", "%s.rpm" % arch)
+    results = [x for x in results if x['name'].endswith(allowed)]
+    if results:  # any packages left ?
+        package_metadata = {}
+        rpm_name = results[0]['name']
+        package_metadata['rpm_name'] = rpm_name
+        with print_lock:
+            print("%s RPM found: %s" % (prefix, rpm_name))
+        rpm_id = results[0]['id']
+
+        # get info about the RPM to
+        # get the arch and build_id
+        result = kojiclient.getRPM(rpm_id)
+        package_metadata['arch'] = result['arch']
+        package_metadata['release'] = result['release']
+        build_id = result['build_id']
+
+        # so we can get the toplevel package name and version
+        result = kojiclient.getBuild(build_id)
+        package_metadata['toplevel_name'] = result['package_name']
+        package_metadata['toplevel_version'] = result['version']
+
+        # and use the information to build the URL
+        url = KOJI_BASE_URL % package_metadata
+        # simple, isn't it ? :)
+
+        # append the RPM cache folder to the filename
+        download_path = os.path.join(RPM_FOLDER_NAME, rpm_name)
+        # check if the download was successful
+        storage_path = download_to_file(url, download_path)
+        if storage_path is not None:
+            with print_lock:
+                print("%s download done: %s" % (prefix, rpm_name))
+            # add successful downloads to the RPM inclusion list
+            rpm_paths.append(storage_path)
+            # GIL should be enough for appending to the list
+            # from multiple threads
+        else:
+            with print_lock:
+                print("%s download failed: %s @ %s" % (prefix, rpm_name, url))
 
 def usage(cmd):
     sys.stdout.write("Usage: %s [OPTION]...\n" % (cmd,))
-- 
1.8.3.1



More information about the anaconda-patches mailing list