# npRmpi attach-mode startup profile template
#
# Usage (batch/cluster):
#   1) copy this file to your working directory as .Rprofile, or set
#      R_PROFILE_USER to this file path.
#   2) launch with mpiexec/mpirun, e.g.
#      mpiexec -n 6 R CMD BATCH --no-save script.R
#
# Non-master ranks enter a receive/eval loop and execute commands sent by
# mpi.bcast.cmd(...). Rank 0 continues script execution.

.nonblock <- TRUE
.sleep <- 0.1
quiet <- FALSE
.recv.timeout <- suppressWarnings(as.numeric(Sys.getenv(
  "NP_RMPI_PROFILE_RECV_TIMEOUT_SEC", "0"
)))
if (!is.finite(.recv.timeout) || .recv.timeout <= 0)
  .recv.timeout <- 0

.profile_canon <- function(path) {
  if (!nzchar(path))
    return("")
  tryCatch(
    normalizePath(path.expand(path), winslash = "/", mustWork = FALSE),
    error = function(e) path.expand(path)
  )
}

.profile_user <- .profile_canon(Sys.getenv("R_PROFILE_USER", unset = ""))
.profile_site <- .profile_canon(Sys.getenv("R_PROFILE", unset = ""))

if (nzchar(.profile_user) && nzchar(.profile_site) &&
    identical(.profile_user, .profile_site)) {
  msg <- paste(
    "npRmpi profile startup misconfiguration:",
    "R_PROFILE_USER and R_PROFILE point to the same file.",
    "This can deadlock MPI profile/manual-broadcast startup.",
    "Remediation: keep only one profile source (prefer R_PROFILE_USER)",
    "and unset R_PROFILE for this launch."
  )
  base::cat(sprintf("\n[npRmpi profile guard] %s\n", msg), file = stderr())
  flush(stderr())
  stop(msg, call. = FALSE)
}

if (!invisible(library(npRmpi, logical.return = TRUE))) {
  warning("npRmpi cannot be loaded")
  q(save = "no")
}

# Profile/manual-broadcast launches keep automatic route selection available
# inside explicitly broadcast commands.
options(npRmpi.autodispatch = TRUE)
options(npRmpi.profile.active = TRUE)

# Guard against explicit Rmpi attachment inside profile scripts. Under the
# embedded npRmpi backend, attaching Rmpi then calling mpi.comm.rank() in this
# mode can crash the process; fail fast with remediation instead.
.npRmpi_profile_rmpi_attach_guard <- function(...) {
  if (!isTRUE(getOption("npRmpi.profile.active", FALSE)))
    return(invisible(NULL))
  msg <- paste(
    "npRmpi profile mode does not support attaching package 'Rmpi' in script body.",
    "Use npRmpi profile/manual-broadcast helpers (mpi.bcast.*) without library(Rmpi),",
    "or switch to attach/session mode if direct Rmpi APIs are required."
  )
  .stub <- function(...) stop(msg, call. = FALSE)
  assign("mpi.comm.rank", .stub, envir = .GlobalEnv)
  assign("mpi.comm.size", .stub, envir = .GlobalEnv)
  base::cat(sprintf("\n[npRmpi profile guard] %s\n", msg), file = stderr())
  flush(stderr())
  if (interactive()) {
    stop(msg, call. = FALSE)
  } else {
    quit(save = "no", status = 1, runLast = FALSE)
  }
}
setHook(packageEvent("Rmpi", "attach"), .npRmpi_profile_rmpi_attach_guard, action = "append")
if ("package:Rmpi" %in% search())
  .npRmpi_profile_rmpi_attach_guard()

# Ensure core base/recommended packages are attached for worker-side eval
# when launched via Rscript + .Rprofile.
for (pkg in c("utils", "stats", "datasets", "grDevices", "graphics", "methods")) {
  if (!(paste0("package:", pkg) %in% search())) {
    suppressPackageStartupMessages(library(pkg, character.only = TRUE))
  }
}

if (mpi.comm.size(0) > 1)
  invisible(mpi.comm.dup(0, 1))

# Profile/manual-broadcast scripts execute only on the rank 0 script body, so
# worker ranks need the embedded np MPI state primed before entering their
# receive/eval loop. Canonical scripts may still broadcast
# np.mpi.initialize() explicitly; repeated initialization is tolerated by the
# embedded backend.
if (mpi.comm.size(0) > 1)
  np.mpi.initialize()

# Own terminal errors only after the profile's communicator setup succeeds.
.npRmpi_ns <- asNamespace("npRmpi")
get(".npRmpi_profile_install_error_handler", envir = .npRmpi_ns,
    inherits = FALSE)()

if (mpi.comm.rank(0) > 0) {
  options(echo = FALSE)
  .comm <- 1
  .npRmpi_ns <- asNamespace("npRmpi")
  .npRmpi_worker_loop <- get(".npRmpi_worker_loop", envir = .npRmpi_ns, inherits = FALSE)
  mpi.barrier(0)
  tryCatch(
    .npRmpi_worker_loop(
      comm = .comm,
      nonblock = .nonblock,
      sleep = .sleep,
      recv.timeout = .recv.timeout,
      loop.label = "profile slave",
      timeout.remediation = paste(
        "check profile env wiring (avoid dual R_PROFILE_USER/R_PROFILE),",
        "verify FI_* interface settings, and retry with a fresh MPI launch."
      )
    ),
    error = get(".npRmpi_profile_abort_error", envir = .npRmpi_ns, inherits = FALSE)
  )
  mpi.comm.free(.comm)
  mpi.quit()
}

if (mpi.comm.rank(0) == 0) {
  mpi.barrier(0)
  if (mpi.comm.size(0) > 1)
    get(".npRmpi_worker_identity_admit", envir = .npRmpi_ns, inherits = FALSE)(1L)
  if (mpi.comm.size(0) > 1 && !quiet)
    slave.hostinfo(1)
}

.Last <- function() {
  finalize <- get(".npRmpi_finalize_before_unload",
                  envir = asNamespace("npRmpi"), inherits = FALSE)
  finalize("npRmpi profile .Last")
}
