· 9 years ago · Nov 14, 2016, 05:12 PM
1#!/bin/bash
2
3# Fail everything if even a single command fails
4set -e
5
6########################## PARALLEL SUPPORT ##########################
7function initialize {
8 local proc_list=( $@ )
9 local_port=$(python -c 'import socket; s=socket.socket(); s.bind(("", 0)); print(s.getsockname()[1]); s.close()')
10 local tmpdir=$(mktemp -d)
11 trap 'rm -rf "$tmpdir"' EXIT INT TERM HUP
12 free_queue="${tmpdir}/Q"
13 barrier_queue="${tmpdir}/B"
14
15 rm -f "${free_queue}" "${barrier_queue}"
16 mkfifo "${free_queue}" "${barrier_queue}"
17 # Launch netcat server
18 (
19 printf -- '-!- %11s: %d\n' "PidServer" "${BASHPID}" >&2
20 local procs=(${proc_list[@]})
21 local proc_no=${#procs[@]}
22 local grabs=0
23 local barrier=false
24 nc -d -k -l ${local_port} | (
25 printf -- '-!- %11s: %d\n' "EventLoop" "${BASHPID}" >&2
26 while read command ; do
27 # echo "*** <- ${command}" >&2
28 if [ "${command}" = B ] ; then
29 if [ ${#procs[@]} -eq ${proc_no} ] ; then
30 # echo "*** -> Barrier" >&2
31 echo "B" "K"
32 else
33 # echo "*** -!- barrier = true" >&2
34 local barrier=true
35 fi
36 elif [ "${command}" = Q ] ; then
37 # echo "*** -!- Exit@${local_port}" >&2
38 exit
39 elif [ "${command}" = G ] ; then
40 if [ ${#procs[@]} -eq 0 ] ; then
41 grabs=$((grabs + 1))
42 # echo "*** -!- grabs = ${grabs}" >&2
43 continue
44 fi
45 local to_send=${procs[0]}
46 local procs=( ${procs[@]:1} )
47 # echo "*** -> ${to_send}" >&2
48 echo "F" "${to_send}"
49 else
50 if [ ${grabs} -gt 0 ] ; then
51 local grabs=$((grabs - 1))
52 # echo "*** -> ${command}" >&2
53 echo "F" "${command}"
54 continue
55 else
56 # echo "*** -!- procs = ${procs[@]}" >&2
57 procs+=( ${command} )
58 if [ ${#procs[@]} -eq ${proc_no} ] && ${barrier} ; then
59 # echo "*** -> K" >&2
60 echo "B" "K"
61 fi
62 fi
63 fi
64 done
65 ) | (
66 printf -- '-!- %11s: %d\n' "QueueWriter" "${BASHPID}" >&2
67 declare -A queues=( ["B"]="${barrier_queue}" ["F"]="${free_queue}" )
68 while read Type Value ; do
69 if [ ${Type} = "B" ] ; then
70 echo "${Value}" > "${queues[${Type}]}" &
71 else
72 echo "${Value}" > "${queues[${Type}]}"
73 fi
74 done
75 )
76 ) &
77}
78
79# Wraps a function so it signals when the processor is available again
80function protocol {
81 local proc=${1}
82 shift 1
83 local cmd=( ${@} )
84
85 ${cmd[@]} ${proc}
86 echo ${proc} | nc localhost ${local_port}
87}
88
89# Launch one job
90function allocate_job {
91 local cmd=( $@ )
92
93 echo "G" | nc localhost ${local_port}
94 local proc=$(cat ${free_queue})
95
96 protocol "${proc}" ${cmd[@]} &
97}
98
99# Waits for pending jobs
100function barrier {
101 echo "B" | nc localhost ${local_port}
102 cat "${barrier_queue}" | read
103}
104
105# Command to close queue with netcat server
106function terminate {
107 barrier
108 echo "Q" | nc localhost ${local_port}
109 echo "Q" | nc localhost ${local_port}
110 rm -f "${free_queue}" "${barrier_queue}"
111 rmdir "$(dirname ${free_queue})"
112 unset free_queue
113 unset barrier_queue
114 unset local_port
115}
116
117########################## UTILITY FUNCTIONS ##########################
118function cleanup {
119 pb_terminate
120 terminate
121}
122
123function displaytime {
124 local T=$1
125 local D=$((T/60/60/24))
126 local H=$((T/60/60%24))
127 local M=$((T/60%60))
128 local S=$((T%60))
129 (( $D > 0 )) && printf '%d d ' $D
130 (( $H > 0 )) && printf '%d h ' $H
131 (( $M > 0 )) && printf '%d m ' $M
132 printf '%d s\n' $S
133}
134
135function remaining {
136 local jobs_done="${1}"
137 shift 1
138 local jobs_total="${1}"
139 shift 1
140 local time_start="${1}"
141 local now=$(date +%s)
142 local eta=$(( (${now} - ${time_start}) / ${jobs_done} * (${jobs_total} - ${jobs_done}) ))
143 displaytime ${eta}
144}
145
146function clearLastLine() {
147 echo -en "\r\033[K"
148}
149
150########################## PLUGIN SUPPORT FUNCTIONS ##########################
151function pb_init {
152 if [ ! ${barrier_queue} ] ; then
153 echo "ERROR: run pb_init after init" >&2
154 exit 1
155 fi
156 local queue_dir="$(dirname ${barrier_queue})"
157 id_queue="${queue_dir}/I"
158 mkfifo ${id_queue}
159
160 pb_port=$(python -c 'import socket; s=socket.socket(); s.bind(("", 0)); print(s.getsockname()[1]); s.close()')
161
162 # Launch netcat server
163 (
164 nc -d -k -l ${pb_port} | (
165 printf -- '-!- %11s: %d\n' "IdServer" "${BASHPID}" >&2
166 while read command ; do
167 # echo ">>> -> ${command}" >&2
168 if [ "${command}" = Q ] ; then
169 # echo ">>> -!- Exit@${pb_port}" >&2
170 exit
171 else
172 # echo ">>> I <- ${command}" >&2
173 echo "${command}"
174 fi
175 done
176 ) | (
177 printf -- '-!- %11s: %d\n' "IdWriter" "${BASHPID}" >&2
178 while read id ; do
179 echo "${id}" > ${id_queue}
180 done
181 )
182 ) &
183}
184
185function pb_terminate {
186 if [ ! -e ${barrier_queue} ] ; then
187 echo "ERROR: run pb_terminate before terminate" >&2
188 exit 1
189 fi
190 echo "Q" | nc localhost "${pb_port}"
191 echo "Q" | nc localhost "${pb_port}"
192 rm -f "${id_queue}"
193 unset id_queue
194 unset pb_port
195}
196
197function id_send {
198 local id=${1}
199 echo ${id} | nc localhost "${pb_port}"
200}
201
202function id_receive {
203 local id="$(cat ${id_queue})"
204 printf "%d" "${id}"
205}
206
207function mod_invoke {
208 module="${1}"
209 shift 1
210 function="${1}"
211 shift 1
212
213 ${module}_${function} $@
214 return $?
215}
216
217function get_modules {
218 find -L ${module_dir} -maxdepth 1 ! -type d | xargs basename -a
219}
220
221########################## BUSINESS LOGIC ##########################
222
223function get_dataset_files {
224 local skip
225 skip=0
226 if [ $# -gt 0 ] ; then
227 skip=${1}
228 fi
229
230 find . -maxdepth 1 -iname "${data_mask}*" | (
231 local file
232 while read file ; do
233 basename ${file}
234 done
235 ) | LC_ALL=C sort | (
236 local file
237 while read file ; do
238 if [ ${skip} -eq 0 ] ; then
239 basename ${file}
240 else
241 skip=$((skip - 1))
242 fi
243 done
244 )
245}
246
247function count_ds_files {
248 local file
249 local count=0
250 while read file ; do
251 if [ -n "${file}" ] ; then
252 count=$((count + 1))
253 fi
254 done << __END__
255$(get_dataset_files)
256__END__
257 echo ${count}
258}
259
260function process_with_function {
261 local func=${1}
262 shift 1
263 local count="${1}"
264
265 local modules=( $(get_modules) )
266
267 declare -A next_ids=( )
268
269 local module
270 for module in ${modules[@]} ; do
271 next_ids[${module}]=0
272 done
273
274 local c_start=$(date +%s)
275 local idx=0
276 local input
277 while read input ; do
278 for module in ${modules[@]} ; do
279 clearLastLine
280 local eta="$(remaining $((${idx} + 1)) ${count} ${c_start})"
281 printf "%13s: [%4s/%4s - %6.2f%% - %15s] %s with %s " "${func}" "$((idx + 1))" "${count}" "$(echo 100 \* $((idx + 1)) / ${count} | bc -l)" "${eta}" "${input}" "${module}"
282 if ! mod_invoke ${module} ${func} "${input}" "${idx}" "${next_ids[${module}]}" ; then
283 printf "!!! Issues in executing %s step for module %s on file %s\n" "${func}" "${module}" "${input}" >&2
284 fi
285 next_ids[${module}]=$(id_receive)
286 done
287 idx=$((idx + 1))
288 done << __END__
289$(get_dataset_files)
290__END__
291 printf "\n"
292}
293
294function get_checksum {
295 local file_name="${1}"
296 local checksum="$(md5sum -b ${file_name} | cut -d ' ' -f 1)"
297 printf "%s" "${checksum}"
298}
299
300function fill_file_table {
301 local db="${1}"
302 shift 1
303 local count="${1}"
304
305 local files_table="log#files"
306 (
307 printf "%d\n" "${count}"
308 get_dataset_files
309 ) | (
310 local cur_file
311 local count
312 local idx=0
313 read count
314 while read cur_file ; do
315 local checksum_file="check#${cur_file}"
316 local size_file="size#${cur_file}"
317 local check
318 local size
319
320 if [ ! -e "${checksum_file}" ] ; then
321 check="$(get_checksum ${cur_file})"
322 printf -- "%s\n" "${check}" > ${checksum_file}
323 else
324 check="$(cat ${checksum_file})"
325 fi
326
327 if [ ! -e "${size_file}" ] ; then
328 size="$(stat -L --format '%s' ${cur_file})"
329 printf -- "%d\n" "${size}" > "${size_file}"
330 else
331 size="$(cat ${size_file})"
332 fi
333
334 clearLastLine >&2
335 printf "%13s: [%4s/%4s - %6.2f%%]" "file import" "$((idx + 1))" "${count}" "$(echo 100 \* $((idx + 1)) / ${count} | bc -l)" >&2
336 idx=$((idx + 1))
337
338 printf "%s,%d,%s\n" "${cur_file}" "${size}" "${check}"
339 done
340 ) > ${files_table}
341 sqlite3 "${db}" << __END__
342pragma foreign_keys = on;
343
344create table files(
345 file_name string not null primary key,
346 file_size integer not null,
347 checksum string not null
348);
349create unique index files_idx on files(file_name);
350.separator ','
351.import '${files_table}' files
352__END__
353 printf "\n" >&2
354}
355
356########################## DS PROCESS ##########################
357function main_func {
358 local dataset=$(basename ${PWD})
359 local ds_name=${dataset##ds_}
360 local result_dir="${1}"
361 shift 1
362 local proc_list=( $@ )
363
364 printf "=== %s\n--- Processors: %d\n--- Processor list: %s\n" "${ds_name}" "${#proc_list[@]}" "${proc_list[*]}"
365
366 # Load all modules
367 if [ ! -d "${module_dir}" ] ; then
368 echo "FAIL: no module directory ${module_dir} found in ${dataset}"
369 exit 1
370 fi
371
372 local modules=( $(get_modules) )
373
374 local module
375 for module in ${modules[@]} ; do
376 source "${module_dir}/${module}"
377 mod_invoke ${module} init
378 done
379
380 # Check if we have all required tools
381 local tools_needed=()
382 for module in ${modules[@]} ; do
383 tools_needed+=( $(mod_invoke ${module} prereq) )
384 done
385
386 local tool
387 for tool in ${tools_needed[@]} ; do
388 if [ ! -e ${tool} ] ; then
389 echo "ERROR: ${tool} not found in ${dataset}"
390 exit 1
391 fi
392 done
393
394 local file_count="$(count_ds_files)"
395 if [ "${file_count}" -eq 0 ] ; then
396 echo "WARNING: no data found in ${ds_name}"
397 return
398 fi
399
400 initialize ${proc_list[@]}
401 pb_init
402
403 # Preprocess all files
404 if ! ${skip_preprocess} ; then
405 process_with_function preprocess "${file_count}"
406 barrier
407 fi
408
409
410 # Build indexes
411 if ! ${skip_build} ; then
412 process_with_function build "${file_count}"
413 barrier
414 fi
415
416 # Post-process
417 if ! ${skip_postprocess} ; then
418 process_with_function postprocess "${file_count}"
419 barrier
420 fi
421
422 # Benchmark
423 if ! ${skip_benchmark} ; then
424 process_with_function benchmark "${file_count}"
425 fi
426
427 # Shuts down parallel facilities
428 pb_terminate
429 terminate
430
431 # Finalize
432 for module in ${modules[@]} ; do
433 mod_invoke "${module}" finalize
434 done
435
436 if ${skip_db} ; then
437 return 0
438 fi
439
440 # Initialize DB
441 printf "%10s\n" "DB Import"
442 local db="${result_dir}/results_${ds_name}.db"
443 rm -f "${db}"
444 fill_file_table "${db}" "${file_count}"
445
446 # Import data into DB for all modules
447 for module in ${modules[@]} ; do
448 mod_invoke ${module} "db_definitions" | sqlite3 "${db}"
449 done
450
451 for module in ${modules[@]} ; do
452 mod_invoke ${module} "db_import" | sqlite3 "${db}"
453 done
454}
455
456########################## MAIN ##########################
457skip_preprocess=false
458skip_build=false
459skip_postprocess=false
460skip_benchmark=false
461skip_db=false
462
463while getopts "pbPkd" p; do
464 case "${p}" in
465 p) skip_preprocess=true;;
466 b) skip_build=true;;
467 P) skip_postprocess=true;;
468 k) skip_benchmark=true;;
469 d) skip_db=true;;
470 *) printf "ERROR: invalid option %s\n" ${p} >&2
471 exit 1
472 esac
473done
474shift $((OPTIND - 1))
475
476if ! ${skip_preprocess} ; then
477 printf -- "--- %13s enabled\n" "Pre-process"
478fi
479if ! ${skip_build} ; then
480 printf -- "--- %13s enabled\n" "Build"
481fi
482if ! ${skip_postprocess} ; then
483 printf -- "--- %13s enabled\n" "Post-process"
484fi
485if ! ${skip_benchmark} ; then
486 printf -- "--- %13s enabled\n" "Benchmark"
487fi
488
489if ! ${skip_db} ; then
490 printf -- "--- %13s enabled\n" "DB import"
491fi
492
493if [ $# -lt 1 ] ; then
494 echo "ERROR: Result directory needed"
495 exit 1
496fi
497
498data_mask="data_"
499module_dir="modules"
500result_dir=$(readlink -f "${1}")
501
502if [ \( -e "${result_dir}" \) -a \( ! -d ${result_dir} \) ] ; then
503 echo "ERROR: ${result_dir} exists and it's not a directory"
504 exit 1
505fi
506
507mkdir -p "${result_dir}"
508
509datasets=( $(find . -mindepth 1 -maxdepth 1 -type d -iname 'ds_*' | xargs basename -a | LC_ALL=C sort ) )
510proc_list_file="processors"
511
512for dataset in ${datasets[@]} ; do
513 pushd ${dataset} &> /dev/null
514 (
515 if [ ! -e "${proc_list_file}" ] ; then
516 printf "=== Dataset %s: processor list file %s not found, skip\n" "${dataset}" "${proc_list_file}"
517 exit
518 fi
519 proc_list=( $(cat ${proc_list_file} ) )
520 main_func "${result_dir}" ${proc_list[@]}
521 )
522 popd &> /dev/null
523done