diff --git a/pom.xml b/pom.xml index 5958be2be3..6ec2c8aca6 100644 --- a/pom.xml +++ b/pom.xml @@ -192,9 +192,9 @@ v0.0.44-dev4 v0.0.40 v0.0.44-dev4 - 1.4.12 - 1.4.12 - 1.4.12 + 1.4.13 + 1.4.13 + 1.4.13 diff --git a/vcell-core/src/main/java/cbit/vcell/resource/PropertyLoader.java b/vcell-core/src/main/java/cbit/vcell/resource/PropertyLoader.java index ee375dc2d6..0672f82b59 100644 --- a/vcell-core/src/main/java/cbit/vcell/resource/PropertyLoader.java +++ b/vcell-core/src/main/java/cbit/vcell/resource/PropertyLoader.java @@ -135,7 +135,8 @@ static Set declaredPropertyNames() { public static final String slurm_langevin_timeoutPerTaskSeconds = record("vcell.slurm.langevin.timeoutPerTaskSeconds",ValueType.GEN); public static final String slurm_langevin_batchMemoryLimitPerTaskMB = record("vcell.slurm.langevin.batchMemoryLimitPerTaskMB",ValueType.GEN); public static final String slurm_langevin_memoryBlockSizeMB = record("vcell.slurm.langevin.memoryBlockSizeMB",ValueType.GEN); - + public static final String slurm_langevin_watchdogTimeoutSeconds = record("vcell.slurm.langevin.watchdogTimeoutSeconds",ValueType.GEN); + public static final String slurm_langevin_watchdogTickSeconds = record("vcell.slurm.langevin.watchdogTickSeconds",ValueType.GEN); // public static final String finiteVolumeExecutableProperty = record("vcell.finitevolume.executable",ValueType.EXE); // diff --git a/vcell-server/src/main/java/cbit/vcell/message/server/htc/slurm/SlurmProxy.java b/vcell-server/src/main/java/cbit/vcell/message/server/htc/slurm/SlurmProxy.java index 709867ec41..f4bd18d0eb 100644 --- a/vcell-server/src/main/java/cbit/vcell/message/server/htc/slurm/SlurmProxy.java +++ b/vcell-server/src/main/java/cbit/vcell/message/server/htc/slurm/SlurmProxy.java @@ -642,9 +642,10 @@ public static String computeSlurmTimeLimit(int totalNumberOfJobs, return String.format("%d-%02d:%02d:00", days, hours, minutes); } } - private void writeScriptControlledVariables(LineStringBuilder lsb, String jobName, - SimulationTask simTask, int jobTimeoutSeconds) { - + private void writeBatchScriptControlledVariables(LineStringBuilder lsb, String jobName, + SimulationTask simTask, int jobTimeoutSeconds, + int watchdogTickSeconds, int watchdogTimeoutSeconds) // watchdog stuff + { String simKey = simTask.getSimulation().getVersion().getVersionKey().toString(); String simOwnerName = simTask.getSimulation().getVersion().getOwner().getName(); // this is the user name String simOwnerId = simTask.getSimulation().getVersion().getOwner().getID().toString(); @@ -659,9 +660,6 @@ private void writeScriptControlledVariables(LineStringBuilder lsb, String jobNam String logFilePath = htcLogDir + "/" + trimmedJobName + ".submit.log"; String messagingConfigFilePath = simDataDir + "/" + simOwnerName + "/SimID_" + simKey + "_0_.langevinMessagingConfig"; - // ---- Alternate / New variables ---- - - lsb.write("# Script-controlled variables (populated by generator in real use)"); // lsb.write("USERID=" + simOwnerName); // lsb.write("SIM_KEY=" + simId); @@ -673,6 +671,8 @@ private void writeScriptControlledVariables(LineStringBuilder lsb, String jobNam lsb.write("TOTAL_JOBS=" + totalJobs + " # to be set by generator to lso.getTotalNumberOfJobs()"); lsb.write("JOB_TIMEOUT_SECONDS=" + jobTimeoutSeconds + " # per-job timeout (seconds), adjust per generator"); + lsb.write("WATCHDOG_TICK_SECONDS=" + watchdogTickSeconds + " # watchdog tick interval (seconds)"); + lsb.write("WATCHDOG_TIMEOUT_SECONDS=" + watchdogTimeoutSeconds + " # watchdog timeout (seconds)"); lsb.write("LOG_FILE=\"" + logFilePath + "\""); lsb.write("MESSAGING_CONFIG_FILE=\"" + messagingConfigFilePath + "\""); lsb.write(""); @@ -687,7 +687,7 @@ private void writeScriptControlledVariables(LineStringBuilder lsb, String jobNam lsb.write("rm -f " + simDataDir + "/${SIM_OWNER_NAME}/SimID_${SIM_KEY}_0_.langevinMessagingConfig"); lsb.write(""); } - private void writeSingularitySetup(LineStringBuilder lsb) { + private void writeBatchSingularitySetup(LineStringBuilder lsb) { String slurmTmpDir = PropertyLoader.getRequiredProperty(PropertyLoader.slurm_tmpdir); String singularityCachedir = PropertyLoader.getRequiredProperty(PropertyLoader.slurm_singularity_cachedir); String singularityPullfolder = PropertyLoader.getRequiredProperty(PropertyLoader.slurm_singularity_pullfolder); @@ -712,7 +712,7 @@ private void writeSingularitySetup(LineStringBuilder lsb) { lsb.write("export SINGULARITY_PULLFOLDER=" + singularityPullfolder); lsb.write(""); } - private void writeSlurmJobMetadata(LineStringBuilder lsb) { + private void writeBatchSlurmJobMetadata(LineStringBuilder lsb) { lsb.write("# Compute memory per task and per job"); lsb.write("MEM_TASK=$(( SLURM_MEM_PER_CPU * SLURM_CPUS_PER_TASK ))"); lsb.write("MEM_JOB=$(( MEM_TASK * SLURM_NTASKS ))"); @@ -739,7 +739,7 @@ private void writeSlurmJobMetadata(LineStringBuilder lsb) { lsb.write("echo \"=================================\""); lsb.write(""); } - private void writeContainerBindingsAndEnv(LineStringBuilder lsb, int javaMemXmx) { + private void writeBatchContainerBindingsAndEnv(LineStringBuilder lsb, int javaMemXmx) { String primaryDataDir = PropertyLoader.getRequiredProperty(PropertyLoader.primarySimDataDirExternalProperty); String secondaryDataDir = PropertyLoader.getRequiredProperty(PropertyLoader.secondarySimDataDirExternalProperty); String archiveDataDirHost = PropertyLoader.getRequiredProperty(PropertyLoader.simDataDirArchiveExternal); @@ -783,7 +783,7 @@ private void writeContainerBindingsAndEnv(LineStringBuilder lsb, int javaMemXmx) lsb.write("container_env+=\"--env serverid=" + serverId + " \""); lsb.write(""); } - private void writeContainerImageAndPrefixes(LineStringBuilder lsb) { + private void writeBatchContainerImageAndPrefixes(LineStringBuilder lsb) { final String batchApptainerImage = PropertyLoader.getRequiredProperty(PropertyLoader.htc_vcellbatch_apptainer_image); final String sifImageDir = PropertyLoader.getRequiredProperty(PropertyLoader.htc_singularity_imagedir); final String sifPath = sifPathFromOrasUrl(batchApptainerImage, sifImageDir); @@ -811,9 +811,6 @@ private void writeSifContainerPrefix(LineStringBuilder lsb, String prefix, Strin String generateLangevinBatchScript(String jobName, ExecutableCommand.Container commandSet, double memSizeMB, Collection postProcessingCommands, SimulationTask simTask) { - // TODO: extractUser is very unrobust, must be fixed - // it may be the userName can be obtained like so: String vcellUserid = simTask.getUser().getName(); - //String userName = extractUser(commandSet); String vcellUserid = simTask.getUser().getName(); KeyValue simID = simTask.getSimulationInfo().getSimulationVersion().getVersionKey(); SolverTaskDescription std = simTask.getSimulation().getSolverTaskDescription(); @@ -833,6 +830,12 @@ String generateLangevinBatchScript(String jobName, ExecutableCommand.Container String sTimeoutPerTaskSeconds = PropertyLoader.getRequiredProperty(PropertyLoader.slurm_langevin_timeoutPerTaskSeconds); String sHardbBtchMemoryLimitPerTask = PropertyLoader.getRequiredProperty(PropertyLoader.slurm_langevin_batchMemoryLimitPerTaskMB); String sBlockSizeMB = PropertyLoader.getRequiredProperty(PropertyLoader.slurm_langevin_memoryBlockSizeMB); + String sWatchdogTickSeconds = PropertyLoader.getProperty(PropertyLoader.slurm_langevin_watchdogTickSeconds, "60"); + String sWatchdogTimeoutSeconds = PropertyLoader.getProperty(PropertyLoader.slurm_langevin_watchdogTimeoutSeconds, "600"); + + // we don't need to convert many of these strings to numeric only to convert them again to strings for the script, + // but it's prudent to validate that they are numeric (and within reasonable bounds?), so we do it here and fail + // early rather than launch the batch job and have it fail server-side int timeoutPerTaskSeconds = Integer.parseInt(sTimeoutPerTaskSeconds); long hardbBtchMemoryLimitPerTask = Long.parseLong(sHardbBtchMemoryLimitPerTask); // MB. we hard limit mem to 2G for langevin batch jobs int blockSizeMB = Integer.parseInt(sBlockSizeMB); // MB. SLURM memory allocation granularity @@ -840,16 +843,18 @@ String generateLangevinBatchScript(String jobName, ExecutableCommand.Container long batchMemoryLimitPerTask = memoryMBAllowed.getMemLimit(); batchMemoryLimitPerTask = Math.min(batchMemoryLimitPerTask, hardbBtchMemoryLimitPerTask); int javaMemXmx = roundUpToBlock(batchMemoryLimitPerTask, blockSizeMB) + blockSizeMB; // add extra block for overhead + int watchdogTickSeconds = Integer.parseInt(sWatchdogTickSeconds); + int watchdogTimeoutSeconds = Integer.parseInt(sWatchdogTimeoutSeconds); // ------------------------------------------------------------- LineStringBuilder lsb = new LineStringBuilder(); slurmBatchScriptInit(jobName, simTask.isPowerUser(), memoryMBAllowed, numberOfConcurrentTasks, slurmJobTimeout, lsb); - writeScriptControlledVariables(lsb, jobName, simTask, timeoutPerTaskSeconds); - writeSingularitySetup(lsb); - writeSlurmJobMetadata(lsb); - writeContainerBindingsAndEnv(lsb, javaMemXmx); - writeContainerImageAndPrefixes(lsb); + writeBatchScriptControlledVariables(lsb, jobName, simTask, timeoutPerTaskSeconds, watchdogTickSeconds, watchdogTimeoutSeconds); + writeBatchSingularitySetup(lsb); + writeBatchSlurmJobMetadata(lsb); + writeBatchContainerBindingsAndEnv(lsb, javaMemXmx); + writeBatchContainerImageAndPrefixes(lsb); String langevinFixture; try { diff --git a/vcell-server/src/main/resources/slurm/templates/langevinFixture.slurm.sub b/vcell-server/src/main/resources/slurm/templates/langevinFixture.slurm.sub index f2870fd0e5..d25223573c 100644 --- a/vcell-server/src/main/resources/slurm/templates/langevinFixture.slurm.sub +++ b/vcell-server/src/main/resources/slurm/templates/langevinFixture.slurm.sub @@ -73,6 +73,30 @@ running_jobs=0 max_concurrent_jobs=${SLURM_NTASKS:-4} # actually these are tasks echo "max_concurrent_jobs: ${max_concurrent_jobs}" >> "${LOG_FILE}" echo "TOTAL_JOBS to launch: ${TOTAL_JOBS}" >> "${LOG_FILE}" +echo "WATCHDOG_TICK_SECONDS: ${WATCHDOG_TICK_SECONDS}" >> "${LOG_FILE}" +echo "WATCHDOG_TIMEOUT_SECONDS: ${WATCHDOG_TIMEOUT_SECONDS}" >> "${LOG_FILE}" + +# ---------------------------------------------------------------------------------------------- +# watchdog solver invocation (monitors the other jobs and sends progress messages) +# timeouts early if no jobs is being launched, does not timeout once in main loop +echo "Starting watchdog job at $(date)" >> "${LOG_FILE}" +${slurm_prefix} ${solver_container_prefix} \ + langevin_x64 watchdog \ + "${INPUT_DIR}/SimID_${SIM_KEY}_0_.langevinInput" \ + ${TOTAL_JOBS} \ + --watchdog-tick="${WATCHDOG_TICK_SECONDS}" \ + --watchdog-timeout="${WATCHDOG_TIMEOUT_SECONDS}" \ +# --output-log="${LOG_DIR}/SimID_${SIM_KEY}_0_W.log" \ + --vc-print-status="${INPUT_DIR}/SimID_${SIM_KEY}_0_.langevinMessagingConfig" & +watchdog_pid=$! +sleep 1 +if ! kill -0 $watchdog_pid 2>/dev/null; then + echo "Watchdog failed to start" >> "${LOG_FILE}" + wait $watchdog_pid + watchdog_exit_code=$? + callExitProcessor $watchdog_exit_code + exit $watchdog_exit_code +fi finished_jobs=0 # messaging counters last_notify_time=0 @@ -99,6 +123,15 @@ for ((i = 0; i < TOTAL_JOBS; i++)); do # wait for a finished task before launching a new one if we hit the concurrency limit while (( running_jobs >= max_concurrent_jobs )); do + # Check watchdog health + if ! kill -0 $watchdog_pid 2>/dev/null; then + echo "Watchdog died unexpectedly at $(date)" >> "${LOG_FILE}" + wait $watchdog_pid + watchdog_exit=$? + callExitProcessor $watchdog_exit + echo "Watchdog failed; exiting $watchdog_exit" | tee -a "${LOG_FILE}" + exit $watchdog_exit + fi for idx in "${!job_pids[@]}"; do pid="${job_pids[$idx]}" if [[ -z "${pid}" ]]; then continue; fi @@ -116,23 +149,23 @@ for ((i = 0; i < TOTAL_JOBS; i++)); do unset "job_pid_map[$pid]" # remove mapping ((running_jobs--)) ((finished_jobs++)) - progress=$(awk "BEGIN {print ${finished_jobs}/(${TOTAL_JOBS}+1)}") # compute progress - timepoint=$(date +%s) - if (( timepoint - last_notify_time >= MIN_NOTIFY_INTERVAL )); then - last_notify_time=$timepoint - statusCode=1001 - statusMsg="Running" - RUNTIME_PROPERTIES="&WorkerEvent_Status=${statusCode}" - RUNTIME_PROPERTIES+="&WorkerEvent_StatusMsg=${statusMsg}" - RUNTIME_PROPERTIES+="&WorkerEvent_TimePoint=${timepoint}" - RUNTIME_PROPERTIES+="&WorkerEvent_Progress=${progress}" - PROPERTIES="${BASE_PROPERTIES}${RUNTIME_PROPERTIES}" - msgCommand="set -o errexit; set -o pipefail; set -o nounset" - msgCommand+=" - curl -v -XPOST \"http://${broker_username}:${broker_password}@${broker_host}:${broker_port}/api/message/workerEvent?type=queue&${PROPERTIES}\"" - ${solver_container_prefix} /bin/bash -c "$msgCommand" # execute inside singularity - echo "progress notification sent, ${progress} done" >> "${LOG_FILE}" - fi +# progress=$(awk "BEGIN {print ${finished_jobs}/(${TOTAL_JOBS}+1)}") # compute progress +# timepoint=$(date +%s) +# if (( timepoint - last_notify_time >= MIN_NOTIFY_INTERVAL )); then +# last_notify_time=$timepoint +# statusCode=1001 +# statusMsg="Running" +# RUNTIME_PROPERTIES="&WorkerEvent_Status=${statusCode}" +# RUNTIME_PROPERTIES+="&WorkerEvent_StatusMsg=${statusMsg}" +# RUNTIME_PROPERTIES+="&WorkerEvent_TimePoint=${timepoint}" +# RUNTIME_PROPERTIES+="&WorkerEvent_Progress=${progress}" +# PROPERTIES="${BASE_PROPERTIES}${RUNTIME_PROPERTIES}" +# msgCommand="set -o errexit; set -o pipefail; set -o nounset" +# msgCommand+=" +# curl -v -XPOST \"http://${broker_username}:${broker_password}@${broker_host}:${broker_port}/api/message/workerEvent?type=queue&${PROPERTIES}\"" +# ${solver_container_prefix} /bin/bash -c "$msgCommand" # execute inside singularity +# echo "progress notification sent, ${progress} done" >> "${LOG_FILE}" +# fi break # exit for-loop once we free up a slot fi done @@ -153,23 +186,23 @@ for pid in "${job_pids[@]}"; do job_index=${job_pid_map[$pid]:-unknown} echo "Task $job_index finished with exit code $exit_code at $(date)" >> "${LOG_FILE}" ((finished_jobs++)) - progress=$(awk "BEGIN {print ${finished_jobs}/(${TOTAL_JOBS}+1)}") # compute progress - timepoint=$(date +%s) - if (( timepoint - last_notify_time >= MIN_NOTIFY_INTERVAL )); then - last_notify_time=$timepoint - statusCode=1001 - statusMsg="Running" - RUNTIME_PROPERTIES="&WorkerEvent_Status=${statusCode}" - RUNTIME_PROPERTIES+="&WorkerEvent_StatusMsg=${statusMsg}" - RUNTIME_PROPERTIES+="&WorkerEvent_TimePoint=${timepoint}" - RUNTIME_PROPERTIES+="&WorkerEvent_Progress=${progress}" - PROPERTIES="${BASE_PROPERTIES}${RUNTIME_PROPERTIES}" - msgCommand="set -o errexit; set -o pipefail; set -o nounset" - msgCommand+=" - curl -v -XPOST \"http://${broker_username}:${broker_password}@${broker_host}:${broker_port}/api/message/workerEvent?type=queue&${PROPERTIES}\"" - ${solver_container_prefix} /bin/bash -c "$msgCommand" # execute inside singularity - echo "progress notification sent, ${progress} done" >> "${LOG_FILE}" - fi +# progress=$(awk "BEGIN {print ${finished_jobs}/(${TOTAL_JOBS}+1)}") # compute progress +# timepoint=$(date +%s) +# if (( timepoint - last_notify_time >= MIN_NOTIFY_INTERVAL )); then +# last_notify_time=$timepoint +# statusCode=1001 +# statusMsg="Running" +# RUNTIME_PROPERTIES="&WorkerEvent_Status=${statusCode}" +# RUNTIME_PROPERTIES+="&WorkerEvent_StatusMsg=${statusMsg}" +# RUNTIME_PROPERTIES+="&WorkerEvent_TimePoint=${timepoint}" +# RUNTIME_PROPERTIES+="&WorkerEvent_Progress=${progress}" +# PROPERTIES="${BASE_PROPERTIES}${RUNTIME_PROPERTIES}" +# msgCommand="set -o errexit; set -o pipefail; set -o nounset" +# msgCommand+=" +# curl -v -XPOST \"http://${broker_username}:${broker_password}@${broker_host}:${broker_port}/api/message/workerEvent?type=queue&${PROPERTIES}\"" +# ${solver_container_prefix} /bin/bash -c "$msgCommand" # execute inside singularity +# echo "progress notification sent, ${progress} done" >> "${LOG_FILE}" +# fi done echo "Batch job completed at $(date)" >> "${LOG_FILE}" @@ -181,8 +214,9 @@ timeout "${JOB_TIMEOUT_SECONDS}s" \ langevin_x64 postprocess \ "${INPUT_DIR}/SimID_${SIM_KEY}_0_.langevinInput" \ ${TOTAL_JOBS} \ - --output-log="${LOG_DIR}/SimID_${SIM_KEY}_0_P.log" \ - --vc-print-status & + --output-log="${LOG_DIR}/SimID_${SIM_KEY}_0_P.log" +# --output-log="${LOG_DIR}/SimID_${SIM_KEY}_0_P.log" \ +# --vc-print-status & last_pid=$! wait $last_pid diff --git a/vcell-server/src/test/resources/slurm_fixtures/langevin/V_TEST2_999999999_0_0.slurm.sub b/vcell-server/src/test/resources/slurm_fixtures/langevin/V_TEST2_999999999_0_0.slurm.sub index a93f73215a..97060df61b 100644 --- a/vcell-server/src/test/resources/slurm_fixtures/langevin/V_TEST2_999999999_0_0.slurm.sub +++ b/vcell-server/src/test/resources/slurm_fixtures/langevin/V_TEST2_999999999_0_0.slurm.sub @@ -26,6 +26,8 @@ VC_JOB_ID=0 VC_TASK_ID=0 TOTAL_JOBS=8 # to be set by generator to lso.getTotalNumberOfJobs() JOB_TIMEOUT_SECONDS=345600 # per-job timeout (seconds), adjust per generator +WATCHDOG_TICK_SECONDS=60 # watchdog tick interval (seconds) +WATCHDOG_TIMEOUT_SECONDS=600 # watchdog timeout (seconds) LOG_FILE="/share/apps/vcell3/htclogs/V_TEST2_999999999_0_.submit.log" MESSAGING_CONFIG_FILE="/share/apps/vcell3/users/danv/SimID_999999999_0_.langevinMessagingConfig" @@ -192,6 +194,30 @@ running_jobs=0 max_concurrent_jobs=${SLURM_NTASKS:-4} # actually these are tasks echo "max_concurrent_jobs: ${max_concurrent_jobs}" >> "${LOG_FILE}" echo "TOTAL_JOBS to launch: ${TOTAL_JOBS}" >> "${LOG_FILE}" +echo "WATCHDOG_TICK_SECONDS: ${WATCHDOG_TICK_SECONDS}" >> "${LOG_FILE}" +echo "WATCHDOG_TIMEOUT_SECONDS: ${WATCHDOG_TIMEOUT_SECONDS}" >> "${LOG_FILE}" + +# ---------------------------------------------------------------------------------------------- +# watchdog solver invocation (monitors the other jobs and sends progress messages) +# timeouts early if no jobs is being launched, does not timeout once in main loop +echo "Starting watchdog job at $(date)" >> "${LOG_FILE}" +${slurm_prefix} ${solver_container_prefix} \ + langevin_x64 watchdog \ + "${INPUT_DIR}/SimID_${SIM_KEY}_0_.langevinInput" \ + ${TOTAL_JOBS} \ + --watchdog-tick="${WATCHDOG_TICK_SECONDS}" \ + --watchdog-timeout="${WATCHDOG_TIMEOUT_SECONDS}" \ +# --output-log="${LOG_DIR}/SimID_${SIM_KEY}_0_W.log" \ + --vc-print-status="${INPUT_DIR}/SimID_${SIM_KEY}_0_.langevinMessagingConfig" & +watchdog_pid=$! +sleep 1 +if ! kill -0 $watchdog_pid 2>/dev/null; then + echo "Watchdog failed to start" >> "${LOG_FILE}" + wait $watchdog_pid + watchdog_exit_code=$? + callExitProcessor $watchdog_exit_code + exit $watchdog_exit_code +fi finished_jobs=0 # messaging counters last_notify_time=0 @@ -218,6 +244,15 @@ for ((i = 0; i < TOTAL_JOBS; i++)); do # wait for a finished task before launching a new one if we hit the concurrency limit while (( running_jobs >= max_concurrent_jobs )); do + # Check watchdog health + if ! kill -0 $watchdog_pid 2>/dev/null; then + echo "Watchdog died unexpectedly at $(date)" >> "${LOG_FILE}" + wait $watchdog_pid + watchdog_exit=$? + callExitProcessor $watchdog_exit + echo "Watchdog failed; exiting $watchdog_exit" | tee -a "${LOG_FILE}" + exit $watchdog_exit + fi for idx in "${!job_pids[@]}"; do pid="${job_pids[$idx]}" if [[ -z "${pid}" ]]; then continue; fi @@ -235,23 +270,23 @@ for ((i = 0; i < TOTAL_JOBS; i++)); do unset "job_pid_map[$pid]" # remove mapping ((running_jobs--)) ((finished_jobs++)) - progress=$(awk "BEGIN {print ${finished_jobs}/(${TOTAL_JOBS}+1)}") # compute progress - timepoint=$(date +%s) - if (( timepoint - last_notify_time >= MIN_NOTIFY_INTERVAL )); then - last_notify_time=$timepoint - statusCode=1001 - statusMsg="Running" - RUNTIME_PROPERTIES="&WorkerEvent_Status=${statusCode}" - RUNTIME_PROPERTIES+="&WorkerEvent_StatusMsg=${statusMsg}" - RUNTIME_PROPERTIES+="&WorkerEvent_TimePoint=${timepoint}" - RUNTIME_PROPERTIES+="&WorkerEvent_Progress=${progress}" - PROPERTIES="${BASE_PROPERTIES}${RUNTIME_PROPERTIES}" - msgCommand="set -o errexit; set -o pipefail; set -o nounset" - msgCommand+=" - curl -v -XPOST \"http://${broker_username}:${broker_password}@${broker_host}:${broker_port}/api/message/workerEvent?type=queue&${PROPERTIES}\"" - ${solver_container_prefix} /bin/bash -c "$msgCommand" # execute inside singularity - echo "progress notification sent, ${progress} done" >> "${LOG_FILE}" - fi +# progress=$(awk "BEGIN {print ${finished_jobs}/(${TOTAL_JOBS}+1)}") # compute progress +# timepoint=$(date +%s) +# if (( timepoint - last_notify_time >= MIN_NOTIFY_INTERVAL )); then +# last_notify_time=$timepoint +# statusCode=1001 +# statusMsg="Running" +# RUNTIME_PROPERTIES="&WorkerEvent_Status=${statusCode}" +# RUNTIME_PROPERTIES+="&WorkerEvent_StatusMsg=${statusMsg}" +# RUNTIME_PROPERTIES+="&WorkerEvent_TimePoint=${timepoint}" +# RUNTIME_PROPERTIES+="&WorkerEvent_Progress=${progress}" +# PROPERTIES="${BASE_PROPERTIES}${RUNTIME_PROPERTIES}" +# msgCommand="set -o errexit; set -o pipefail; set -o nounset" +# msgCommand+=" +# curl -v -XPOST \"http://${broker_username}:${broker_password}@${broker_host}:${broker_port}/api/message/workerEvent?type=queue&${PROPERTIES}\"" +# ${solver_container_prefix} /bin/bash -c "$msgCommand" # execute inside singularity +# echo "progress notification sent, ${progress} done" >> "${LOG_FILE}" +# fi break # exit for-loop once we free up a slot fi done @@ -272,23 +307,23 @@ for pid in "${job_pids[@]}"; do job_index=${job_pid_map[$pid]:-unknown} echo "Task $job_index finished with exit code $exit_code at $(date)" >> "${LOG_FILE}" ((finished_jobs++)) - progress=$(awk "BEGIN {print ${finished_jobs}/(${TOTAL_JOBS}+1)}") # compute progress - timepoint=$(date +%s) - if (( timepoint - last_notify_time >= MIN_NOTIFY_INTERVAL )); then - last_notify_time=$timepoint - statusCode=1001 - statusMsg="Running" - RUNTIME_PROPERTIES="&WorkerEvent_Status=${statusCode}" - RUNTIME_PROPERTIES+="&WorkerEvent_StatusMsg=${statusMsg}" - RUNTIME_PROPERTIES+="&WorkerEvent_TimePoint=${timepoint}" - RUNTIME_PROPERTIES+="&WorkerEvent_Progress=${progress}" - PROPERTIES="${BASE_PROPERTIES}${RUNTIME_PROPERTIES}" - msgCommand="set -o errexit; set -o pipefail; set -o nounset" - msgCommand+=" - curl -v -XPOST \"http://${broker_username}:${broker_password}@${broker_host}:${broker_port}/api/message/workerEvent?type=queue&${PROPERTIES}\"" - ${solver_container_prefix} /bin/bash -c "$msgCommand" # execute inside singularity - echo "progress notification sent, ${progress} done" >> "${LOG_FILE}" - fi +# progress=$(awk "BEGIN {print ${finished_jobs}/(${TOTAL_JOBS}+1)}") # compute progress +# timepoint=$(date +%s) +# if (( timepoint - last_notify_time >= MIN_NOTIFY_INTERVAL )); then +# last_notify_time=$timepoint +# statusCode=1001 +# statusMsg="Running" +# RUNTIME_PROPERTIES="&WorkerEvent_Status=${statusCode}" +# RUNTIME_PROPERTIES+="&WorkerEvent_StatusMsg=${statusMsg}" +# RUNTIME_PROPERTIES+="&WorkerEvent_TimePoint=${timepoint}" +# RUNTIME_PROPERTIES+="&WorkerEvent_Progress=${progress}" +# PROPERTIES="${BASE_PROPERTIES}${RUNTIME_PROPERTIES}" +# msgCommand="set -o errexit; set -o pipefail; set -o nounset" +# msgCommand+=" +# curl -v -XPOST \"http://${broker_username}:${broker_password}@${broker_host}:${broker_port}/api/message/workerEvent?type=queue&${PROPERTIES}\"" +# ${solver_container_prefix} /bin/bash -c "$msgCommand" # execute inside singularity +# echo "progress notification sent, ${progress} done" >> "${LOG_FILE}" +# fi done echo "Batch job completed at $(date)" >> "${LOG_FILE}" @@ -300,8 +335,9 @@ timeout "${JOB_TIMEOUT_SECONDS}s" \ langevin_x64 postprocess \ "${INPUT_DIR}/SimID_${SIM_KEY}_0_.langevinInput" \ ${TOTAL_JOBS} \ - --output-log="${LOG_DIR}/SimID_${SIM_KEY}_0_P.log" \ - --vc-print-status & + --output-log="${LOG_DIR}/SimID_${SIM_KEY}_0_P.log" +# --output-log="${LOG_DIR}/SimID_${SIM_KEY}_0_P.log" \ +# --vc-print-status & last_pid=$! wait $last_pid