bash job pool for // execution

gistfile1.txt

#!/bin/bash 

debug=0;

## Clean up working temp files
function clean_up {
  ((debug)) || rm -rf "${path_to_file}"
}

trap clean_up EXIT

## Create and move to a temporary working dir
path_to_file="$(mktemp -d)/"
cd "${path_to_file}"

## Get couch view and split output in n files containing 1000 curl fomarted user for insert 

## Coreutils "split" with --filter : 

#mkdir -p /tmp/coreutils
#cd /tmp/coreutils
#wget http://ftp.gnu.org/gnu/coreutils/coreutils-8.25.tar.xz
#yum install xz -y
#xz -d coreutils-8.25.tar.xz
#tar xvf coreutils-8.25.tar
#cd coreutils-8.25
#mkdir -p /tmp/coreutils/coreutils-8.25/output
#./configure --prefix /tmp/coreutils/coreutils-8.25/output
#make 
#make install 
#alias split="/tmp/coreutils/coreutils-8.25/output/bin/split"
 
unset http_proxy
historyDepth=2
from=$(date -d $(date -d"-${historyDepth} day" +"%Y-%m-%d") +%s000)
historyDepth=0
until=$(date -d $(date -d"-${historyDepth} day" +"%Y-%m-%d") +%s000)
skip=0
#limit="&limit=100000"
limit=""
esEndPoint="http://couch1.zebestof.com:8092/dormant/_design/pools/_view/pools?stale=ok&startkey=${from}&endkey=${until}&inclusive_end=false&reduce=false&connection_timeout=60000&skip=${skip}${limit}"

curl -sSN ${esEndPoint} \
  | mawk 'NR!=1{print substr($0, 1, length($0)-2)}' \
  | 2>/dev/null /tmp/coreutils/coreutils-8.25/output/bin/split -u -e -l 1000 -a 6 --filter='jq -r -c --unbuffered "\"--next \\n--url http://127.0.0.1:9200/pool/adv/\" + .id + \"?ttl=30d \\n--data \" + (.value | tostring) + \"\\n--write-out \\\"\\\n\\\" \" " - > $FILE.curl' -

## Set the max number of // injectors
max=8;
running=();

debug() {
  ((debug)) && echo "$(date): $@"
}

## Update the running job PIDs list
update_job_pool() {
  for pid in ${running[@]}; do
    [[ -d "/proc/${pid}" ]] || { running=( ${running[@]/${pid}/} ) && debug "Freeing one slot in the pool"; }
  done
}

## ForEach file in working temp dir ...
files=($(ls -1 "${path_to_file}"))
for i in ${!files[@]}; do

  file="${path_to_file}${files[$i]}"

  # Wait for an empty slot into the job pool
  while [[ ${#running[@]} -ge ${max} ]]; do
    debug "${#running[@]}/${max} Waiting for an empty slot in the pool";
    sleep 0.3;
    update_job_pool;
  done

  #sleep $(shuf -i 1-5 -n 1) && echo "Job DONE | ${file}" &

  # Inject infot elasticSeach from a sub shell 
  curl -sSK "${file}" && rm -rf "${file}" &

  pid="$!"
  # Add new subShell PID into the running job pool
  running=(${running[@]} ${pid})

  debug "${#running[@]}/${max} New Job ${pid} started | ${file}"

done \
  | pv -N "curl OUTPUT" -l -W -c >/dev/null 

## Wait for all jobs to be done 
while [[ ${#running[@]} -gt 0 ]]; do
  debug "${#running[@]}/${max} Waiting for all the job to be DONE"
  sleep 0.3;
  update_job_pool;
done

debug "${#running[@]}/${max} All jobs DONE"