Newer
Older
Collect <- function(startr_exec, wait = TRUE, remove = TRUE) {
if (!('startR_exec' %in% class(startr_exec))) {
stop("Parameter 'startr_exec' must be an object of the class ",
}
cluster <- startr_exec[['cluster']]
ecflow_host <- startr_exec[['ecflow_host']]
suite_id <- startr_exec[['suite_id']]
ecflow_output_dir <- startr_exec[['ecflow_output_dir']]
done <- FALSE
attempt <- 1
while (!done) {
Sys.sleep(min(sqrt(attempt), 5))
status <- system(paste0("ecflow_client --get_state=STARTR_CHUNKING_",
suite_id, " --host=",
ecflow_host[['name']], " --port=", ecflow_host[['port']]),
intern = TRUE)
if (any(grepl(paste0("suite STARTR_CHUNKING_", suite_id, " #.* state:complete"), status))) {
done <- TRUE
} else if (!wait) {
stop("Computation in progress...")
}
attempt <- attempt + 1
}
if (remove) {
.warning("ATTENTION: The source chunks will be removed from the ",
"system. Store the result after Collect() ends if needed.")
}
result <- startR:::.MergeChunks(ecflow_output_dir, suite_id, remove)
if (remove) {
system(paste0("ecflow_client --delete=force yes /STARTR_CHUNKING_",
suite_id, " --host=", ecflow_host[['name']],
" --port=", ecflow_host[['port']]))
unlink(paste0(ecflow_output_dir, '/STARTR_CHUNKING_', suite_id),
#system("ecflow_client --shutdown --port=5678")
#system("ecflow_stop.sh -p 5678")
#result <- readRDS(paste0(ecflow_output_dir, '/result.Rds'))
#file.remove(paste0(ecflow_output_dir, '/result.Rds'))