allaun
e5bfdc0492
Containerized LKB worker (Dockerfile.lkb), Postgres-backed queue, auto-deploy timer, and eigensolid convergence check module. Build: 0 jobs (no Lean build needed)
280 lines
9 KiB
R
280 lines
9 KiB
R
source("boot.R")
|
|
box_use(./lkb)
|
|
box_use(./laurent2)
|
|
|
|
CHUNK_SIZE <- 5L
|
|
CHECKPOINT_INTERVAL <- 20L
|
|
STUCK_TIMEOUT_MINUTES <- 60L
|
|
|
|
chunk_cache_mem <- new.env(parent = emptyenv(), size = 5000L)
|
|
|
|
#' @export
|
|
chunk_key <- function(gens_seq) {
|
|
paste(gens_seq, collapse = ".")
|
|
}
|
|
|
|
#' @export
|
|
seed_generators <- function(con, n) {
|
|
expected <- n - 1L
|
|
row <- DBI::dbGetQuery(con,
|
|
"SELECT COUNT(*) AS cnt FROM lkb.generators WHERE n = ?",
|
|
params = list(n))
|
|
if (row$cnt[[1]] == expected) {
|
|
cat(sprintf(" Generators for B%d already cached (%d matrices)\n", n, expected))
|
|
return(invisible(NULL))
|
|
}
|
|
|
|
lock_key <- as.integer(bitwXor(n, 0x4C4B47L))
|
|
DBI::dbExecute(con, "SELECT pg_advisory_lock(?)", params = list(lock_key))
|
|
tryCatch({
|
|
row <- DBI::dbGetQuery(con,
|
|
"SELECT COUNT(*) AS cnt FROM lkb.generators WHERE n = ?",
|
|
params = list(n))
|
|
if (row$cnt[[1]] == expected) {
|
|
cat(sprintf(" Generators for B%d already seeded by another worker\n", n))
|
|
return(invisible(NULL))
|
|
}
|
|
|
|
cat(sprintf(" Seeding generator matrices for B%d...\n", n))
|
|
DBI::dbExecute(con, "DELETE FROM lkb.generators WHERE n = ?", params = list(n))
|
|
gens <- lkb$lkb_generators(n)
|
|
for (g_idx in seq_along(gens)) {
|
|
blob <- laurent2$serialize_matrix(gens[[g_idx]])
|
|
DBI::dbExecute(con,
|
|
"INSERT INTO lkb.generators (n, gen_idx, matrix_data) VALUES (?, ?, ?) ON CONFLICT (n, gen_idx) DO UPDATE SET matrix_data = EXCLUDED.matrix_data",
|
|
params = list(n, g_idx, blob))
|
|
}
|
|
cat(sprintf(" Seeded %d generator matrices\n", length(gens)))
|
|
}, finally = {
|
|
DBI::dbExecute(con, "SELECT pg_advisory_unlock(?)", params = list(lock_key))
|
|
})
|
|
}
|
|
|
|
#' @export
|
|
load_generators <- function(con, n) {
|
|
rows <- DBI::dbGetQuery(con,
|
|
"SELECT gen_idx, matrix_data FROM lkb.generators WHERE n = ? ORDER BY gen_idx",
|
|
params = list(n))
|
|
gens <- list()
|
|
for (r in seq_len(nrow(rows))) {
|
|
gens[[rows$gen_idx[[r]]]] <- laurent2$deserialize_matrix(rows$matrix_data[[r]])
|
|
}
|
|
gens
|
|
}
|
|
|
|
#' @export
|
|
save_checkpoint <- function(con, eq_id, step, mat) {
|
|
blob <- laurent2$serialize_matrix(mat)
|
|
DBI::dbExecute(con,
|
|
"INSERT INTO lkb.checkpoint (equation_id, step, matrix_data) VALUES (?, ?, ?) ON CONFLICT (equation_id, step) DO UPDATE SET matrix_data = EXCLUDED.matrix_data",
|
|
params = list(eq_id, step, blob))
|
|
DBI::dbExecute(con,
|
|
"UPDATE lkb.compute_queue SET steps_done = ? WHERE equation_id = ?",
|
|
params = list(step, eq_id))
|
|
}
|
|
|
|
#' @export
|
|
load_checkpoint <- function(con, eq_id, step) {
|
|
row <- DBI::dbGetQuery(con,
|
|
"SELECT matrix_data FROM lkb.checkpoint WHERE equation_id = ? AND step = ?",
|
|
params = list(eq_id, step))
|
|
if (nrow(row) == 0) return(NULL)
|
|
laurent2$deserialize_matrix(row$matrix_data[[1]])
|
|
}
|
|
|
|
#' @export
|
|
chunk_product <- function(con, gens, chunk, n) {
|
|
key <- chunk_key(chunk)
|
|
|
|
cached <- chunk_cache_mem[[key]]
|
|
if (!is.null(cached)) return(cached)
|
|
|
|
row <- DBI::dbGetQuery(con,
|
|
"SELECT matrix_data FROM lkb.subseq_cache WHERE n = ? AND seq_key = ?",
|
|
params = list(n, key))
|
|
if (nrow(row) > 0) {
|
|
mat <- laurent2$deserialize_matrix(row$matrix_data[[1]])
|
|
chunk_cache_mem[[key]] <<- mat
|
|
tryCatch(
|
|
DBI::dbExecute(con,
|
|
"UPDATE lkb.subseq_cache SET hit_count = hit_count + 1 WHERE n = ? AND seq_key = ?",
|
|
params = list(n, key)),
|
|
error = function(e) NULL
|
|
)
|
|
return(mat)
|
|
}
|
|
|
|
N <- n * (n - 1) / 2
|
|
mat <- laurent2$l2_mat_identity(N)
|
|
for (g in chunk) {
|
|
mat <- laurent2$l2_mat_mul(mat, gens[[g]])
|
|
}
|
|
|
|
chunk_cache_mem[[key]] <<- mat
|
|
blob <- laurent2$serialize_matrix(mat)
|
|
DBI::dbExecute(con,
|
|
"INSERT INTO lkb.subseq_cache (seq_key, n, matrix_data) VALUES (?, ?, ?) ON CONFLICT (n, seq_key) DO NOTHING",
|
|
params = list(key, n, blob))
|
|
|
|
mat
|
|
}
|
|
|
|
#' @export
|
|
chunk_cache_stats <- function() {
|
|
list(memory_entries = length(ls(chunk_cache_mem)))
|
|
}
|
|
|
|
#' @export
|
|
chunk_cache_clear <- function() {
|
|
rm(list = ls(chunk_cache_mem), envir = chunk_cache_mem)
|
|
}
|
|
|
|
#' @export
|
|
enqueue <- function(con, equation_id, sidon_set, braid_word, n) {
|
|
word_str <- paste(unclass(braid_word), collapse = ",")
|
|
if (word_str == "") word_str <- "ID"
|
|
DBI::dbExecute(con,
|
|
"INSERT INTO lkb.compute_queue (equation_id, sidon_set, braid_word, n, word_length) VALUES (?, ?, ?, ?, ?) ON CONFLICT (equation_id) DO UPDATE SET status = 'pending', steps_done = 0, result_entry = NULL, started_at = NULL, finished_at = NULL",
|
|
params = list(equation_id, paste(sidon_set, collapse = ","), word_str, n, length(unclass(braid_word))))
|
|
}
|
|
|
|
#' @export
|
|
claim_next <- function(con) {
|
|
row <- DBI::dbGetQuery(con,
|
|
"UPDATE lkb.compute_queue SET status = 'running', started_at = now() WHERE equation_id = (SELECT equation_id FROM lkb.compute_queue WHERE status = 'pending' ORDER BY word_length, equation_id LIMIT 1 FOR UPDATE SKIP LOCKED) RETURNING *")
|
|
if (nrow(row) == 0) return(NULL)
|
|
row
|
|
}
|
|
|
|
#' @export
|
|
reap_stuck_jobs <- function(con) {
|
|
DBI::dbExecute(con,
|
|
sprintf("UPDATE lkb.compute_queue SET status = 'pending', steps_done = 0, started_at = NULL WHERE status = 'running' AND started_at < now() - interval '%d minutes'", STUCK_TIMEOUT_MINUTES))
|
|
}
|
|
|
|
#' @export
|
|
mark_done <- function(con, eq_id, result_entry) {
|
|
DBI::dbExecute(con,
|
|
"UPDATE lkb.compute_queue SET status = 'done', result_entry = ?, finished_at = now() WHERE equation_id = ?",
|
|
params = list(result_entry, eq_id))
|
|
}
|
|
|
|
#' @export
|
|
mark_error <- function(con, eq_id, msg = "") {
|
|
DBI::dbExecute(con,
|
|
"UPDATE lkb.compute_queue SET status = 'error', result_entry = ?, finished_at = now() WHERE equation_id = ?",
|
|
params = list(msg, eq_id))
|
|
}
|
|
|
|
#' @export
|
|
lkb_compute <- function(con) {
|
|
reap_stuck_jobs(con)
|
|
job <- claim_next(con)
|
|
if (is.null(job)) return(invisible(NULL))
|
|
eq_id <- job$equation_id[[1]]
|
|
n <- job$n[[1]]
|
|
N <- n * (n - 1) / 2
|
|
|
|
seed_generators(con, n)
|
|
gens <- load_generators(con, n)
|
|
|
|
word_str <- job$braid_word[[1]]
|
|
|
|
if (word_str == "ID" || word_str == "") {
|
|
I_mat <- laurent2$l2_mat_identity(N)
|
|
entry <- laurent2$l2_str(I_mat[[1, 3]])
|
|
mark_done(con, eq_id, entry)
|
|
cat(sprintf(" [%s] identity word, done\n", eq_id))
|
|
return(I_mat)
|
|
}
|
|
|
|
steps <- as.integer(strsplit(word_str, ",")[[1]])
|
|
start_step <- job$steps_done[[1]]
|
|
cat(sprintf(" [%s] B%d word, %d generators, resuming from step %d (N=%d)\n",
|
|
eq_id, n, length(steps), start_step, N))
|
|
|
|
result <- tryCatch({
|
|
if (start_step > 0) {
|
|
mat <- load_checkpoint(con, eq_id, start_step)
|
|
if (is.null(mat)) {
|
|
cat(" Checkpoint missing, restarting\n")
|
|
start_step <- 0
|
|
mat <- laurent2$l2_mat_identity(N)
|
|
}
|
|
} else {
|
|
mat <- laurent2$l2_mat_identity(N)
|
|
}
|
|
|
|
chunk_starts <- if (start_step < length(steps)) {
|
|
seq(start_step + 1L, length(steps), by = CHUNK_SIZE)
|
|
} else {
|
|
integer(0)
|
|
}
|
|
|
|
for (cs in chunk_starts) {
|
|
ce <- min(cs + CHUNK_SIZE - 1L, length(steps))
|
|
chunk <- steps[cs:ce]
|
|
|
|
cached_mat <- chunk_product(con, gens, chunk, n)
|
|
mat <- laurent2$l2_mat_mul(mat, cached_mat)
|
|
|
|
if (ce %% CHECKPOINT_INTERVAL < CHUNK_SIZE || ce == length(steps)) {
|
|
save_checkpoint(con, eq_id, ce, mat)
|
|
cat(sprintf(" Step %d/%d checkpointed (cache: %d entries)\n",
|
|
ce, length(steps), chunk_cache_stats()$memory_entries))
|
|
}
|
|
}
|
|
|
|
entry <- laurent2$l2_str(mat[[1, 3]])
|
|
mark_done(con, eq_id, entry)
|
|
cat(sprintf(" Done: %s\n",
|
|
if (nchar(entry) > 60) paste0(substr(entry, 1, 57), "...") else entry))
|
|
mat
|
|
}, error = function(e) {
|
|
mark_error(con, eq_id, e$message)
|
|
cat(sprintf(" ERROR [%s]: %s\n", eq_id, e$message))
|
|
NULL
|
|
})
|
|
|
|
result
|
|
}
|
|
|
|
#' @export
|
|
lkb_run <- function(con, max_jobs = Inf) {
|
|
cat(sprintf("\n=== LKB Worker (auto-detecting Bn from queue) ===\n"))
|
|
processed <- 0L
|
|
while (processed < max_jobs) {
|
|
result <- lkb_compute(con)
|
|
if (is.null(result)) {
|
|
remaining <- DBI::dbGetQuery(con,
|
|
"SELECT COUNT(*) AS n FROM lkb.compute_queue WHERE status = 'pending'")
|
|
if (remaining$n[[1]] == 0) break
|
|
Sys.sleep(2)
|
|
next
|
|
}
|
|
processed <- processed + 1L
|
|
}
|
|
cat(sprintf("\n Processed %d equations\n", processed))
|
|
}
|
|
|
|
#' @export
|
|
lkb_status <- function(con) {
|
|
q <- function(sql) DBI::dbGetQuery(con, sql)$n[[1]]
|
|
list(
|
|
pending = q("SELECT COUNT(*) AS n FROM lkb.compute_queue WHERE status = 'pending'"),
|
|
running = q("SELECT COUNT(*) AS n FROM lkb.compute_queue WHERE status = 'running'"),
|
|
done = q("SELECT COUNT(*) AS n FROM lkb.compute_queue WHERE status = 'done'"),
|
|
errors = q("SELECT COUNT(*) AS n FROM lkb.compute_queue WHERE status = 'error'"),
|
|
cached_chunks_db = q("SELECT COUNT(*) AS n FROM lkb.subseq_cache"),
|
|
cached_chunks_mem = chunk_cache_stats()$memory_entries
|
|
)
|
|
}
|
|
|
|
#' @export
|
|
reset_errors <- function(con) {
|
|
n <- DBI::dbExecute(con,
|
|
"UPDATE lkb.compute_queue SET status = 'pending', steps_done = 0, result_entry = NULL, started_at = NULL, finished_at = NULL WHERE status = 'error' OR status = 'running'")
|
|
cat(sprintf(" Reset %d jobs to pending\n", n))
|
|
DBI::dbExecute(con, "DELETE FROM lkb.checkpoint")
|
|
cat(" Cleared checkpoints\n")
|
|
}
|