1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
|
dir = function(reg, what) {
fs::path(fs::path_expand(reg$file.dir), what)
}
getResultFiles = function(reg, ids) {
fs::path(dir(reg, "results"), sprintf("%i.rds", if (is.atomic(ids)) ids else ids$job.id))
}
getLogFiles = function(reg, ids) {
job.hash = log.file = NULL
tab = reg$status[list(ids), c("job.id", "job.hash", "log.file")]
tab[is.na(log.file) & !is.na(job.hash), log.file := sprintf("%s.log", job.hash)]
tab[!is.na(log.file), log.file := fs::path(dir(reg, "logs"), log.file)]$log.file
}
getJobFiles = function(reg, hash) {
fs::path(reg$file.dir, "jobs", sprintf("%s.rds", hash))
}
getExternalDirs = function(reg, ids) {
fs::path(dir(reg, "external"), if (is.atomic(ids)) ids else ids$job.id)
}
mangle = function(x) {
sprintf("%s.rds", base32_encode(x, use.padding = FALSE))
}
unmangle = function(x) {
base32_decode(stri_sub(x, to = -5L), use.padding = FALSE)
}
file_remove = function(x) {
fs::file_delete(x[fs::file_exists(x)])
while(any(i <- fs::file_exists(x))) {
Sys.sleep(0.5)
fs::file_delete(x[i])
}
}
file_mtime = function(x) {
fs::file_info(x)$modification_time
}
writeRDS = function(object, file, compress = "gzip") {
file_remove(file)
saveRDS(object, file = file, version = 2L, compress = compress)
waitForFile(file, 300)
invisible(TRUE)
}
|
