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 ",
ecflow_server <- startr_exec[['ecflow_server']]
ecflow_suite_dir <- startr_exec[['ecflow_suite_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_server[['host']], " --port=", ecflow_server[['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 (!is.null(cluster[['temp_dir']])) {
ecflow_suite_dir_suite <- paste0(cluster[['temp_dir']], '/STARTR_CHUNKING_', suite_id)
system(paste0('ssh ', cluster[['queue_host']], ' "rm -rf ',
ecflow_suite_dir_suite, '"'))
}
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_suite_dir, suite_id, remove)
if (remove) {
system(paste0("ecflow_client --delete=force yes /STARTR_CHUNKING_",
suite_id, " --host=", ecflow_server[['host']],
" --port=", ecflow_server[['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'))