cbaines/guix-data-service
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add meaningful parallelism to processing jobs

Browse source 
Make parallel use of inferiors when computing channel instance derivations,
and when extracting information about a revision. This should allow for some
horizontal scalability, reducing the impact of additional systems for which
derivations need computing.

This commit also fixes an apparent issue with package replacements, as
previously the wrong id was used, and this hid some issues around
deduplication.
This commit is contained in:
Christopher Baines 2024-01-18 14:44:04 +00:00
parent 6842a432d6
commit c1d2f3a1b7
4 changed files with 482 additions and 314 deletions
Download patch file Download diff file Expand all files Collapse all files

16
guix-data-service/jobs.scm
View file

@ -125,7 +125,8 @@ guix-data-service: error: missing log line: ~A
(define* (process-jobs conn #:key max-processes (define* (process-jobs conn #:key max-processes
latest-branch-revision-max-processes latest-branch-revision-max-processes
skip-system-tests?) skip-system-tests?
per-job-parallelism)
(define (fetch-new-jobs) (define (fetch-new-jobs)
(fetch-unlocked-jobs conn)) (fetch-unlocked-jobs conn))
@ -133,11 +134,14 @@ guix-data-service: error: missing log line: ~A
(let ((log-port (start-thread-for-process-output job-id))) (let ((log-port (start-thread-for-process-output job-id)))
(spawn (spawn
"guix-data-service-process-job" "guix-data-service-process-job"
(cons* "guix-data-service-process-job" `("guix-data-service-process-job"
job-id ,job-id
(if skip-system-tests? ,@(if skip-system-tests?
'("--skip-system-tests") '("--skip-system-tests")
'())) '())
,@(if per-job-parallelism
(list (simple-format #f "--parallelism=~A" per-job-parallelism))
'()))
#:output log-port #:output log-port
#:error log-port))) #:error log-port)))

761
guix-data-service/jobs/load-new-guix-revision.scm
View file

@ -24,10 +24,13 @@
#:use-module (ice-9 threads) #:use-module (ice-9 threads)
#:use-module (ice-9 textual-ports) #:use-module (ice-9 textual-ports)
#:use-module (ice-9 hash-table) #:use-module (ice-9 hash-table)
#:use-module (ice-9 suspendable-ports)
#:use-module ((ice-9 ports internal) #:select (port-poll))
#:use-module (rnrs exceptions) #:use-module (rnrs exceptions)
#:use-module (json) #:use-module (json)
#:use-module (squee) #:use-module (squee)
#:use-module (fibers) #:use-module (fibers)
#:use-module (fibers channels)
#:use-module (guix monads) #:use-module (guix monads)
#:use-module (guix store) #:use-module (guix store)
#:use-module (guix channels) #:use-module (guix channels)
@ -185,7 +188,7 @@
(let ((system-test-data (let ((system-test-data
(with-time-logging "getting system tests" (with-time-logging "getting system tests"
(inferior-eval-with-store inf store extract)))) (inferior-eval-with-store/non-blocking inf store extract))))
(for-each (lambda (derivation-file-names-by-system) (for-each (lambda (derivation-file-names-by-system)
(for-each (lambda (derivation-file-name) (for-each (lambda (derivation-file-name)
@ -342,9 +345,11 @@
#:unwind? #t))) #:unwind? #t)))
gds-inferior-packages)))) gds-inferior-packages))))
(ensure-gds-inferior-packages-defined! inf)
(with-time-logging (simple-format #f "getting ~A lint warnings" (with-time-logging (simple-format #f "getting ~A lint warnings"
checker-name) checker-name)
(inferior-eval-with-store (inferior-eval-with-store/non-blocking
inf inf
store store
lint-warnings-for-checker))) lint-warnings-for-checker)))
@ -588,12 +593,13 @@
(with-time-logging (with-time-logging
(simple-format #f "getting derivations for ~A" (cons system target)) (simple-format #f "getting derivations for ~A" (cons system target))
(inferior-eval-with-store (inferior-eval-with-store/non-blocking
inf inf
store store
proc))) proc)))
(define (sort-and-deduplicate-inferior-packages packages) (define (sort-and-deduplicate-inferior-packages packages
pkg-to-replacement-hash-table)
(pair-fold (pair-fold
(lambda (pair result) (lambda (pair result)
(if (null? (cdr pair)) (if (null? (cdr pair))
@ -604,8 +610,8 @@
(b-name (inferior-package-name b)) (b-name (inferior-package-name b))
(a-version (inferior-package-version a)) (a-version (inferior-package-version a))
(b-version (inferior-package-version b)) (b-version (inferior-package-version b))
(a-replacement (inferior-package-replacement a)) (a-replacement (hashq-ref pkg-to-replacement-hash-table a))
(b-replacement (inferior-package-replacement b))) (b-replacement (hashq-ref pkg-to-replacement-hash-table b)))
(if (and (string=? a-name b-name) (if (and (string=? a-name b-name)
(string=? a-version b-version) (string=? a-version b-version)
(eq? a-replacement b-replacement)) (eq? a-replacement b-replacement))
@ -638,8 +644,24 @@
b-name))))))) b-name)))))))
(define (inferior-packages-plus-replacements inf) (define (inferior-packages-plus-replacements inf)
(let* ((packages (inferior-packages inf)) (let* ((packages
(replacements (filter-map inferior-package-replacement packages)) ;; The use of force in (guix inferior) introduces a continuation
;; barrier
(call-with-temporary-thread
(lambda ()
(inferior-packages inf))))
(replacements (map inferior-package-replacement packages))
(pkg-to-replacement-hash-table
(let ((ht (make-hash-table)))
(for-each
(lambda (pkg replacement)
(when replacement
(hashq-set! ht
pkg
replacement)))
packages
replacements)
ht))
(non-exported-replacements (non-exported-replacements
(let ((package-id-hash-table (make-hash-table))) (let ((package-id-hash-table (make-hash-table)))
(for-each (lambda (pkg) (for-each (lambda (pkg)
@ -648,27 +670,29 @@
#t)) #t))
packages) packages)
(filter (lambda (pkg) (filter
(eq? #f (lambda (pkg)
(hash-ref package-id-hash-table (and pkg
(inferior-package-id pkg)))) (eq? #f
replacements))) (hash-ref package-id-hash-table
(inferior-package-id pkg)))))
replacements)))
(deduplicated-packages (deduplicated-packages
;; This isn't perfect, sometimes there can be two packages with the ;; This isn't perfect, sometimes there can be two packages with the
;; same name and version, but different derivations. Guix will warn ;; same name and version, but different derivations. Guix will warn
;; about this case though, generally this means only one of the ;; about this case though, generally this means only one of the
;; packages should be exported. ;; packages should be exported.
(sort-and-deduplicate-inferior-packages (call-with-temporary-thread
(append! packages non-exported-replacements))) (lambda ()
;; TODO Sort introduces a continuation barrier
(sort-and-deduplicate-inferior-packages
(append! packages non-exported-replacements)
pkg-to-replacement-hash-table))))
(deduplicated-packages-length (deduplicated-packages-length
(length deduplicated-packages))) (length deduplicated-packages)))
(inferior-eval
`(use-modules (srfi srfi-43))
inf)
(inferior-eval (inferior-eval
`(define gds-inferior-packages `(define gds-inferior-packages
(make-vector ,deduplicated-packages-length)) (make-vector ,deduplicated-packages-length))
@ -685,9 +709,14 @@
(list ,@(map inferior-package-id deduplicated-packages))) (list ,@(map inferior-package-id deduplicated-packages)))
inf) inf)
(list->vector deduplicated-packages))) (values (list->vector deduplicated-packages)
pkg-to-replacement-hash-table)))
(define* (all-inferior-packages-data inf packages) (define (ensure-gds-inferior-packages-defined! inf)
(unless (inferior-eval '(defined? 'gds-inferior-packages) inf)
(inferior-packages-plus-replacements inf)))
(define* (all-inferior-packages-data inf packages pkg-to-replacement-hash-table)
(define inferior-package-id->packages-index-hash-table (define inferior-package-id->packages-index-hash-table
(let ((hash-table (make-hash-table))) (let ((hash-table (make-hash-table)))
(vector-for-each (vector-for-each
@ -717,7 +746,7 @@
(package-replacement-data (package-replacement-data
(vector-map (vector-map
(lambda (_ pkg) (lambda (_ pkg)
(let ((replacement (inferior-package-replacement pkg))) (let ((replacement (hashq-ref pkg-to-replacement-hash-table pkg)))
(if replacement (if replacement
;; I'm not sure if replacements can themselves be ;; I'm not sure if replacements can themselves be
;; replaced, but I do know for sure that there are ;; replaced, but I do know for sure that there are
@ -726,8 +755,16 @@
;; example). ;; example).
;; ;;
;; So this might be #f in these cases ;; So this might be #f in these cases
(hash-ref inferior-package-id->packages-index-hash-table (let ((index
(inferior-package-id pkg)) (hash-ref inferior-package-id->packages-index-hash-table
(inferior-package-id replacement))))
(unless index
(simple-format
(current-error-port)
"warning: replacement for ~A (~A) is unknown\n"
pkg
replacement))
index)
#f))) #f)))
packages))) packages)))
@ -743,13 +780,14 @@
(let* ((names (assq-ref inferior-packages-data 'names)) (let* ((names (assq-ref inferior-packages-data 'names))
(versions (assq-ref inferior-packages-data 'versions)) (versions (assq-ref inferior-packages-data 'versions))
(package-license-set-ids (package-license-set-ids
(inferior-packages->license-set-ids (with-time-logging "inserting package license sets"
conn (inferior-packages->license-set-ids
(inferior-packages->license-id-lists conn
conn (inferior-packages->license-id-lists
;; TODO Don't needlessly convert conn
(vector->list ;; TODO Don't needlessly convert
(assq-ref inferior-packages-data 'license-data))))) (vector->list
(assq-ref inferior-packages-data 'license-data))))))
(all-package-metadata-ids (all-package-metadata-ids
new-package-metadata-ids new-package-metadata-ids
(with-time-logging "inserting package metadata entries" (with-time-logging "inserting package metadata entries"
@ -898,15 +936,80 @@
(build-derivations store (list derivation))) (build-derivations store (list derivation)))
(derivation->output-path derivation))) (derivation->output-path derivation)))
(define (channel->source-and-derivation-file-names-by-system conn store channel (define (non-blocking-port port)
fetch-with-authentication?) "Make PORT non-blocking and return it."
(let ((flags (fcntl port F_GETFL)))
(when (zero? (logand O_NONBLOCK flags))
(fcntl port F_SETFL (logior O_NONBLOCK flags)))
port))
(define (ensure-non-blocking-store-connection store)
(match (store-connection-socket store)
((? file-port? port)
(non-blocking-port port))
(_ #f)))
(define (call-with-temporary-blocking-store store proc)
(let* ((port (store-connection-socket store))
(flags (fcntl port F_GETFL)))
(unless (zero? (logand O_NONBLOCK flags))
(fcntl port F_SETFL (logxor O_NONBLOCK flags)))
(call-with-values
(lambda ()
(proc store))
(lambda vals
(fcntl port F_SETFL (logior O_NONBLOCK flags))
(apply values vals)))))
(define (make-inferior-non-blocking! inferior)
(non-blocking-port
((@@ (guix inferior) inferior-socket) inferior)))
(define (call-with-temporary-thread thunk)
(let ((channel (make-channel)))
(call-with-new-thread
(lambda ()
(parameterize
((current-read-waiter (lambda (port) (port-poll port "r")))
(current-write-waiter (lambda (port) (port-poll port "w"))))
(with-exception-handler
(lambda (exn)
(put-message channel `(exception ,exn)))
(lambda ()
(with-throw-handler #t
(lambda ()
(call-with-values thunk
(lambda values
(put-message channel `(values ,@values)))))
(lambda _
(backtrace))))
#:unwind? #t))))
(match (get-message channel)
(('values . results)
(apply values results))
(('exception . args)
(apply throw args)))))
(define (inferior-eval-with-store/non-blocking inferior store proc)
(call-with-temporary-thread
(lambda ()
(inferior-eval-with-store inferior store proc))))
(define* (channel->source-and-derivation-file-names-by-system
conn store channel
fetch-with-authentication?
#:key parallelism)
(define use-container? (defined? (define use-container? (defined?
'open-inferior/container 'open-inferior/container
(resolve-module '(guix inferior)))) (resolve-module '(guix inferior))))
(define (inferior-code channel-instance systems) (define (inferior-code channel-instance system)
`(lambda (store) `(lambda (store)
(let* ((instances (let* ((system ,system)
(instances
(list (list
(channel-instance (channel-instance
(channel (name ',(channel-name channel)) (channel (name ',(channel-name channel))
@ -915,61 +1018,57 @@
(commit ,(channel-commit channel))) (commit ,(channel-commit channel)))
,(channel-instance-commit channel-instance) ,(channel-instance-commit channel-instance)
,(channel-instance-checkout channel-instance))))) ,(channel-instance-checkout channel-instance)))))
(map (simple-format
(lambda (system) (current-error-port)
(simple-format "guix-data-service: computing the derivation-file-name for ~A\n"
(current-error-port) system)
"guix-data-service: computing the derivation-file-name for ~A\n"
system)
(let ((manifest (let ((manifest
(catch #t (catch #t
(lambda () (lambda ()
((channel-instances->manifest instances #:system system) store)) ((channel-instances->manifest instances #:system system) store))
(lambda (key . args) (lambda (key . args)
(simple-format (simple-format
(current-error-port) (current-error-port)
"error: while computing manifest entry derivation for ~A\n" "error: while computing manifest entry derivation for ~A\n"
system) system)
(simple-format (simple-format
(current-error-port) (current-error-port)
"error ~A: ~A\n" key args) "error ~A: ~A\n" key args)
#f)))) #f))))
(define (add-tmp-root-and-return-drv drv) (define (add-tmp-root-and-return-drv drv)
(add-temp-root store drv) (add-temp-root store drv)
drv) drv)
`(,system `((manifest-entry-item
. . ,(and manifest
((manifest-entry-item (add-tmp-root-and-return-drv
. ,(and manifest (derivation-file-name
(manifest-entry-item
(first
(manifest-entries manifest)))))))
(profile
. ,(catch #t
(lambda ()
(and manifest
(add-tmp-root-and-return-drv (add-tmp-root-and-return-drv
(derivation-file-name (derivation-file-name
(manifest-entry-item (parameterize ((%current-system system))
(first (run-with-store store
(manifest-entries manifest))))))) (profile-derivation
(profile manifest
. ,(catch #t #:hooks %channel-profile-hooks)))))))
(lambda () (lambda (key . args)
(and manifest (simple-format
(add-tmp-root-and-return-drv (current-error-port)
(derivation-file-name "error: while computing profile derivation for ~A\n"
(parameterize ((%current-system system)) system)
(run-with-store store (simple-format
(profile-derivation (current-error-port)
manifest "error ~A: ~A\n" key args)
#:hooks %channel-profile-hooks))))))) #f))))))))
(lambda (key . args)
(simple-format
(current-error-port)
"error: while computing profile derivation for ~A\n"
system)
(simple-format
(current-error-port)
"error ~A: ~A\n" key args)
#f)))))))
(list ,@systems)))))
(define (start-inferior inferior-store)
(let ((inferior (let ((inferior
(if use-container? (if use-container?
(open-inferior/container (open-inferior/container
@ -985,85 +1084,99 @@
(open-inferior (guix-store-path store) (open-inferior (guix-store-path store)
#:error-port (current-error-port)))))) #:error-port (current-error-port))))))
(define (start-inferior-and-return-derivation-file-names) ;; /etc is only missing if open-inferior/container has been used
;; /etc is only missing if open-inferior/container has been used (when use-container?
(when use-container? (inferior-eval
(inferior-eval '(begin
'(begin ;; Create /etc/pass, as %known-shorthand-profiles in (guix
;; Create /etc/pass, as %known-shorthand-profiles in (guix ;; profiles) tries to read from this file. Because the environment
;; profiles) tries to read from this file. Because the environment ;; is cleaned in build-self.scm, xdg-directory in (guix utils)
;; is cleaned in build-self.scm, xdg-directory in (guix utils) ;; falls back to accessing /etc/passwd.
;; falls back to accessing /etc/passwd. (mkdir "/etc")
(mkdir "/etc") (call-with-output-file "/etc/passwd"
(call-with-output-file "/etc/passwd" (lambda (port)
(lambda (port) (display "root:x:0:0::/root:/bin/bash" port))))
(display "root:x:0:0::/root:/bin/bash" port)))) inferior))
inferior))
(let ((channel-instance (inferior-eval '(use-modules (srfi srfi-1)
;; Obtain a session level lock here, to avoid conflicts with (ice-9 history)
;; other jobs over the Git repository. (guix channels)
(with-advisory-session-lock/log-time (guix grafts)
conn (guix profiles))
'latest-channel-instances inferior)
(lambda () (inferior-eval '(%graft? #f)
(first inferior)
(latest-channel-instances store (inferior-eval '(disable-value-history!)
(list channel) inferior)
#:authenticate? (inferior-eval '(define channel-instance
fetch-with-authentication?)))))) (@@ (guix channels) channel-instance))
(inferior-eval '(use-modules (srfi srfi-1) inferior)
(ice-9 history)
(guix channels)
(guix grafts)
(guix profiles))
inferior)
(inferior-eval '(%graft? #f)
inferior)
(inferior-eval '(disable-value-history!)
inferior)
(inferior-eval '(define channel-instance
(@@ (guix channels) channel-instance))
inferior)
(let* ((systems inferior))
(inferior-eval '(@ (guix packages) %supported-systems)
inferior))
(result
(inferior-eval-with-store
inferior
store
(inferior-code channel-instance systems))))
(close-inferior inferior) (let* ((channel-instance
;; Obtain a session level lock here, to avoid conflicts with
;; other jobs over the Git repository.
(with-advisory-session-lock/log-time
conn
'latest-channel-instances
(lambda ()
;; TODO (guix serialization) uses dynamic-wind
(call-with-temporary-thread
(lambda ()
(first
(latest-channel-instances store
(list channel)
#:authenticate?
fetch-with-authentication?)))))))
(inferior-and-store-pool
(make-resource-pool
(lambda ()
(let* ((inferior-store (open-connection))
(inferior (start-inferior inferior-store)))
(ensure-non-blocking-store-connection inferior-store)
(make-inferior-non-blocking! inferior)
(cons inferior inferior-store)))
parallelism
#:min-size 0
#:idle-seconds 10
#:destructor (match-lambda
((inferior . store)
(close-inferior inferior)
(close-connection store)))))
(systems
(with-resource-from-pool inferior-and-store-pool res
(match res
((inferior . inferior-store)
(inferior-eval '(@ (guix packages) %supported-systems)
inferior)))))
(result
(par-map&
(lambda (system)
(with-resource-from-pool inferior-and-store-pool res
(match res
((inferior . inferior-store)
(cons system
(inferior-eval-with-store/non-blocking
inferior
inferior-store
(inferior-code channel-instance system)))))))
systems)))
(cons (cons
(channel-instance-checkout channel-instance) (channel-instance-checkout channel-instance)
result)))) result)))
(catch (define* (channel->source-and-derivations-by-system conn store channel
#t fetch-with-authentication?
(lambda () #:key parallelism)
(with-throw-handler #t
start-inferior-and-return-derivation-file-names
(lambda (key . parameters)
(display (backtrace) (current-error-port))
(display "\n" (current-error-port))
(simple-format (current-error-port)
"error: channel->derivation-file-names-by-system: ~A: ~A\n"
key parameters))))
(lambda args
(close-inferior inferior)
#f))))
(define (channel->source-and-derivations-by-system conn store channel
fetch-with-authentication?)
(match (with-time-logging "computing the channel derivation" (match (with-time-logging "computing the channel derivation"
(channel->source-and-derivation-file-names-by-system (channel->source-and-derivation-file-names-by-system
conn conn
store store
channel channel
fetch-with-authentication?)) fetch-with-authentication?
#:parallelism parallelism))
((source . derivation-file-names-by-system) ((source . derivation-file-names-by-system)
(for-each (for-each
(match-lambda (match-lambda
@ -1148,17 +1261,9 @@
output))) output)))
(define (start-inferior-for-data-extration store store-path) (define (start-inferior-for-data-extration store store-path guix-locpath)
(let* ((guix-locpath (getenv "GUIX_LOCPATH")) (let* ((original-guix-locpath (getenv "GUIX_LOCPATH"))
(inf (let ((guix-locpath (inf (begin
;; Augment the GUIX_LOCPATH to include glibc-locales from
;; the Guix at store-path, this should mean that the
;; inferior Guix works, even if it's build using a different
;; glibc version
(string-append
(glibc-locales-for-guix-store-path store store-path)
"/lib/locale"
":" guix-locpath)))
;; Unset the GUILE_LOAD_PATH and GUILE_LOAD_COMPILED_PATH to ;; Unset the GUILE_LOAD_PATH and GUILE_LOAD_COMPILED_PATH to
;; avoid the values for these being used in the ;; avoid the values for these being used in the
;; inferior. Even though the inferior %load-path and ;; inferior. Even though the inferior %load-path and
@ -1185,7 +1290,7 @@
(simple-format #t "debug: using open-inferior\n") (simple-format #t "debug: using open-inferior\n")
(open-inferior store-path (open-inferior store-path
#:error-port (current-error-port))))))) #:error-port (current-error-port)))))))
(setenv "GUIX_LOCPATH" guix-locpath) ; restore GUIX_LOCPATH (setenv "GUIX_LOCPATH" original-guix-locpath) ; restore GUIX_LOCPATH
(when (eq? inf #f) (when (eq? inf #f)
(error "error: inferior is #f")) (error "error: inferior is #f"))
@ -1202,6 +1307,7 @@
(inferior-eval '(use-modules (srfi srfi-1) (inferior-eval '(use-modules (srfi srfi-1)
(srfi srfi-34) (srfi srfi-34)
(srfi srfi-43)
(ice-9 history) (ice-9 history)
(guix grafts) (guix grafts)
(guix derivations) (guix derivations)
@ -1221,144 +1327,187 @@
(inferior-eval '(when (defined? '%graft?) (%graft? #f)) inf) (inferior-eval '(when (defined? '%graft?) (%graft? #f)) inf)
;; TODO Have Guix make this easier
((@@ (guix inferior) ensure-store-bridge!) inf)
(non-blocking-port ((@@ (guix inferior) inferior-bridge-socket) inf))
inf)) inf))
(define* (extract-information-from conn store guix-revision-id commit (define* (extract-information-from conn store guix-revision-id commit
guix-source store-path guix-source store-path
#:key skip-system-tests?) #:key skip-system-tests?
parallelism)
(define guix-locpath
;; Augment the GUIX_LOCPATH to include glibc-locales from
;; the Guix at store-path, this should mean that the
;; inferior Guix works, even if it's build using a different
;; glibc version
(string-append
(glibc-locales-for-guix-store-path store store-path)
"/lib/locale"
":" (getenv "GUIX_LOCPATH")))
(define inf-and-store-pool
(make-resource-pool
(lambda ()
(let* ((inferior-store (open-connection))
(inferior (start-inferior-for-data-extration inferior-store
store-path
guix-locpath)))
(ensure-non-blocking-store-connection inferior-store)
(make-inferior-non-blocking! inferior)
(cons inferior inferior-store)))
parallelism
#:min-size 0
#:idle-seconds 10
#:destructor (match-lambda
((inferior . store)
(close-inferior inferior)
(close-connection store)))))
(simple-format #t "debug: extract-information-from: ~A\n" store-path) (simple-format #t "debug: extract-information-from: ~A\n" store-path)
(let ((inf (start-inferior-for-data-extration store store-path))) (letpar& ((inferior-lint-checkers-and-warnings-data
(catch (with-resource-from-pool inf-and-store-pool res
#t (match res
(lambda () ((inferior . inferior-store)
(let* ((packages (let ((inferior-lint-checkers-data
(with-time-logging "fetching inferior packages" (inferior-lint-checkers inferior)))
(inferior-packages-plus-replacements inf))) (cons
(inferior-lint-checkers-data inferior-lint-checkers-data
(inferior-lint-checkers inf)) (if inferior-lint-checkers-data
(inferior-lint-warnings-data (with-time-logging "fetching inferior lint warnings"
(and inferior-lint-checkers-data (map
(with-time-logging "fetching inferior lint warnings" (match-lambda
(map ((checker-name _ network-dependent?)
(match-lambda (and (and (not network-dependent?)
((checker-name _ network-dependent?) ;; Running the derivation linter is
(and (and (not network-dependent?) ;; currently infeasible
;; Running the derivation linter is (not (eq? checker-name 'derivation)))
;; currently infeasible (inferior-lint-warnings inferior
(not (eq? checker-name 'derivation))) inferior-store
(inferior-lint-warnings inf checker-name))))
store inferior-lint-checkers-data))
checker-name)))) #f)))))))
inferior-lint-checkers-data)))) (inferior-packages-system-and-target-to-derivations-alist
(inferior-system-target-pairs (with-time-logging "getting inferior derivations"
(inferior-fetch-system-target-pairs inf)) (par-map&
(inferior-packages-system-and-target-to-derivations-alist (match-lambda
(with-time-logging "getting inferior derivations" ((system . target)
(map (with-resource-from-pool inf-and-store-pool res
(match-lambda (match res
((system . target) ((inferior . inferior-store)
(cons (cons system target) (ensure-gds-inferior-packages-defined! inferior)
(inferior-package-derivations store
inf
system
target))))
inferior-system-target-pairs)))
(inferior-system-tests
(if skip-system-tests?
(begin
(simple-format #t "debug: skipping system tests\n")
'())
(with-time-logging "getting inferior system tests"
(all-inferior-system-tests inf store
guix-source commit))))
(packages-data
(with-time-logging "getting all inferior package data"
(all-inferior-packages-data inf packages))))
(cons (cons system target)
(inferior-package-derivations inferior-store
inferior
system
target)))))))
(with-resource-from-pool inf-and-store-pool res
(match res
((inferior . inferior-store)
(inferior-fetch-system-target-pairs inferior)))))))
(inferior-system-tests
(if skip-system-tests?
(begin
(simple-format #t "debug: skipping system tests\n")
'())
(with-resource-from-pool inf-and-store-pool res
(match res
((inferior . inferior-store)
(with-time-logging "getting inferior system tests"
(all-inferior-system-tests inferior inferior-store
guix-source commit)))))))
(packages-data
(with-time-logging "getting all inferior package data"
(with-resource-from-pool inf-and-store-pool res
(match res
((inferior . inferior-store)
(with-time-logging "fetching inferior packages"
(let ((packages
pkg-to-replacement-hash-table
(inferior-packages-plus-replacements inferior)))
(all-inferior-packages-data inferior
packages
pkg-to-replacement-hash-table)))))))))
(destroy-resource-pool inf-and-store-pool)
(simple-format
#t "debug: finished loading information from inferior\n")
(with-time-logging
"acquiring advisory transaction lock: load-new-guix-revision-inserts"
;; Wait until this is the only transaction inserting data, to
;; avoid any concurrency issues
(obtain-advisory-transaction-lock conn
'load-new-guix-revision-inserts))
(with-time-logging
"inserting data"
(let* ((package-ids
(insert-packages conn packages-data)))
(when inferior-lint-warnings
(let* ((lint-checker-ids
(lint-checkers->lint-checker-ids
conn
(map (match-lambda
((name descriptions-by-locale network-dependent)
(list
name
network-dependent
(lint-checker-description-data->lint-checker-description-set-id
conn descriptions-by-locale))))
(car inferior-lint-checkers-and-warnings-data))))
(lint-warning-ids
(insert-lint-warnings
conn
package-ids
lint-checker-ids
(cdr inferior-lint-checkers-and-warnings-data))))
(insert-guix-revision-lint-checkers conn
guix-revision-id
lint-checker-ids)
(chunk-for-each!
(lambda (lint-warning-ids-chunk)
(insert-guix-revision-lint-warnings conn
guix-revision-id
lint-warning-ids-chunk))
5000
lint-warning-ids)))
(when inferior-system-tests
(insert-system-tests-for-guix-revision conn
guix-revision-id
inferior-system-tests))
(let* ((package-derivation-ids
(with-time-logging "inferior-data->package-derivation-ids"
(inferior-data->package-derivation-ids
conn
inf
package-ids
inferior-packages-system-and-target-to-derivations-alist)))
(ids-count
(length package-derivation-ids)))
(chunk-for-each! (lambda (package-derivation-ids-chunk)
(insert-guix-revision-package-derivations
conn
guix-revision-id
package-derivation-ids-chunk))
2000
package-derivation-ids)
(simple-format (simple-format
#t "debug: finished loading information from inferior\n") #t "Successfully loaded ~A package/derivation pairs\n"
(close-inferior inf) ids-count))
(with-time-logging (with-time-logging
"acquiring advisory transaction lock: load-new-guix-revision-inserts" "insert-guix-revision-package-derivation-distribution-counts"
;; Wait until this is the only transaction inserting data, to (insert-guix-revision-package-derivation-distribution-counts
;; avoid any concurrency issues conn
(obtain-advisory-transaction-lock conn guix-revision-id))))))
'load-new-guix-revision-inserts))
(with-time-logging
"inserting data"
(let* ((package-ids
(insert-packages conn packages-data)))
(when inferior-lint-warnings
(let* ((lint-checker-ids
(lint-checkers->lint-checker-ids
conn
(map (match-lambda
((name descriptions-by-locale network-dependent)
(list
name
network-dependent
(lint-checker-description-data->lint-checker-description-set-id
conn descriptions-by-locale))))
inferior-lint-checkers-data)))
(lint-warning-ids
(insert-lint-warnings
conn
package-ids
lint-checker-ids
inferior-lint-warnings-data)))
(insert-guix-revision-lint-checkers conn
guix-revision-id
lint-checker-ids)
(chunk-for-each!
(lambda (lint-warning-ids-chunk)
(insert-guix-revision-lint-warnings conn
guix-revision-id
lint-warning-ids-chunk))
5000
lint-warning-ids)))
(when inferior-system-tests
(insert-system-tests-for-guix-revision conn
guix-revision-id
inferior-system-tests))
(let* ((package-derivation-ids
(with-time-logging "inferior-data->package-derivation-ids"
(inferior-data->package-derivation-ids
conn
inf
package-ids
inferior-packages-system-and-target-to-derivations-alist)))
(ids-count
(length package-derivation-ids)))
(chunk-for-each! (lambda (package-derivation-ids-chunk)
(insert-guix-revision-package-derivations
conn
guix-revision-id
package-derivation-ids-chunk))
2000
package-derivation-ids)
(simple-format
#t "Successfully loaded ~A package/derivation pairs\n"
ids-count))
(with-time-logging
"insert-guix-revision-package-derivation-distribution-counts"
(insert-guix-revision-package-derivation-distribution-counts
conn
guix-revision-id)))))
#t)
(lambda (key . args)
(simple-format (current-error-port)
"Failed extracting information from commit: ~A\n\n" commit)
(simple-format (current-error-port)
" ~A ~A\n\n" key args)
#f)
(lambda (key . args)
(display-backtrace (make-stack #t) (current-error-port))))))
(prevent-inlining-for-tests extract-information-from) (prevent-inlining-for-tests extract-information-from)
@ -1409,7 +1558,7 @@
(prevent-inlining-for-tests load-channel-instances) (prevent-inlining-for-tests load-channel-instances)
(define* (load-new-guix-revision conn store git-repository-id commit (define* (load-new-guix-revision conn store git-repository-id commit
#:key skip-system-tests?) #:key skip-system-tests? parallelism)
(let* ((git-repository-fields (let* ((git-repository-fields
(select-git-repository conn git-repository-id)) (select-git-repository conn git-repository-id))
(git-repository-url (git-repository-url
@ -1421,10 +1570,12 @@
(url git-repository-url) (url git-repository-url)
(commit commit))) (commit commit)))
(source-and-channel-derivations-by-system (source-and-channel-derivations-by-system
(channel->source-and-derivations-by-system conn (channel->source-and-derivations-by-system
store conn
channel-for-commit store
fetch-with-authentication?)) channel-for-commit
fetch-with-authentication?
#:parallelism parallelism))
(guix-source (guix-source
(car source-and-channel-derivations-by-system)) (car source-and-channel-derivations-by-system))
(channel-derivations-by-system (channel-derivations-by-system
@ -1442,7 +1593,8 @@
guix-revision-id guix-revision-id
commit guix-source store-item commit guix-source store-item
#:skip-system-tests? #:skip-system-tests?
skip-system-tests?) skip-system-tests?
#:parallelism parallelism)
(if (defined? 'channel-news-for-commit (if (defined? 'channel-news-for-commit
(resolve-module '(guix channels))) (resolve-module '(guix channels)))
@ -1817,13 +1969,15 @@ SKIP LOCKED")
(define (with-store-connection f) (define (with-store-connection f)
(with-store store (with-store store
(ensure-non-blocking-store-connection store)
(set-build-options store #:fallback? #t) (set-build-options store #:fallback? #t)
(f store))) (f store)))
(prevent-inlining-for-tests with-store-connection) (prevent-inlining-for-tests with-store-connection)
(define* (process-load-new-guix-revision-job id #:key skip-system-tests?) (define* (process-load-new-guix-revision-job id #:key skip-system-tests?
parallelism)
(with-postgresql-connection (with-postgresql-connection
(simple-format #f "load-new-guix-revision ~A" id) (simple-format #f "load-new-guix-revision ~A" id)
(lambda (conn) (lambda (conn)
@ -1860,7 +2014,8 @@ SKIP LOCKED")
store store
git-repository-id git-repository-id
commit commit
#:skip-system-tests? #t)))) #:skip-system-tests? #t
#:parallelism parallelism))))
(lambda (key . args) (lambda (key . args)
(simple-format (current-error-port) (simple-format (current-error-port)
"error: load-new-guix-revision: ~A ~A\n" "error: load-new-guix-revision: ~A ~A\n"

5
scripts/guix-data-service-process-job.in
View file

@ -78,7 +78,8 @@
(lambda () (lambda ()
(process-load-new-guix-revision-job (process-load-new-guix-revision-job
job job
#:skip-system-tests? (assq-ref opts 'skip-system-tests))) #:skip-system-tests? (assq-ref opts 'skip-system-tests)
#:parallelism (assq-ref opts 'parallelism)))
#:hz 0 #:hz 0
#:parallelism (assq-ref opts 'parallelism) #:parallelism 1
#:drain? #t))))) #:drain? #t)))))

14
scripts/guix-data-service-process-jobs.in
View file

@ -44,11 +44,17 @@
result))) result)))
(option '("skip-system-tests") #f #f (option '("skip-system-tests") #f #f
(lambda (opt name _ result) (lambda (opt name _ result)
(alist-cons 'skip-system-tests #t result))))) (alist-cons 'skip-system-tests #t result)))
(option '("per-job-parallelism") #t #f
(lambda (opt name arg result)
(alist-cons 'per-job-parallelism
(string->number arg)
result)))))
(define %default-options (define %default-options
;; Alist of default option values ;; Alist of default option values
`((max-processes . ,default-max-processes))) `((max-processes . ,default-max-processes)
(per-job-parallelism . 1)))
(define (parse-options args) (define (parse-options args)
(args-fold (args-fold
@ -77,4 +83,6 @@
(or (assq-ref opts 'latest-branch-revision-max-processes) (or (assq-ref opts 'latest-branch-revision-max-processes)
(* 2 (assq-ref opts 'max-processes))) (* 2 (assq-ref opts 'max-processes)))
#:skip-system-tests? #:skip-system-tests?
(assq-ref opts 'skip-system-tests))))) (assq-ref opts 'skip-system-tests)
#:per-job-parallelism
(assq-ref opts 'per-job-parallelism)))))