Collect.R

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 ",
         "'startR_exec', as returned by Collect(..., wait = FALSE).")
  }
  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), 
           recursive = TRUE)
  }
  #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'))
  result
}