Merge pull request #1 from orivej/parallel-update

Parallel fetching

misc.lisp

@@ -162,31 +162,41 @@
     (intern (string-upcase project))))
 
 
-(defun call-with-skipping (fun &key (stream *standard-output*))
-  (handler-bind ((error (lambda (condition)
-                          (when (find-restart 'skip)
-                            (when (boundp '*current-mapped-source*)
-                              (format stream "~&* ~A~%" *current-mapped-source*)
-                              (format stream ":: from ~A~%"
-                                      (find-source *current-mapped-source*)))
-                            (format stream "~&SKIPPING (~A)~%" condition)
-                            (invoke-restart 'skip)))))
-    (funcall fun)))
-
-(defun update-what-you-can (&optional file)
+(defvar *output-lock* (bt:make-lock "output-lock"))
+
+(defun call-with-skipping (fun &key (stream *standard-output*) parallel)
+  (flet ((invoke-skip (condition)
+           (when (find-restart 'skip)
+             (bt:with-lock-held (*output-lock*)
+               (when (boundp '*current-mapped-source*)
+                 (format stream "~&* ~A~%" *current-mapped-source*)
+                 (format stream ":: from ~A~%"
+                         (find-source *current-mapped-source*)))
+               (format stream "~&SKIPPING (~A)~%" condition))
+             (invoke-restart 'skip))))
+    (if (not parallel)
+        (handler-bind ((error #'invoke-skip))
+          (funcall fun))
+        (lparallel:task-handler-bind ((error #'invoke-skip))
+          (funcall fun)))))
+
+(defun update-what-you-can (&optional file parallel)
   (flet ((action (stream)
            (call-with-skipping
             (lambda ()
-              (map-sources (lambda (source)
-                             (force-output stream)
-                             (format t "~&Updating ~S from ~A~%"
-                                     (project-name source)
-                                     (location source))
-                             (update-source-cache source))))
-            :stream stream)))
+              (funcall (if parallel 'pmap-sources 'map-sources)
+                       (lambda (source)
+                         (bt:with-lock-held (*output-lock*)
+                           (force-output stream)
+                           (format t "~&Updating ~S from ~A~%"
+                                   (project-name source)
+                                   (location source)))
+                         (update-source-cache source))))
+            :stream stream
+            :parallel parallel)))
     (if file
         (with-open-file (stream file :direction :output
-                                :if-exists :rename-and-delete)
+                                     :if-exists :rename-and-delete)
           (action (make-broadcast-stream *standard-output* stream)))
         (action *standard-output*))))
 

quicklisp-controller.asd

@@ -10,7 +10,8 @@
                #:cl-ppcre
                #:alexandria
                #:drakma
-               #:ironclad)
+               #:ironclad
+               #:lparallel)
   :serial t
   :components ((:file "tarhash")
                (:file "package")

upstream-bzr.lisp

@@ -9,10 +9,15 @@
    :checkout-subcommand "branch"
    :update-subcommand "merge"))
 
+(defmethod source-host :around ((source bzr-source))
+  (if (string= "lp:" (location source) :end2 3)
+      "launchpad.net"
+      (call-next-method)))
+
 (defmethod make-release-tarball ((source bzr-source) output-file)
   (let* ((prefix (release-tarball-prefix source))
          (tar-name (string-right-trim "/" prefix))
-        (checkout (ensure-source-cache source)))
+         (checkout (ensure-source-cache source)))
     (in-temporary-directory prefix
       (let ((tempgz (make-pathname :name tar-name :type "tgz")))
         (with-posix-cwd checkout

upstream-cvs.lisp

@@ -18,6 +18,12 @@
 (defmethod source-location-initargs ((source cvs-oddmodule-source))
   (list :location :module-name))
 
+(defmethod source-host ((source cvs-source))
+  (let* ((location (location source))
+         (host-start (1+ (position #\@ location)))
+         (host-end (position #\: location :start host-start)))
+    (subseq location host-start host-end)))
+
 (defmethod source-description ((source cvs-source))
   (format nil "cvs -d ~A co ~A"
           (location source)

upstream.lisp

@@ -83,6 +83,10 @@
   (:method (source)
     (list :location)))
 
+(defgeneric source-host (source)
+  (:method (source)
+    (puri:uri-host (puri:parse-uri (location source)))))
+
 (defgeneric parse-location (source location-string)
   (:documentation "Update an instance by parsing its location value.")
   (:method (source location-string)
@@ -117,14 +121,35 @@
 
 (defvar *current-mapped-source* nil)
 
+(defun map-source (fun source)
+  (let ((*current-mapped-source* (project-name source)))
+    (with-simple-restart (skip "Skip ~A source" *current-mapped-source*)
+      (funcall fun source))))
+
 (defun map-sources (fun)
   (with-simple-restart (abort "Give up entirely")
     (dolist (source-file
               (directory #p"quicklisp-controller:projects;**;source.txt"))
-      (let* ((project-name (pathname-project-name source-file))
-             (*current-mapped-source* project-name))
-        (with-simple-restart (skip "Skip ~A source" project-name)
-          (funcall fun (load-source-file project-name source-file)))))))
+      (let ((project-name (pathname-project-name source-file)))
+        (map-source fun (load-source-file project-name source-file))))))
+
+(defun pmap-sources (fun)
+  (let ((dependency-tree (lparallel:make-ptree))
+        (host-dependency (make-hash-table :test 'equal))
+        (i 0))
+    (map-sources (lambda (source)
+                   (let ((host (source-host source)))
+                     (lparallel:ptree-fn i (gethash host host-dependency)
+                                         (lambda (&optional arg)
+                                           (declare (ignore arg))
+                                           (map-source fun source))
+                                         dependency-tree)
+                     (setf (gethash host host-dependency) (list i))
+                     (incf i))))
+    (lparallel:ptree-fn 'everything (loop for j below i collect j)
+                        (constantly nil) dependency-tree)
+    (lparallel:call-ptree 'everything dependency-tree)
+    nil))
 
 (defun find-source (name)
   (block nil