Skip to content

Commit 4d6fb61

Browse files
sawenzelsawenzel
committed
Improvements to taskwrapper
* More direct CPU monitoring: - calc CPU usage directly in script * Prototypic possibility to limit the total CPU load of a task running within taskwrapper by using cpulimit (adaptively). This may be useful for several reasons. Here, it started as an investigation to prevent CPU spikes from workflows when running on single-core queues (until DPL has it's own way of forcing single-core). Or one might just want to emulate the constraints of a GRID multicore queue locally. By default the feature is switched off. It can be enabled via environment variables. For example ``` JOBUTILS_LIMITLOAD=300 taskwrapper task1.log task1 ``` will constrain task1 to using ~3 full CPU cores. The regulation mechanism is quite simplistic but should be enough to demonstrate the feature.
1 parent 35967c5 commit 4d6fb61

1 file changed

Lines changed: 88 additions & 11 deletions

File tree

‎Utilities/Tools/jobutils.sh‎

Lines changed: 88 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -13,8 +13,11 @@
1313
# This file contains a couple of utility functions for reuse
1414
# in shell job scripts (such as on the GRID).
1515
# In order to use these functions in scripts, this file needs to be
16-
# simply sourced into the target script.
16+
# simply sourced into the target script. The script needs bash versions > 4
1717

18+
# TODOs:
19+
# -harmonize use of bc/awk for calculations
20+
# -harmonize coding style for variables
1821

1922
o2_cleanup_shm_files() {
2023
# check if we have lsof (otherwise we do nothing)
@@ -91,6 +94,8 @@ taskwrapper_cleanup_handler() {
9194
# (until DPL offers signal handling and automatic shutdown)
9295
# - possibility to provide user hooks for "start" and "failure"
9396
# - possibility to skip (jump over) job alltogether
97+
# - possibility to define timeout
98+
# - possibility to control/limit the CPU load
9499
taskwrapper() {
95100
local logfile=$1
96101
shift 1
@@ -141,7 +146,7 @@ taskwrapper() {
141146
finalcommand="TIME=\"#walltime %e\" ${O2_ROOT}/share/scripts/monitor-mem.sh ${TIMECOMMAND} './${SCRIPTNAME}'"
142147
fi
143148
echo "Running: ${finalcommand}" > ${logfile}
144-
eval ${finalcommand} >> ${logfile} 2>&1 &
149+
eval ${finalcommand} >> ${logfile} 2>&1 & disown
145150

146151
# THE NEXT PART IS THE SUPERVISION PART
147152
# get the PID
@@ -151,7 +156,9 @@ taskwrapper() {
151156
trap "taskwrapper_cleanup_handler ${PID} SIGTERM" SIGTERM
152157

153158
cpucounter=1
159+
NLOGICALCPUS=$(getNumberOfLogicalCPUCores)
154160

161+
reduction_factor=1
155162
while [ 1 ]; do
156163
# We don't like to see critical problems in the log file.
157164

@@ -202,21 +209,83 @@ taskwrapper() {
202209
ps -p $PID > /dev/null
203210
[ $? == 1 ] && break
204211

205-
if [ "${JOBUTILS_MONITORCPU}" ]; then
212+
if [ "${JOBUTILS_MONITORCPU}" ] || [ "${JOBUTILS_LIMITLOAD}" ]; then
213+
# NOTE: The following section is "a bit" compute intensive and currently not optimized
214+
# A careful evaluation of awk vs bc or other tools might be needed -- or a move to a more
215+
# system oriented language/tool
216+
217+
for p in $limitPIDs; do
218+
wait ${p}
219+
done
220+
206221
# get some CPU usage statistics per process --> actual usage can be calculated thereafter
207-
for p in $(childprocs ${PID}); do
208-
total=`awk 'BEGIN{s=0}/cpu /{for (i=1;i<=NF;i++) s+=$i;} END {print s}' /proc/stat`
209-
utime=`awk '//{print $14}' /proc/${p}/stat 2> /dev/null`
210-
stime=`awk '//{print $15}' /proc/${p}/stat 2> /dev/null`
211-
name=`awk '//{print $2}' /proc/${p}/stat 2> /dev/null`
212-
echo "${cpucounter} ${p} ${total} ${utime} ${stime} ${name}" >> ${logfile}_cpuusage
222+
total=`awk 'BEGIN{s=0}/cpu /{for (i=1;i<=NF;i++) s+=$i;} END {print s}' /proc/stat`
223+
previous_total=${current_total}
224+
current_total=${total}
225+
# quickly fetch the data
226+
childpids=$(childprocs ${PID})
227+
228+
for p in $childpids; do
229+
while read -r name utime stime; do
230+
echo "${cpucounter} ${p} ${total} ${utime} ${stime} ${name}" >> ${logfile}_cpuusage
231+
previous[$p]=${current[$p]}
232+
current[$p]=${utime}
233+
name[$p]=${name}
234+
done <<<$(awk '//{print $2" "$14" "$15}' /proc/${p}/stat 2>/dev/null)
213235
done
236+
# do some calculations based on the data
237+
totalCPU=0 # actual CPU load measured
238+
totalCPU_unlimited=0 # extrapolated unlimited CPU load
239+
line=""
240+
for p in $childpids; do
241+
C=${current[$p]}
242+
P=${previous[$p]}
243+
CT=${total}
244+
PT=${previous_total}
245+
# echo "${p} : current ${C} previous ${P} ${CT} ${PT}"
246+
thisCPU[$p]=$(awk -v "c=${C}" -v "p=${P}" -v "ct=${CT}" -v "pt=${PT}" -v "ncpu=${NLOGICALCPUS}" 'BEGIN { print 100.*ncpu*(c-p)/(ct-pt); }')
247+
line="${line} $p:${thisCPU[$p]}"
248+
totalCPU=$(awk -v "t=${totalCPU}" -v "this=${thisCPU[$p]}" 'BEGIN { print (t + this); }')
249+
previousfactor=1
250+
[ ${waslimited[$p]} ] && previousfactor=${reduction_factor}
251+
totalCPU_unlimited=$(awk -v "t=${totalCPU_unlimited}" -v "this=${thisCPU[$p]}" -v f="${previousfactor}" 'BEGIN { print (t + this/f); }')
252+
# echo "CPU last time window ${p} : ${thisCPU[$p]}"
253+
done
254+
255+
echo "${line}"
256+
echo "${cpucounter} totalCPU = ${totalCPU} -- without limitation ${totalCPU_unlimited}"
257+
# We can check if the total load is above a resource limit
258+
# And take corrective actions if we extend by 10%
259+
limitPIDs=""
260+
unset waslimited
261+
if (( $(echo "${totalCPU_unlimited} > 1.1*${JOBUTILS_LIMITLOAD}" | bc -l) )); then
262+
# we reduce each pid proportionally for the time until the next check and record the reduction factor in place
263+
oldreduction=${reduction_factor}
264+
reduction_factor=$(awk -v limit="${JOBUTILS_LIMITLOAD}" -v cur="${totalCPU_unlimited}" 'BEGIN{ print limit/cur;}')
265+
echo "APPLYING REDUCTION = ${reduction_factor}"
266+
267+
for p in $childpids; do
268+
cpulim=$(awk -v a="${thisCPU[${p}]}" -v newr="${reduction_factor}" -v oldr="${oldreduction}" 'BEGIN { r=(a/oldr)*newr; print r; if(r > 0.05) {exit 0;} exit 1; }')
269+
if [ $? = "0" ]; then
270+
# we only apply to jobs above a certain threshold
271+
echo "Setting CPU lim for job ${p} / ${name[$p]} to ${cpulim}";
272+
273+
timeout ${JOBUTILS_WRAPPER_SLEEP} ${O2_ROOT}/share/scripts/cpulimit -l ${cpulim} -p ${p} > /dev/null 2> /dev/null & disown
274+
proc=$!
275+
limitPIDs="${limitPIDs} ${proc}"
276+
waslimited[$p]=1
277+
fi
278+
done
279+
else
280+
echo "RESETING REDUCTION = 1"
281+
reduction_factor=1.
282+
fi
214283
let cpucounter=cpucounter+1
215284
fi
216285

217286
# a good moment to check for jobs timeout (or other resources)
218287
if [ "$JOBUTILS_JOB_TIMEOUT" ]; then
219-
$(awk -v S="${SECONDS}" -v T="${JOBUTILS_JOB_TIMEOUT}" -v START="${STARTTIME}" '//{} END{if((S-START)>T){exit 1;} exit 0;}' < /dev/null)
288+
$(awk -v S="${SECONDS}" -v T="${JOBUTILS_JOB_TIMEOUT}" -v START="${STARTTIME}" 'BEGIN {if((S-START)>T){exit 1;} exit 0;}')
220289
if [ "$?" = "1" ]; then
221290
echo "task timeout reached .. killing all processes";
222291
taskwrapper_cleanup $PID SIGKILL
@@ -230,7 +299,7 @@ taskwrapper() {
230299
fi
231300

232301
# sleep for some time (can be customized for power user)
233-
sleep ${JOBUTILS_WRAPPER_SLEEP:-5}
302+
sleep ${JOBUTILS_WRAPPER_SLEEP:-10}
234303
done
235304

236305
# wait for PID and fetch return code
@@ -281,3 +350,11 @@ getNumberOfPhysicalCPUCores() {
281350
echo "${N}"
282351
}
283352

353+
getNumberOfLogicalCPUCores() {
354+
if [ "$(uname)" == "Darwin" ]; then
355+
echo $(sysctl -n hw.logicalcpu)
356+
else
357+
# Do something under GNU/Linux platform
358+
echo $(grep "processor" /proc/cpuinfo | wc -l)
359+
fi
360+
}

0 commit comments

Comments
 (0)