[[PageOutline]]
= Job Checkpointing Examples =

== Before you adopt checkpointing ==

'''Before you adopt a checkpointing workflow''', make a determination of whether your workflow consists - either entirely or in large part - of data elements or other components that can be processed '''entirely independently''' of each other - rather than each one depending on the previous. This is also referred to as '''embarrassingly parallel'''. (See also [https://www.freecodecamp.org/news/embarrassingly-parallel-algorithms-explained-with-examples/ Embarrassingly Parallel Algorithms Explained].)

If so, namely that your case is the former (embarrassingly parallel) in the above, then please first refer to the material in [wiki:Workshops/JobParallelism Job Parallelism].

== Quick start ==

For a quick start using checkpointing, see [wiki:Workshops/JobCheckpointing/Examples#WorkingExamples Working Examples] below.

== Background on checkpointing ==

=== Use checkpointing for sequential workflows ===

This checkpointing implementation should be used for running '''sequential workflows''' on Cypress - where each component result of the workflow depends the previous component result in the workflow. (See above [wiki:Workshops/JobCheckpointing/Examples#Beforeyouadoptcheckpointing Before you adopt checkpointing].)

=== Checkpointing - an HPC workflow design pattern ===

This checkpointing implementation can be considered an implementation of an HPC workflow design pattern (see for example [https://github.com/deepakkum21/Books/blob/master/Design%20Patterns%20-%20Elements%20of%20Reusable%20Object%20Oriented%20Software%20-%20GOF.pdf Design Patterns]) that can be described by any of the following expressions.

* submitter/runner pattern (Again see [https://github.com/deepakkum21/Books/blob/master/Design%20Patterns%20-%20Elements%20of%20Reusable%20Object%20Oriented%20Software%20-%20GOF.pdf Design Patterns].)
* job chaining (See also [https://gjbex.github.io/Workflows-for-HPC/ Workflow for HPC].)
* sequential workflows (See [https://docs.nersc.gov/jobs/best-practices/ NERSC - Best Practices for Jobs].)
* restart workflows (See [https://support.ceci-hpc.be/doc/SubmittingJobs/WorkflowManagement/#introduction Wide,Deep, & Cyclic Workflows] and [https://nlsq.readthedocs.io/en/latest/tutorials/routine/three_workflows/hpc_workflow.html Checkpointing implemented in Python].)
* checkpoint/restart job series (See [https://compendium.hpc.tu-dresden.de/jobs_and_resources/checkpoint_restart/ Checkpoint Motivation].)
* using job dependencies to restart jobs that checkpoint their state (See [https://github.com/hongthai-lenguyen/Slurm_HPC_Guide/blob/main/slurm_guide.md#52-creating-pipelines-with-job-dependencies Creating Pipelines with (SLURM) Job Dependencies].)

=== How does the workflow determine when to stop? ===

==== Limiting the work: using the sbatch --dependencies option ====

Note that in the context of the last reference in the above list - using job dependencies, when you use the SLURM '''sbatch --dependencies...''' option, the stopping point is not determined automatically.

You may need '''additional effort''' to determine the precise number of '''sbatch --dependencies...''' commands to exiecute - in other words, to determine the length of the dependency chain - before starting the workflow.

==== Limiting the work: using the application's own work limit ====

This implementation differs from the above dependency chain, in that rather than using the '''sbatch --dependencies''' option, the stopping point is determined automatically during runtime.

In this implementation, the workflow self restarts or resumes as needed and self terminates when the application's work limit is reached - in other words when the application's work is done. See for example the setting '''MAX_ITER''' in [wiki://Workshops/JobCheckpointing/Examples#checkpoint_submitter.sh checkpoint_submitter.sh] below.

== Checkpoint workflow implementation ==

=== Components explained ===

In this implementation, the checkpointing workflow consists of the following components.

1. '''the checkpoint submitter''' - to perform the following
 a. self restart as needed, and
 b. re-submit the runner once per self-restart
2. '''the checkpoint runner''' - to perform the following
 a. set its own application timer via the '''timeout''' command
 b. re-start the application as needed
 c. send the signal SIGTERM to the application at timeout
3. '''the checkpoint application''' - to perform the following
 a. compute
 b. catch the runner's signal SIGTERM
 c. "handle" the runner's signal SIGTERM by quickly saving the work and stop running (See [wiki:Workshops/JobCheckpointing/Examples#WorkingExamples Working Examples] for applications in BASH, Python and R.)

=== Checkpoint Submitter ===

Here is the job script for the '''checkpoint submitter''', a fully self-restarting job that controls the workflow by self-restarting as many times as needed and submitting one instance per each restart of the checkpoint runner job script. (See [wiki:Workshops/JobCheckpointing/Examples#CheckpointRunner Checkpoint Runner] further below.)

==== Overriding default settings ====

This checkpoint submitter uses settings - with default values - implemented using environment variables (see [wiki:cypress/WorkingOnClusters#EnvironmentVariables Environment Variables]), where the default values can be overridden by either of the following.

1. Override default settings by one or more '''export''' commands prior to the '''sbatch''' command
  
{{{
# example: export all on one line, then submit
[tulaneId@cypress1 ~]$export VAR1=<value1> VAR2=<value2> ...
[tulaneId@cypress1 ~]$sbatch checkpoint_submitter.sh
}}}

{{{
# example: or export one line for each variable, then submit
[tulaneId@cypress1 ~]$export VAR1=<value1>
[tulaneId@cypress1 ~]$export VAR2=<value2>
...
[tulaneId@cypress1 ~]$sbatch checkpoint_submitter.sh
}}}
 
2. Alternatively, override settings by assigning values at the beginning, in front of the '''sbatch''' command

{{{
# example: all on one line with submit
[tulaneId@cypress1 ~]$VAR1=<value1> VAR2=<value2> ... sbatch checkpoint_submitter.sh
}}}

{{{
# example: on multiple lines using the line continuation character "\" with submit
[tulaneId@cypress1 ~]$VAR1=<value1> \
VAR2=<value2> \
... \
sbatch checkpoint_submitter.sh
}}}

==== checkpoint_submitter.sh ====

{{{
#!/bin/bash
###############################################################################
# CHECKPOINT SUBMITTER: self-restarting to re-submit runner
# Controls repeated submissions of checkpoint_runner.sh
###############################################################################
# User settings with defaults
###############################################################################
# CKPT_PATH            : path to checkpoint file
# MAX_ITER             : stop application after this many iterations total
# RUNNER_SCRIPT        : job script with checkpoint before end of time limit
# SUBMITTER_MARGIN_SEC : time margin to restart before time limit
# SUBMITTER_TIME_LIMIT : wall time <= 24 hours (format as --time in man sbatch)
###############################################################################
# Edit settings below
# Or pass values via...
#   sbatch --export=ALL,Var1=...,...,VarN=... checkpoint_submitter.sh
# Or pass values via...
#   Var1=... VarN=... sbatch checkpoint_submitter.sh
###############################################################################
CKPT_PATH="${CKPT_PATH:-state_iter.txt}"
MAX_ITER="${MAX_ITER:-500}"
RUNNER_SCRIPT="${RUNNER_SCRIPT:-checkpoint_runner.sh}"
SUBMITTER_MARGIN_SEC="${SUBMITTER_MARGIN_SEC:-60}"
SUBMITTER_TIME_LIMIT="${SUBMITTER_TIME_LIMIT:-3:30}"
###############################################################################
# Slurm directives (keep SUBMITTER_TIME_LIMIT <= 24 hours)
###############################################################################
#SBATCH --job-name=ckpt_submitter
#SBATCH --output=log_submitter_%j.out
#SBATCH --error=log_submitter_%j.err
#SBATCH --time=24:00:00
#SBATCH --ntasks=1
#SBATCH --cpus-per-task=1
#SBATCH --mem=256M
#SBATCH --partition=centos7
#SBATCH --qos=normal
###############################################################################
# Exit on error
###############################################################################
set -euo pipefail
###############################################################################
# Helpers
###############################################################################
shopt -s expand_aliases
alias dtstamp='date +%Y%m%d-%H:%M:%S'
info(){ echo "Info[$(dtstamp)]: $*"; }
get_ckpt_iter() {
  [[ -f "${CKPT_PATH}" ]] || { echo 0; return; }
  tr -cd '0-9' < "${CKPT_PATH}" 2>/dev/null || echo 0
}
submit_worker() {
  sbatch "${RUNNER_SCRIPT}" | awk '{print $4}'
}
to_seconds() {
        # Acceptable time formats include "minutes", "minutes:seconds", "hours:min- utes:seconds",
        # "days-hours", "days-hours:minutes" and "days-hours:minutes:seconds"
        local input="$1"
        # Handle days-hours format by replacing '-' with ':' to unify the delimiter
        # Then split components into an array
        IFS=':' read -r -a parts <<< "${input//-/:}"
        case ${#parts[@]} in
            1) # Format: minutes
                seconds=$((parts[0] * 60))
                ;;
            2) # Formats: minutes:seconds OR days-hours
                if [[ $input == *"-"* ]]; then
                    seconds=$((parts[0] * 86400 + parts[1] * 3600))
                else
                    seconds=$((parts[0] * 60 + parts[1]))
                fi
                ;;
            3) # Formats: hours:minutes:seconds OR days-hours:minutes
                if [[ $input == *"-"* ]]; then
                    seconds=$((parts[0] * 86400 + parts[1] * 3600 + parts[2] * 60))
                else
                    seconds=$((parts[0] * 3600 + parts[1] * 60 + parts[2]))
                fi
                ;;
            4) # Format: days-hours:minutes:seconds
                seconds=$((parts[0] * 86400 + parts[1] * 3600 + parts[2] * 60 + parts[3]))
                ;;
            *)
                echo "Unsupported format"
                exit 1
                ;;
        esac
        echo "$seconds"
}
should_restart() {
  now=$(date +%s)
  elapsed=$(( now - restart_start_sec ))
  remaining=$(( restart_limit_sec - elapsed ))
  echo -ne "Time before restart: $remaining (sec)\r" >&2
  (( remaining < SUBMITTER_MARGIN_SEC ))
}
###############################################################################
# Log settings
###############################################################################
info "Start on $(hostname); JOBID=${SLURM_JOB_ID:-unknown}"
info "Submitter settings:"
info "CKPT_PATH=${CKPT_PATH}"
info "MAX_ITER=${MAX_ITER}"
info "RUNNER_SCRIPT=${RUNNER_SCRIPT}"
info "SUBMITTER_MARGIN_SEC=${SUBMITTER_MARGIN_SEC}"
info "SUBMITTER_TIME_LIMIT=${SUBMITTER_TIME_LIMIT}"
###############################################################################
# Restore state if present
###############################################################################
state_file="submitter_state.txt"
submitter_cycle=0
if [[ -f "${state_file}" ]]; then
  source "${state_file}"
  info "Loaded state: submitter_cycle=${submitter_cycle}"
else
  info "No previous state found; starting fresh."
fi
###############################################################################
# Track time for restart
###############################################################################
restart_start_sec=$(date +%s)
restart_limit_sec=$(to_seconds "${SUBMITTER_TIME_LIMIT}")
###############################################################################
# Main Loop
###############################################################################
info "Submitter cycle: ${submitter_cycle}"
module load anaconda3/2023.07 # using python3
while true; do
  current_iter=$(get_ckpt_iter)
  info "Current checkpoint iteration: ${current_iter}"
  if (( current_iter >= MAX_ITER )); then
          info "Reached MAX_ITER=${MAX_ITER}; running final analytics."
    # List of analytics scripts to check
    analytics_scripts=(
      "aggregate_usage.py"
      "plot_progress.py"
    )
          # Run resource usage aggregation
          if command -v python3 >/dev/null 2>&1; then
      # Loop over each script and run only if present
      for script in "${analytics_scripts[@]}"; do
        if [[ -f "${script}" ]]; then
          info "Running analytics ${script} ..."
          python3 "${script}"
        else
          info "WARNING: ${script} not found; skipping."
        fi
      done
          else
            info "WARNING: python3 not found; skipping analytics."
          fi
          info "Workflow complete. Exiting submitter."
    exit 0

  fi
  JOBID=$(submit_worker)
  info "Submitted worker: ${JOBID}"
        # Log worker JobIDs for later aggregation
        id_log="worker_ids.txt"
        echo "${JOBID}" >> "${id_log}"
  # Wait for worker to finish
  info "Waiting on first of worker ${JOBID} to finish or self-restart."
  while squeue -j "${JOBID}" >/dev/null 2>&1; do
    sleep 10
    if should_restart; then
      info "Submitter approaching timeout -> restarting self"
      # Save state
      echo "submitter_cycle=$((submitter_cycle+1))" > "${state_file}"
      # Submit a new submitter instance
      sbatch checkpoint_submitter.sh
      exit 0

    fi
  done
  # Read worker exit code
  rc=$(sacct -j "${JOBID}" --format=ExitCode -P | awk -F: 'NR==2{print $1}')
  info "Worker finished with code ${rc}"
  case "${rc}" in
    0)
      info "Worker reports completed workflow."
      exit 0

      ;;
    99|124)
      info "Checkpoint advance -> continuing workflow."
      ;;
    *)
      info "Unexpected worker failure (code ${rc}). Stopping."
      exit "${rc}"

      ;;
  esac
  # Loop continues to submit next worker
done
}}}

=== Checkpoint Runner ===

Here is the job script for the '''checkpoint runner''' to be used with the following.

* for jobs submitted by [wiki:Workshops/JobCheckpointing/Examples#CheckpointSubmitter Checkpoint Submitter] above
* for running any of the following, minimal working examples of checkpointed applications in BASH, Python, or R.
 * For BASH, see [wiki:Workshops/JobCheckpointing/Examples#BASHCheckpointingExample BASH Checkpointint Example]
 * For Python, see [wiki:Workshops/JobCheckpointing/Examples#PythonCheckpointingExample Python Checkpointint Example]
 * For R, see [wiki:Workshops/JobCheckpointing/Examples#RCheckpointingExample R Checkpointint Example]

==== checkpoint_runner.sh ====

{{{
#!/bin/bash
###############################################################################
# CHECKPOINT RUNNER: run application w/ checkpoint + controlled stop
# Use with checkpoint_submitter.sh to re-submit as needed
###############################################################################
# User settings with defaults
###############################################################################
# APP_CMD          : command to run (string) - must handle signal SIGTERM
# CHECKPOINT_EVERY : checkpoint after every this many application iterations
# CKPT_PATH        : path to checkpoint file
# LAUNCH_MODE      : run app command directly or via srun
# MAX_ITER         : stop application after this many iterations total
# MODULE_LIST      : modules loaded before running app command
# RUNNER_MARGIN_SEC: seconds before wall time to checkpoint (for timeout)
# RUNNER_TIME_LIMIT: wall time <= 24 hours (format as --time in man sbatch)
# SRUN_ARGS        : arguments to srun, if needed
###############################################################################
# Edit settings below
# Or pass values via...
#   sbatch --export=ALL,Var1=...,...,VarN=... checkpoint_submitter.sh
# Or pass values via...
#   Var1=... VarN=... sbatch checkpoint_submitter.sh
###############################################################################
APP_CMD="${APP_CMD:-python3 checkpoint_signal_iter.py}"
CHECKPOINT_EVERY="${CHECKPOINT_EVERY:-20}"
CKPT_PATH="${CKPT_PATH:-state_iter.txt}"
LAUNCH_MODE="${LAUNCH_MODE:-direct}"
MAX_ITER="${MAX_ITER:-500}"
MODULE_LIST="${MODULE_LIST:-anaconda3/2023.07}"
RUNNER_MARGIN_SEC="${RUNNER_MARGIN_SEC:-60}"
RUNNER_TIME_LIMIT="${RUNNER_TIME_LIMIT:-00:03:00}"
SRUN_ARGS="${SRUN_ARGS:--n 1}"
###############################################################################
# Slurm directives (keep RUNNER_TIME_LIMIT <= 24 hours)
###############################################################################
#SBATCH --job-name=ckpt_runner_demo
#SBATCH --output=log_runner_%j.out
#SBATCH --error=log_runner_%j.err
#SBATCH --time=00:03:00
#SBATCH --ntasks=1
#SBATCH --cpus-per-task=1
#SBATCH --mem=512M
#SBATCH --partition=centos7
#SBATCH --qos=normal
###############################################################################
# Exit on error
###############################################################################
set -euo pipefail
###############################################################################
# Helpers
###############################################################################
shopt -s expand_aliases
alias dtstamp="date +%Y%m%d-%H:%M:%S"
info(){ echo "Info[$(dtstamp)]: $*"; }
get_ckpt_iter() {
  [[ -f "${CKPT_PATH}" ]] || { echo 0; return; }
  tr -cd '0-9' < "${CKPT_PATH}" 2>/dev/null || echo 0
}
to_seconds() {
        # Acceptable time formats include "minutes", "minutes:seconds", "hours:min- utes:seconds",
        # "days-hours", "days-hours:minutes" and "days-hours:minutes:seconds"
        local input="$1"
        # Handle days-hours format by replacing '-' with ':' to unify the delimiter
        # Then split components into an array
        IFS=':' read -r -a parts <<< "${input//-/:}"
        case ${#parts[@]} in
            1) # Format: minutes
                seconds=$((parts[0] * 60))
                ;;
            2) # Formats: minutes:seconds OR days-hours
                if [[ $input == *"-"* ]]; then
                    seconds=$((parts[0] * 86400 + parts[1] * 3600))
                else
                    seconds=$((parts[0] * 60 + parts[1]))
                fi
                ;;
            3) # Formats: hours:minutes:seconds OR days-hours:minutes
                if [[ $input == *"-"* ]]; then
                    seconds=$((parts[0] * 86400 + parts[1] * 3600 + parts[2] * 60))
                else
                    seconds=$((parts[0] * 3600 + parts[1] * 60 + parts[2]))
                fi
                ;;
            4) # Format: days-hours:minutes:seconds
                seconds=$((parts[0] * 86400 + parts[1] * 3600 + parts[2] * 60 + parts[3]))
                ;;
            *)
                echo "Unsupported format"
                exit 1
                ;;
        esac
        echo "$seconds"
}
###############################################################################
# Log settings
###############################################################################
info "Start on $(hostname); JOBID=${SLURM_JOB_ID:-unknown}"
info "Runner settings:"
info "APP_CMD=${APP_CMD}"
info "CHECKPOINT_EVERY=${CHECKPOINT_EVERY}"
info "CKPT_PATH=${CKPT_PATH}"
info "LAUNCH_MODE=${LAUNCH_MODE}"
info "MAX_ITER=${MAX_ITER}"
info "MODULE_LIST=${MODULE_LIST}"
info "RUNNER_MARGIN_SEC=${RUNNER_MARGIN_SEC}"
info "RUNNER_TIME_LIMIT=${RUNNER_TIME_LIMIT}"
info "SRUN_ARGS=${SRUN_ARGS}"
###############################################################################
# Load requested modules
###############################################################################
info "Loading modules."
module load "${MODULE_LIST}"
###############################################################################
# Timeout window
###############################################################################
TOTAL_SEC=$(to_seconds "${RUNNER_TIME_LIMIT}")
RUN_WINDOW_SEC=$(( TOTAL_SEC - RUNNER_MARGIN_SEC ))
if (( RUN_WINDOW_SEC <= 0 )); then
  info "WARNING: RUN_WINDOW_SEC <= 0; using 1s."
  RUN_WINDOW_SEC=1
fi
before_iter=$(get_ckpt_iter)
set +e
if [[ "${LAUNCH_MODE}" == "srun" ]]; then
  # timeout -> srun -> bash -lc "APP_CMD"
  # On expiry, timeout sends TERM to srun; srun forwards signals to its job step tasks.
  timeout "${RUN_WINDOW_SEC}s" srun ${SRUN_ARGS} bash -lc "${APP_CMD}"
else
  # direct mode: run in the batch shell
  timeout "${RUN_WINDOW_SEC}s" bash -lc "${APP_CMD}"
fi
rc=$?
set -e
info "Program exit code (from timeout wrapper): ${rc}"
###############################################################################
# Interpret exit-code from timeout
#   0   = completed all iterations (no further submissions necessary)
#   99  = app voluntarily checkpointed early -> submitter should continue
#   124 = timeout before walltime expiry -> checkpoint likely written -> continue
###############################################################################
if [[ ${rc} -eq 0 ]]; then
  info "Application finished all iterations."
  exit 0
elif [[ ${rc} -eq 99 ]]; then
  info "Application exited after writing checkpoint."
  exit 99
elif [[ ${rc} -eq 124 ]]; then
  after_iter=$(get_ckpt_iter)
  if (( after_iter > before_iter )); then
    info "Timeout + checkpoint advance detected (${before_iter}->${after_iter})."
    exit 124
  else
    info "Timeout but checkpoint did not advance. Marking failure."
    exit 1
  fi
else
  info "Unexpected exit code: ${rc}"
  exit "${rc}"
fi
}}}

== Working Examples ==

=== Make copies of checkpoint_submitter.sh and checkpoint_runner.sh

All of the following examples make use of the two job scripts mentioned above.

* the [wiki://Workshops/JobCheckpointing/Examples#checkpoint_submitter.sh checkpoint_submitter.sh] and
* the [wiki://Workshops/JobCheckpointing/Examples#checkpoint_runner.sh checkpoint_runner.sh] 

=== BASH Checkpointing Example ===

See [wiki:Workshops/JobCheckpointing/Examples/BASH BASH Checkpointing Example]

=== Python Checkpointing Example ===

See [wiki:Workshops/JobCheckpointing/Examples/Python Python Checkpointing Example]

=== R Checkpointing Example ===

See [wiki:Workshops/JobCheckpointing/Examples/R R Checkpointing Example]