[[PageOutline]] = Python Checkpointing Example = == Python Checkpointing Application == Here is an example of a checkpointing application implemented in Python script. By default, the Python checkpointing application behaves as follows. * checkpointing every 20 application iterations (one second per iteration) and * running a total of 500 iterations. == Running the Python checkpointing example with defaults == To run the application on Cypress, perform the following. 1. Create your own versions of the following files in your current directory on Cypress. (For file editing with '''nano''' on Cypress, see [[https://wiki.hpc.tulane.edu/trac/wiki/cypress/FileEditingSoftware/Example|File Editing Example]].) * [wiki://Workshops/JobCheckpointing/Examples#checkpoint_submitter.sh checkpoint_submitter.sh] * [wiki://Workshops/JobCheckpointing/Examples#checkpoint_runner.sh checkpoint_runner.sh] * [wiki://Workshops/JobCheckpointing/Examples#aggregate_usage.py aggregate_usage.py] * [wiki://Workshops/JobCheckpointing/Examples#plot_progress.py plot_progress.py] 2. Create your own version of the following Python application script file '''checkpoint_signal_iter.py'''. {{{ #!/usr/bin/env python3 import os, time, sys, signal, tempfile CKPT = os.environ.get("CKPT_PATH", "state_iter.json") EVERY = int(os.environ.get("CHECKPOINT_EVERY", "20")) MAX_ITER = int(os.environ.get("MAX_ITER", "500")) _current = None def atomic_save(path: str, data: bytes): # atomic write (tmp + fsync + replace) dirpath = os.path.dirname(os.path.abspath(path)) or "." fd, tmppath = tempfile.mkstemp(prefix=".ckpt.", dir=dirpath) try: with os.fdopen(fd, "wb") as f: f.write(data); f.flush(); os.fsync(f.fileno()) os.replace(tmppath, path) finally: try: if os.path.exists(tmppath): os.remove(tmppath) except Exception: pass def save(i: int): atomic_save(CKPT, f"{i}".encode("utf-8")) def load() -> int: try: with open(CKPT, "rb") as f: return int(f.read().decode("utf-8").strip()) except Exception: return 0 def on_sigterm(signum, frame): i = _current if _current is not None else load() print(f"SIGTERM: saving i={i} and exiting 99", flush=True) save(i) sys.exit(99) signal.signal(signal.SIGTERM, on_sigterm) def main(): global _current i = load() print(f"Resuming from i={i} (iteration-based every {EVERY})", flush=True) while True: i += 1 _current = i time.sleep(1) if i % EVERY == 0: save(i) print(f"[periodic/iter] saved i={i}", flush=True) if i > MAX_ITER: print(f"Reached i={i} > {MAX_ITER}; exiting 0", flush=True) save(i) sys.exit(0) if __name__ == "__main__": main() }}} 3. Submit the job via the following command. {{{ [tulaneID@cypress1 ~]$ CKPT_PATH=state_iter_py.txt sbatch checkpoint_runner.sh }}} 4. You can monitor job outputs of both the checkpoint_submitter.sh and the checkpoint_runner.sh jobs as they are updated by executing the following '''tail''' command using the '''-F''' option to retry once per second, then using Ctrl+C to exit the retry process entirely. {{{ [tulaneID@cypress1 ~]$ tail -F log* }}} == Overriding the default settings == To customize your workflow, see [wiki://Workshops/JobCheckpointing/Examples#Overridingdefaultsettings Overriding default settings] for overriding the default settings in the checkpoint submitter according to your needs. == Sample run outputs == === List of output files === Here is a list of all output files from a sample run sorted by time, latest-to-earliest {{{ [tulaneID@cypress2 ~]$ls -lt *.csv *.err *.out *.png *.txt -rw-r--r-- 1 tulaneID groupID 1404 Apr 1 16:14 log_submitter_3303041.out -rw-r--r-- 1 tulaneID groupID 480 Apr 1 16:14 anomaly_details.csv -rw-r--r-- 1 tulaneID groupID 54778 Apr 1 16:14 anomaly_stats_table.png -rw-r--r-- 1 tulaneID groupID 44362 Apr 1 16:14 resource_trend.png -rw-r--r-- 1 tulaneID groupID 757 Apr 1 16:14 log_submitter_3303041.err -rw-r--r-- 1 tulaneID groupID 48683 Apr 1 16:14 progress_with_anomalies.png -rw-r--r-- 1 tulaneID groupID 509 Apr 1 16:14 worker_usage.csv -rw-r--r-- 1 tulaneID groupID 727 Apr 1 16:14 log_submitter_3303039.out -rw-r--r-- 1 tulaneID groupID 18 Apr 1 16:14 submitter_state.txt -rw-r--r-- 1 tulaneID groupID 824 Apr 1 16:12 log_runner_3303040.out -rw------- 1 tulaneID groupID 3 Apr 1 16:12 state_iter_py.txt -rw-r--r-- 1 tulaneID groupID 0 Apr 1 16:11 log_runner_3303040.err -rw-r--r-- 1 tulaneID groupID 727 Apr 1 16:11 log_submitter_3303037.out -rw-r--r-- 1 tulaneID groupID 0 Apr 1 16:11 log_submitter_3303039.err -rw-r--r-- 1 tulaneID groupID 40 Apr 1 16:11 worker_ids.txt -rw-r--r-- 1 tulaneID groupID 957 Apr 1 16:11 log_runner_3303038.out -rw-r--r-- 1 tulaneID groupID 0 Apr 1 16:09 log_runner_3303038.err -rw-r--r-- 1 tulaneID groupID 727 Apr 1 16:09 log_submitter_3303035.out -rw-r--r-- 1 tulaneID groupID 0 Apr 1 16:09 log_submitter_3303037.err -rw-r--r-- 1 tulaneID groupID 957 Apr 1 16:08 log_runner_3303036.out -rw-r--r-- 1 tulaneID groupID 0 Apr 1 16:06 log_runner_3303036.err -rw-r--r-- 1 tulaneID groupID 727 Apr 1 16:06 log_submitter_3303033.out -rw-r--r-- 1 tulaneID groupID 0 Apr 1 16:06 log_submitter_3303035.err -rw-r--r-- 1 tulaneID groupID 957 Apr 1 16:05 log_runner_3303034.out -rw-r--r-- 1 tulaneID groupID 0 Apr 1 16:03 log_runner_3303034.err -rw-r--r-- 1 tulaneID groupID 734 Apr 1 16:03 log_submitter_3303031.out -rw-r--r-- 1 tulaneID groupID 0 Apr 1 16:03 log_submitter_3303033.err -rw-r--r-- 1 tulaneID groupID 949 Apr 1 16:03 log_runner_3303032.out -rw-r--r-- 1 tulaneID groupID 0 Apr 1 16:01 log_runner_3303032.err -rw-r--r-- 1 tulaneID groupID 0 Apr 1 16:01 log_submitter_3303031.err }}} === Sample checkpoint submitter logs === Here are the first and last instances from among the multiple log files generated by the sequence of jobs running the job script [wiki://Workshops/JobCheckpointing/Examples#checkpoint_submitter.sh checkpoint_submitter.sh]. The first job in the sequence was submitted by the user, and all other jobs thereafter in that sequence were submitted by the job script itself via '''self-restarting'''. ==== Listing of the initial checkpoint submitter log - starting workflow ==== Here is the listing of the first log file, '''log_submitter_3303031.out''', generated by [wiki://Workshops/JobCheckpointing/Examples#checkpoint_submitter.sh checkpoint_submitter.sh] ending with the message "Submitter approaching timeout -> restarting self. Submitted batch job 3303033." {{{ [tulaneID@cypress2 ~]$cat log_submitter_3303031.out Info[20260401-16:01:12]: Start on cypress01-066; JOBID=3303031 Info[20260401-16:01:12]: Submitter settings: Info[20260401-16:01:12]: CKPT_PATH=state_iter_py.txt Info[20260401-16:01:12]: MAX_ITER=500 Info[20260401-16:01:12]: RUNNER_SCRIPT=checkpoint_runner.sh Info[20260401-16:01:12]: SUBMITTER_MARGIN_SEC=60 Info[20260401-16:01:12]: SUBMITTER_TIME_LIMIT=00:03:30 Info[20260401-16:01:12]: No previous state found; starting fresh. Info[20260401-16:01:12]: Submitter cycle: 0 Info[20260401-16:01:12]: Current checkpoint iteration: 0 Info[20260401-16:01:12]: Submitted worker: 3303032 Info[20260401-16:01:12]: Waiting on worker 3303032 Info[20260401-16:03:52]: Submitter approaching timeout -> restarting self Submitted batch job 3303033 }}} ==== Listing of the final checkpoint submitter log - completed workflow ==== Here is the listing of the final log file, '''log_submitter_3303041.out''', generated by [wiki://Workshops/JobCheckpointing/Examples#checkpoint_submitter.sh checkpoint_submitter.sh] ending with the message "Workflow complete. Exiting submitter." {{{ [tulaneID@cypress2 ~]$cat log_submitter_3303041.out Info[20260401-16:14:35]: Start on cypress01-066; JOBID=3303041 Info[20260401-16:14:35]: Submitter settings: Info[20260401-16:14:35]: CKPT_PATH=state_iter_py.txt Info[20260401-16:14:35]: MAX_ITER=500 Info[20260401-16:14:35]: RUNNER_SCRIPT=checkpoint_runner.sh Info[20260401-16:14:35]: SUBMITTER_MARGIN_SEC=60 Info[20260401-16:14:35]: SUBMITTER_TIME_LIMIT=00:03:30 Info[20260401-16:14:35]: Loaded state: submitter_cycle=5 Info[20260401-16:14:35]: Submitter cycle: 5 Info[20260401-16:14:35]: Current checkpoint iteration: 501 Info[20260401-16:14:35]: Reached MAX_ITER=500; running final analytics. Info[20260401-16:14:35]: Running analytics aggregate_usage.py ... Wrote worker_usage.csv JobID ElapsedSeconds TotalCPU ... ReqMem NodeList State 0 3303032 121 00:00.389 ... 512Mn cypress01-066 FAILED 1 3303034 120 00:00.388 ... 512Mn cypress01-066 FAILED 2 3303036 121 00:00.389 ... 512Mn cypress01-066 FAILED 3 3303038 120 00:00.391 ... 512Mn cypress01-066 FAILED 4 3303040 22 00:00.374 ... 512Mn cypress01-066 COMPLETED [5 rows x 9 columns] Info[20260401-16:14:36]: Running analytics plot_progress.py ... Saved progress_with_anomalies.png Saved resource_trend.png Saved anomaly_details.csv Saved anomaly_stats_table.png Info[20260401-16:14:41]: Workflow complete. Exiting submitter. }}} === Sample checkpoint runner logs === Here are the first and last output logs of the sequence of jobs generated by the job script [wiki://Workshops/JobCheckpointing/Examples#checkpoint_runner.sh checkpoint_runner.sh]. All of the jobs in that sequence were submitted by the [wiki://Workshops/JobCheckpointing/Examples#checkpoint_submitter.sh checkpoint_submitter.sh] job script. ==== Listing of the initial checkpoint runner log - starting application ==== Here is the listing of the intial log file, '''log_runner_3303032.out''', generated by the job script [wiki://Workshops/JobCheckpointing/Examples#checkpoint_runner.sh checkpoint_runner.sh] ending with message "Timeout + checkpoint advance detected (0->120)." {{{ [tulaneID@cypress2 ~]$cat log_runner_3303032.out Info[20260401-16:01:13]: Start on cypress01-066; JOBID=3303032 Info[20260401-16:01:13]: Runner settings: Info[20260401-16:01:13]: APP_CMD=python3 checkpoint_signal_iter.py Info[20260401-16:01:13]: CHECKPOINT_EVERY=20 Info[20260401-16:01:13]: CKPT_PATH=state_iter_py.txt Info[20260401-16:01:13]: LAUNCH_MODE=direct Info[20260401-16:01:13]: MAX_ITER=500 Info[20260401-16:01:13]: MODULE_LIST=anaconda3/2023.07 Info[20260401-16:01:13]: RUNNER_MARGIN_SEC=60 Info[20260401-16:01:13]: RUNNER_TIME_LIMIT=00:03:00 Info[20260401-16:01:13]: SRUN_ARGS=-n 1 Info[20260401-16:01:13]: Loading modules. Resuming from i=0 (iteration-based every 20) [periodic/iter] saved i=20 [periodic/iter] saved i=40 [periodic/iter] saved i=60 [periodic/iter] saved i=80 [periodic/iter] saved i=100 SIGTERM: saving i=120 and exiting 99 Info[20260401-16:03:13]: Program exit code (from timeout wrapper): 124 Info[20260401-16:03:13]: Timeout + checkpoint advance detected (0->120). }}} ==== Listing of the final checkpoint runner log - completed application ==== Here is the listing of the final log file, '''log_runner_3303040.out''', generated by [wiki://Workshops/JobCheckpointing/Examples#checkpoint_runner.sh checkpoint_runner.sh] ending with message "Application finished all iterations." {{{ [tulaneID@cypress2 ~]$cat log_runner_3303040.out Info[20260401-16:11:54]: Start on cypress01-066; JOBID=3303040 Info[20260401-16:11:54]: Runner settings: Info[20260401-16:11:54]: APP_CMD=python3 checkpoint_signal_iter.py Info[20260401-16:11:54]: CHECKPOINT_EVERY=20 Info[20260401-16:11:54]: CKPT_PATH=state_iter_py.txt Info[20260401-16:11:54]: LAUNCH_MODE=direct Info[20260401-16:11:54]: MAX_ITER=500 Info[20260401-16:11:54]: MODULE_LIST=anaconda3/2023.07 Info[20260401-16:11:54]: RUNNER_MARGIN_SEC=60 Info[20260401-16:11:54]: RUNNER_TIME_LIMIT=00:03:00 Info[20260401-16:11:55]: SRUN_ARGS=-n 1 Info[20260401-16:11:55]: Loading modules. Resuming from i=480 (iteration-based every 20) [periodic/iter] saved i=500 Reached i=501 > 500; exiting 0 Info[20260401-16:12:16]: Program exit code (from timeout wrapper): 0 Info[20260401-16:12:16]: Application finished all iterations. }}} === Listings of post-workflow analysis output files === The following is the complete, annotated listing of all the generated .csv and .txt files {{{ [tulaneID@cypress2 ~]$tail -n +1 *.csv *.txt ==> anomaly_details.csv <== JobID,StartTime,ElapsedSeconds,MaxRSS_MiB,TimeZ,MemModZ,AnomalyType 3303032,2026-04-01 16:01:13,121.0,14.88494873046875,0.5126573320092711,0.0,none 3303034,2026-04-01 16:03:53,120.0,14.8773193359375,0.487278256167228,0.0,none 3303036,2026-04-01 16:06:33,121.0,14.8773193359375,0.5126573320092711,0.0,none 3303038,2026-04-01 16:09:14,120.0,14.881134033203123,0.487278256167228,0.0,none 3303040,2026-04-01 16:11:54,22.0,13.06915283203125,-1.999871176352998,639.4260000005955,memory ==> worker_usage.csv <== JobID,ElapsedSeconds,TotalCPU,MaxRSS,MaxRSS_bytes,MaxRSS_MiB,ReqMem,NodeList,State 3303032,121,00:00.389,15608K,15608000.0,14.88494873046875,512Mn,cypress01-066,FAILED 3303034,120,00:00.388,15600K,15600000.0,14.8773193359375,512Mn,cypress01-066,FAILED 3303036,121,00:00.389,15600K,15600000.0,14.8773193359375,512Mn,cypress01-066,FAILED 3303038,120,00:00.391,15604K,15604000.0,14.881134033203125,512Mn,cypress01-066,FAILED 3303040,22,00:00.374,13704K,13704000.0,13.06915283203125,512Mn,cypress01-066,COMPLETED ==> state_iter_py.txt <== 501 ==> submitter_state.txt <== submitter_cycle=5 ==> worker_ids.txt <== 3303032 3303034 3303036 3303038 3303040 }}} === Post-workflow analysis plots === ==== Background ==== For background on the following plots, see [wiki://Workshops/JobCheckpointing/Examples#Plottingexecutiontimeandmemoryusage Plotting execution time and memory usage]. ==== Table of anomalies, compute time and memory usage ==== Here is the plot, '''anomaly_stats_table.png''', showing the table of anomalies, if any, from among all of the [wiki://Workshops/JobCheckpointing/Examples#checkpoint_runner.sh checkpoint_runner.sh] jobs. [[Image(python_anomaly_stats_table.png, width=800px)]] ==== Compute progress ==== Here is the plot, '''progress_with_anomalies.png''', showing '''Checkpoint Progress''' and '''MaxRSS''' (SLURM max job memory usage) vs '''Wall Time'''. [[Image(python_progress_with_anomalies.png, width=800px)]] ==== Execution time and memory usage trends ==== Here is the plot, '''resource_trend.png''', showing '''Elapsed (time)''' and '''MaxRSS''' vs. '''Worker job index'''. [[Image(python_resource_trend.png, width=800px)]]