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

Rewrite the key parts of loading data to be even more parallel

Browse source 
Use a pool for the database connection, and a fibers promise for the package
ids, and run other parts of the process in parallel too. This change also
means that inserting starts as soon as some data is available, rather than
when all the data is available.
This commit is contained in:
Christopher Baines 2024-07-19 17:06:56 +01:00
parent 5439159a16
commit 3f1c2ad603
1 changed files with 169 additions and 172 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -902,32 +902,6 @@
lint-checker-ids lint-checker-ids
lint-warnings-data))) lint-warnings-data)))
(define (inferior-data->package-derivation-ids
conn inf
package-ids
inferior-packages-system-and-target-to-derivations-alist)
(append-map!
(lambda (data)
(let* ((system-and-target (car data))
(derivations-vector (cdr data))
(derivation-ids
(with-time-logging
(simple-format #f "derivation-file-names->derivation-ids (~A)"
system-and-target)
(derivation-file-names->derivation-ids
conn
derivations-vector))))
(with-time-logging
(simple-format #f "insert-package-derivations (~A)"
system-and-target)
(insert-package-derivations conn
(car system-and-target)
(or (cdr system-and-target) "")
package-ids
derivation-ids))))
inferior-packages-system-and-target-to-derivations-alist))
(define guix-store-path (define guix-store-path
(let ((store-path #f)) (let ((store-path #f))
(lambda (store) (lambda (store)
@ -1418,8 +1392,8 @@
inf)))) inf))))
(define* (extract-information-from conn guix-revision-id commit (define* (extract-information-from db-conn guix-revision-id commit
guix-source store-path guix-source store-item
#:key skip-system-tests? #:key skip-system-tests?
extra-inferior-environment-variables extra-inferior-environment-variables
parallelism) parallelism)
@ -1432,7 +1406,7 @@
(string-append (string-append
(with-store-connection (with-store-connection
(lambda (store) (lambda (store)
(glibc-locales-for-guix-store-path store store-path))) (glibc-locales-for-guix-store-path store store-item)))
"/lib/locale" "/lib/locale"
":" (getenv "GUIX_LOCPATH"))) ":" (getenv "GUIX_LOCPATH")))
@ -1442,7 +1416,7 @@
(let* ((inferior-store (open-store-connection)) (let* ((inferior-store (open-store-connection))
(inferior (start-inferior-for-data-extration (inferior (start-inferior-for-data-extration
inferior-store inferior-store
store-path store-item
guix-locpath guix-locpath
extra-inferior-environment-variables))) extra-inferior-environment-variables)))
(ensure-non-blocking-store-connection inferior-store) (ensure-non-blocking-store-connection inferior-store)
@ -1462,17 +1436,58 @@
(close-connection store) (close-connection store)
(close-inferior inferior))))) (close-inferior inferior)))))
(simple-format #t "debug: extract-information-from: ~A\n" store-path) (define postgresql-connection-pool
(make-resource-pool
(lambda ()
(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 db-conn
'load-new-guix-revision-inserts))
db-conn)
1
#:min-size 0))
(letpar& ((inferior-lint-checkers-and-warnings-data (define package-ids-promise
(let ((inferior-lint-checkers-data (fibers-delay
(lambda ()
(let ((packages-data
(with-resource-from-pool inf-and-store-pool res (with-resource-from-pool inf-and-store-pool res
(match res (match res
((inferior . inferior-store) ((inferior . inferior-store)
(inferior-lint-checkers inferior)))))) (with-time-logging "getting all inferior package data"
(cons (let ((packages
inferior-lint-checkers-data pkg-to-replacement-hash-table
(and inferior-lint-checkers-data (inferior-packages-plus-replacements inferior)))
(all-inferior-packages-data
inferior
packages
pkg-to-replacement-hash-table))))))))
(with-resource-from-pool postgresql-connection-pool conn
(insert-packages conn packages-data))))))
(define (extract-and-store-lint-checkers-and-warnings)
(define inferior-lint-checkers-data
(with-resource-from-pool inf-and-store-pool res
(match res
((inferior . inferior-store)
(inferior-lint-checkers inferior)))))
(when inferior-lint-checkers-data
(letpar& ((lint-checker-ids
(with-resource-from-pool postgresql-connection-pool conn
(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-warnings-data
(par-map& (par-map&
(match-lambda (match-lambda
((checker-name _ network-dependent?) ((checker-name _ network-dependent?)
@ -1486,9 +1501,30 @@
(inferior-lint-warnings inferior (inferior-lint-warnings inferior
inferior-store inferior-store
checker-name))))))) checker-name)))))))
inferior-lint-checkers-data))))) inferior-lint-checkers-data)))
(inferior-packages-system-and-target-to-derivations-alist
(par-map& (let ((package-ids (fibers-force package-ids-promise)))
(with-resource-from-pool postgresql-connection-pool conn
(insert-guix-revision-lint-checkers conn
guix-revision-id
lint-checker-ids)
(let ((lint-warning-ids
(insert-lint-warnings
conn
package-ids
lint-checker-ids
lint-warnings-data)))
(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)))))))
(define (extract-and-store-package-derivations)
(fibers-for-each
(match-lambda (match-lambda
((system . target) ((system . target)
(let loop ((wal-bytes (stat:size (stat "/var/guix/db/db.sqlite-wal")))) (let loop ((wal-bytes (stat:size (stat "/var/guix/db/db.sqlite-wal"))))
@ -1499,6 +1535,7 @@
(sleep 30) (sleep 30)
(loop (stat:size (stat "/var/guix/db/db.sqlite-wal"))))) (loop (stat:size (stat "/var/guix/db/db.sqlite-wal")))))
(let ((derivations-vector
(with-resource-from-pool inf-and-store-pool res (with-resource-from-pool inf-and-store-pool res
(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))
@ -1506,117 +1543,77 @@
((inferior . inferior-store) ((inferior . inferior-store)
(ensure-gds-inferior-packages-defined! inferior) (ensure-gds-inferior-packages-defined! inferior)
(let ((drvs
(inferior-package-derivations (inferior-package-derivations
inferior-store inferior-store
inferior inferior
system system
target))) target)))))))
(cons (cons system target)
drvs))))))))
(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-resource-from-pool inf-and-store-pool res
(match res
((inferior . inferior-store)
(with-time-logging "getting all inferior package data"
(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")
(let ((package-ids (fibers-force package-ids-promise)))
(with-resource-from-pool postgresql-connection-pool conn
(let* ((derivation-ids
(with-time-logging (with-time-logging
"acquiring advisory transaction lock: load-new-guix-revision-inserts" (simple-format #f "derivation-file-names->derivation-ids (~A ~A)"
;; Wait until this is the only transaction inserting data, to system target)
;; avoid any concurrency issues (derivation-file-names->derivation-ids
(obtain-advisory-transaction-lock conn
'load-new-guix-revision-inserts))
(with-time-logging
"inserting data"
(let* ((package-ids
(insert-packages conn packages-data)))
(let* ((package-derivation-ids
(with-time-logging "inferior-data->package-derivation-ids"
(inferior-data->package-derivation-ids
conn conn
inf derivations-vector))))
(let ((package-derivation-ids
(with-time-logging
(simple-format #f "insert-package-derivations (~A ~A)"
system target)
(insert-package-derivations conn
system
(or target "")
package-ids package-ids
inferior-packages-system-and-target-to-derivations-alist))) derivation-ids))))
(ids-count
(length package-derivation-ids)))
(chunk-for-each! (lambda (package-derivation-ids-chunk) (chunk-for-each! (lambda (package-derivation-ids-chunk)
(insert-guix-revision-package-derivations (insert-guix-revision-package-derivations
conn conn
guix-revision-id guix-revision-id
package-derivation-ids-chunk)) package-derivation-ids-chunk))
2000 2000
package-derivation-ids) package-derivation-ids))))))))
(simple-format (with-resource-from-pool inf-and-store-pool res
#t "Successfully loaded ~A package/derivation pairs\n" (match res
ids-count)) ((inferior . inferior-store)
(inferior-fetch-system-target-pairs inferior)))))
(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))
(with-resource-from-pool postgresql-connection-pool conn
(with-time-logging (with-time-logging
"insert-guix-revision-package-derivation-distribution-counts" "insert-guix-revision-package-derivation-distribution-counts"
(insert-guix-revision-package-derivation-distribution-counts (insert-guix-revision-package-derivation-distribution-counts
conn conn
guix-revision-id)))))) guix-revision-id))))
(define (extract-and-store-system-tests)
(if skip-system-tests?
(begin
(simple-format #t "debug: skipping system tests\n")
'())
(let ((data
(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)))))))
(when data
(with-resource-from-pool postgresql-connection-pool conn
(insert-system-tests-for-guix-revision conn
guix-revision-id
data))))))
(simple-format #t "debug: extract-information-from: ~A\n" store-path)
(parallel-via-fibers
(fibers-force package-ids-promise)
(extract-and-store-lint-checkers-and-warnings)
(extract-and-store-package-derivations)
(extract-and-store-system-tests)))
(prevent-inlining-for-tests extract-information-from) (prevent-inlining-for-tests extract-information-from)