Cretae smaller chunk lists

lumi_monitor/dst_lumi_watcher.py

@@ -58,9 +58,10 @@ def make_abs(path, wd):
 WD=os.path.dirname(os.path.realpath(__file__))
 parse_shell_config(make_abs('config.sh', WD))
 
-
-ABS_LUMI_DST_LIST=make_abs(DATA_DIR + '/' + LUMI_DST_LIST, WD)
+ABS_DATA_DIR=make_abs(DATA_DIR, WD)
 ABS_DST_DIR=make_abs(DATA_DIR + '/' + DST_DIR, WD)
+ABS_LUMI_DST_LIST=make_abs(DATA_DIR + '/' + LUMI_DST_LIST, WD)
+ABS_LUMI_SUBMIT_LIST=make_abs(DATA_DIR + '/' + LUMI_SUBMIT_LIST, WD)
 
 full_name = make_abs(ABS_DST_DIR + '/' + DST_MASK, WD)
 re_full_name = wildcard_to_regex(full_name, decorate=True)
@@ -120,6 +121,21 @@ class EventHandler(pyinotify.ProcessEvent):
 def on_loop(notifier):
     pass
 
+def make_temp_lists(file_list, split_size = 5000):
+    """
+    Get list of files, split into chunks,
+    create tmp files, fill and return paths.
+    """
+    chunked_list = [file_list[i:i+split_size] for i in range(0, len(file_list), split_size)]
+    tmp_files = []
+    for l in chunked_list:
+        # file for new files to submit
+        tmp_list_file = tempfile.mktemp(dir=ABS_DATA_DIR, prefix='.lumi_tmp_', suffix='.txt')
+        write_list(tmp_list_file, l)
+        tmp_files.append(tmp_list_file)
+    return tmp_files
+
+
 # TODO make also for other critical files, as function
 dirname = os.path.dirname(ABS_LUMI_DST_LIST)
 try:
@@ -146,20 +162,25 @@ if __name__ == "__main__":
         write_list(ABS_LUMI_DST_LIST, new_files_list)
 
         if len(new_files_list):
-            tmp_list_file = tempfile.mktemp(dir=DATA_DIR,   # file for new files to submit
-                                            prefix='.lumi_tmp_',
-                                            suffix='.txt')
-            write_list(tmp_list_file, new_files_list)
-
-            print(f"Write new list for submission: {tmp_list_file}")
-            submit_file(tmp_list_file)
+            tmp_files = make_temp_lists(new_files_list)
+            for f in tmp_files:
+                print(f"Created new list for submission: {f}")
+                submit_file(f)
         else:
             print("No new files to submit")
 
     elif args.submit:
-        l = get_file_from_list(ABS_LUMI_DST_LIST)
-        for ll in l:
-            submit_file(ll)
+        all_files = get_list_from_file(ABS_LUMI_DST_LIST)
+        submitted = get_list_from_file(ABS_LUMI_SUBMIT_LIST)
+        new_files_list = []
+        for ll in all_files:
+            if ll not in submitted:
+                new_files_list.append(ll)
+        tmp_files = make_temp_lists(new_files_list)
+        for f in tmp_files:
+            print(f"Created new list for submission: {f}")
+            submit_file(f)
+
     else:
         wm = pyinotify.WatchManager()
         handler = EventHandler()

lumi_monitor/lumi_batch_script.sh

@@ -26,8 +26,9 @@ ofile=$ldir/$(basename $file .root).lumi
 function update_with_lock {
     args=( "$@" )
     while ! flock -n -x $1 "${args[@]:1}"; do
-        sleep $(($RANDOM % 10))
-        echo Retrying flock for $1
+        t=$(($RANDOM % 100))
+        echo Retrying flock for $1 in $t
+        sleep $t
     done
 }
 

lumi_monitor/submit_lumi_job.sh

@@ -49,12 +49,12 @@ ABS_LUMI_DIR=$ABS_DATA_DIR/$LUMI_DIR
 mkdir $ABS_LUMI_DIR -p
 
 # submit file, containing directory must exists
-ABS_LUMI_SUBMIT_FILE=$ABS_DATA_DIR/$LUMI_SUBMIT_FILE
-mkdir $(dirname $ABS_LUMI_SUBMIT_FILE) -p
-[ ! -e $ABS_LUMI_SUBMIT_FILE ] && touch $ABS_LUMI_SUBMIT_FILE
+ABS_LUMI_SUBMIT_LIST=$ABS_DATA_DIR/$LUMI_SUBMIT_LIST
+mkdir $(dirname $ABS_LUMI_SUBMIT_LIST) -p
+[ ! -e $ABS_LUMI_SUBMIT_LIST ] && touch $ABS_LUMI_SUBMIT_LIST
 
-ABS_LUMI_STATUS_FILE=$ABS_DATA_DIR/$LUMI_STATUS_FILE
-mkdir $(dirname $ABS_LUMI_STATUS_FILE) -p
+ABS_LUMI_STATUS_LIST=$ABS_DATA_DIR/$LUMI_STATUS_LIST
+mkdir $(dirname $ABS_LUMI_STATUS_LIST) -p
 
 is_array=
 array_string=
@@ -64,7 +64,7 @@ if [[ $1 == *.txt ]]; then
     is_array=1
     array_string="--array=1-$(cat $1 | wc -l)"
 else
-    grep -q "^$1" $ABS_LUMI_SUBMIT_FILE
+    grep -q "^$1" $ABS_LUMI_SUBMIT_LIST
     is_submitted=$?
 fi
 
@@ -78,9 +78,9 @@ if [[ $is_array -eq 1 || ! $is_submitted -eq 1 || $FORCE_SUBMIT -eq 1 ]]; then
     sbatch -o $ABS_DATA_DIR/slurm_log/$LUMI_LOG_PREFIX-%j.log \
         ${array_string} \
         --time=60:00 --mem-per-cpu=100mb -p main -J L$(basename $1 .root) \
-        --export="pattern=$1,events=$DST_EVENTS,lumitool=$ABS_LUMI_TOOL,ldir=$ABS_LUMI_DIR,sfile=$ABS_LUMI_STATUS_FILE" \
+        --export="pattern=$1,events=$DST_EVENTS,lumitool=$ABS_LUMI_TOOL,ldir=$ABS_LUMI_DIR,sfile=$ABS_LUMI_STATUS_LIST" \
         -- $WD/wrap.sh $ABS_LUMI_BATCH_SCRIPT \
     && \
-        flock -x $ABS_LUMI_SUBMIT_FILE echo "$1   $(date)" >> $ABS_LUMI_SUBMIT_FILE
+        flock -x $ABS_LUMI_SUBMIT_LIST echo "$1" >> $ABS_LUMI_SUBMIT_LIST
         # add dst file to list only if not add before
 fi

online_dst/config.sh.example

@@ -7,8 +7,8 @@ HLD_WATCH_LOG_FILE=/tmp/.hld_watcher.log
 HLD_LIST_FILE=hld_list.txt
 HLD_DIR=/lustre/hades/raw/feb22/22
 HLD_MASK=*/be*.hld
-DST_STATUS_FILE=hld_dst_status.txt
-DST_SUBMIT_FILE=hld_dst_submit.txt
+DST_STATUS_LIST=hld_dst_status.txt
+DST_SUBMIT_LIST=hld_dst_submit.txt
 DST_DIR=out
 DST_EVENTS=1000000
 DST_LOG_PREFIX=dst_slurm
@@ -23,8 +23,8 @@ LUMI_WATCH_PID_FILE=/tmp/.dst_lumi_watcher.pid
 LUMI_WATCH_LOG_FILE=/tmp/.dst_lumi_watcher.log
 LUMI_SUBMIT_SCRIPT=submit_lumi_job.sh
 LUMI_BATCH_SCRIPT=lumi_batch_script.sh
-LUMI_SUBMIT_FILE=dst_lumi_submit.txt
-LUMI_STATUS_FILE=dst_lumi_status.txt
+LUMI_SUBMIT_LIST=dst_lumi_submit.txt
+LUMI_STATUS_LIST=dst_lumi_status.txt
 LUMI_LOG_PREFIX=dst_lumi_slurm
 LUMI_TOOL=lumi_monitor
 LUMI_DIR=lumi

online_dst/dst_batch_script.sh

@@ -35,8 +35,9 @@ fi
 function update_with_lock {
     args=( "$@" )
     while ! flock -n -x $1 "${args[@]:1}"; do
-        sleep $(($RANDOM % 10))
-        echo Retrying flock for $1
+        t=$(($RANDOM % 100))
+        echo Retrying flock for $1 in $t
+        sleep $t
     done
 }
 

online_dst/rerun_broken_dst.sh

@@ -44,13 +44,13 @@ make_abs "$DST_CHECK_SCRIPT" "$WD" ABS_DST_CHECK_SCRIPT
 [[ -z $DST_CHECK_SCRIPT ]] || [[ -f $ABS_DST_CHECK_SCRIPT ]] || exit 3
 
 # status file must be set if the check script is set
-[[ -z $HLD_CHECK_SCRIPT ]] || [[ -n $DST_STATUS_FILE ]] || exit 4
-ABS_DST_STATUS_FILE=$ABS_DATA_DIR/$DST_STATUS_FILE
-mkdir $(dirname $ABS_DST_STATUS_FILE) -p
+[[ -z $HLD_CHECK_SCRIPT ]] || [[ -n $DST_STATUS_LIST ]] || exit 4
+ABS_DST_STATUS_LIST=$ABS_DATA_DIR/$DST_STATUS_LIST
+mkdir $(dirname $ABS_DST_STATUS_LIST) -p
 
 REGEX="^(.*\/[a-z]{2}[0-9]{13}\.hld).*$"
 
-echo "Checking in ${ABS_DST_STATUS_FILE}"
+echo "Checking in ${ABS_DST_STATUS_LIST}"
 
 while IFS= read -r line; do
     stringarray=($line)
@@ -79,12 +79,12 @@ while IFS= read -r line; do
         status=$?
         echo "  => $status"
 
-        if grep -q "^$dst" $ABS_DST_STATUS_FILE; then
-            $PRETEND sed -e "s|$dst .\(\ .*\)\?|$dst $status $source|" -i $ABS_DST_STATUS_FILE
+        if grep -q "^$dst" $ABS_DST_STATUS_LIST; then
+            $PRETEND sed -e "s|$dst .\(\ .*\)\?|$dst $status $source|" -i $ABS_DST_STATUS_LIST
         else
-            $PRETEND echo "$dst $status $source" >> $ABS_DST_STATUS_FILE
+            $PRETEND echo "$dst $status $source" >> $ABS_DST_STATUS_LIST
         fi
-        echo "  ==> update $ABS_DST_STATUS_FILE status $?"
+        echo "  ==> update $ABS_DST_STATUS_LIST status $?"
 
     else
         if [[ "$status" -ne "0" ]]; then
@@ -93,4 +93,4 @@ while IFS= read -r line; do
         fi
     fi
 
-done < $ABS_DST_STATUS_FILE
+done < $ABS_DST_STATUS_LIST

online_dst/submit_dst_job.sh

@@ -44,9 +44,9 @@ make_abs "$DST_CHECK_SCRIPT" "$WD" ABS_DST_CHECK_SCRIPT
 [[ -z $DST_CHECK_SCRIPT ]] || [[ -f $ABS_DST_CHECK_SCRIPT ]] || exit 3
 
 # status file must be set if the check script is set
-[[ -z $HLD_CHECK_SCRIPT ]] || [[ -n $DST_STATUS_FILE ]] || exit 4
-ABS_DST_STATUS_FILE=$ABS_DATA_DIR/$DST_STATUS_FILE
-mkdir $(dirname $ABS_DST_STATUS_FILE) -p
+[[ -z $HLD_CHECK_SCRIPT ]] || [[ -n $DST_STATUS_LIST ]] || exit 4
+ABS_DST_STATUS_LIST=$ABS_DATA_DIR/$DST_STATUS_LIST
+mkdir $(dirname $ABS_DST_STATUS_LIST) -p
 
 # dst output dir must be defined
 [[ -n $DST_DIR ]] || exit 5
@@ -54,9 +54,9 @@ ABS_DST_DIR=$ABS_DATA_DIR/$DST_DIR
 mkdir $ABS_DST_DIR -p
 
 # submit file, containing directory must exists
-ABS_DST_SUBMIT_FILE=$ABS_DATA_DIR/$DST_SUBMIT_FILE
-mkdir $(dirname $ABS_DST_SUBMIT_FILE) -p
-[ ! -e $ABS_DST_SUBMIT_FILE ] && touch $ABS_DST_SUBMIT_FILE
+ABS_DST_SUBMIT_LIST=$ABS_DATA_DIR/$DST_SUBMIT_LIST
+mkdir $(dirname $ABS_DST_SUBMIT_LIST) -p
+[ ! -e $ABS_DST_SUBMIT_LIST ] && touch $ABS_DST_SUBMIT_LIST
 
 is_array=
 array_string=
@@ -66,7 +66,7 @@ if [[ $1 == *.txt ]]; then
     is_array=1
     array_string="--array=1-$(cat $1 | wc -l)"
 else
-    grep -q "^$1" $ABS_DST_SUBMIT_FILE
+    grep -q "^$1" $ABS_DST_SUBMIT_LIST
     is_submitted=$?
 fi
 
@@ -80,9 +80,9 @@ if [[ $is_array -eq 1 || ! $is_submitted -eq 1 || $FORCE_SUBMIT -eq 1 ]]; then
     sbatch -o $ABS_DATA_DIR/slurm_log/$DST_LOG_PREFIX-%j.log \
         ${array_string} \
         --time=3:00:00 --mem-per-cpu=1000mb -p main -J D$(basename $1 .hld) \
-        --export="pattern=$1,events=$DST_EVENTS,odir=$ABS_DST_DIR,checktool=$ABS_DST_CHECK_SCRIPT,sfile=$ABS_DST_STATUS_FILE" \
+        --export="pattern=$1,events=$DST_EVENTS,odir=$ABS_DST_DIR,checktool=$ABS_DST_CHECK_SCRIPT,sfile=$ABS_DST_STATUS_LIST" \
         -- $WD/wrap.sh $ABS_DST_BATCH_SCRIPT    \
     && \
-        flock -x $ABS_DST_SUBMIT_FILE echo "$1   $(date)" >> $ABS_DST_SUBMIT_FILE
+        flock -x $ABS_DST_SUBMIT_LIST echo "$1" >> $ABS_DST_SUBMIT_LIST
         # add hld file to list only if not add before
 fi