Parallel queue processing in shell (bash)

add_task_lock.sh

#!/usr/bin/env bash
SCRIPT_ROOT=$(cd "$(dirname "$0")"; pwd)

WORKER_ID=$1
QUEUE_ITEM=$2
LOCK_FILE=$3

if [[ ! -f "$LOCK_FILE" ]]; then
    echo "$WORKER_ID" > "$LOCK_FILE"
    # echo "[$(date)] Worker $WORKER_ID: Acquired lock for queue item $QUEUE_ITEM."
fi

demo-start.sh

#!/usr/bin/env bash
SCRIPT_ROOT=$(cd "$(dirname "$0")"; pwd)

QUEUE_FILE="$SCRIPT_ROOT/queue.txt"
QUEUE_WORKER_SIZE=3
QUEUE_ITEM_LOCK_DIR="$SCRIPT_ROOT/queue_item_locks.d"

rm -rf "$QUEUE_FILE"

i=0

while [ $i -lt 10 ]; do
    echo "Added iteration $i to queue."
    ((i++))
    echo "$i" >> "$QUEUE_FILE"
done

"$SCRIPT_ROOT/start_queue.sh" \
    "$QUEUE_FILE" \
    "$QUEUE_WORKER_SIZE" \
    "$QUEUE_ITEM_LOCK_DIR" \
    "$SCRIPT_ROOT/demo-worker.sh"

echo "Done";
rm -rf "$QUEUE_FILE"

demo-worker.sh

#!/usr/bin/env bash
SCRIPT_ROOT=$(cd "$(dirname "$0")"; pwd)

WORKER_ID=$1
QUEUE_ITEM=$2

sleep $QUEUE_ITEM

queue_worker.sh

#!/usr/bin/env bash
SCRIPT_ROOT=$(cd "$(dirname "$0")"; pwd)

WORKER_ID=$1
QUEUE_FILE=$2
QUEUE_ITEM_LOCK_DIR=$3
TASK_WORKER_SCRIPT=$4

mkdir -p "$QUEUE_ITEM_LOCK_DIR"

for QUEUE_ITEM in $(cat "$QUEUE_FILE"); do

    RUNNER_LOCK_FILE="$QUEUE_ITEM_LOCK_DIR/task.lock"
    TASK_LOCK_FILE="$QUEUE_ITEM_LOCK_DIR/$QUEUE_ITEM.lock"

    if ! flock "$RUNNER_LOCK_FILE" -c "$SCRIPT_ROOT/add_task_lock.sh $WORKER_ID $QUEUE_ITEM $TASK_LOCK_FILE"; then
        echo "[$(date)] Worker $WORKER_ID: Could not acquire lock for queue item $QUEUE_ITEM. Skipping."
        continue
    fi

    # echo "[$(date)] Worker $WORKER_ID: Read worker id $(cat "$TASK_LOCK_FILE") for item $QUEUE_ITEM."
    
    if [[ "$WORKER_ID" != "$(cat "$TASK_LOCK_FILE")" ]]; then
        continue
    fi

    echo "[$(date)] Worker $WORKER_ID: Processing item $QUEUE_ITEM."
    $TASK_WORKER_SCRIPT "$WORKER_ID" "$QUEUE_ITEM"
    # echo "[$(date)] Worker $WORKER_ID: Finished processing item $QUEUE_ITEM."
done

start_queue.sh

#!/usr/bin/env bash
SCRIPT_ROOT=$(cd "$(dirname "$0")"; pwd)

QUEUE_FILE=$1
QUEUE_WORKER_SIZE=$2
QUEUE_ITEM_LOCK_DIR=$3
TASK_WORKER_SCRIPT=$4

mkdir -p "$QUEUE_ITEM_LOCK_DIR"

QUEUE_WORKER_ID=0
while [ $QUEUE_WORKER_ID -lt "$QUEUE_WORKER_SIZE" ]; do
    "$SCRIPT_ROOT/queue_worker.sh" "$QUEUE_WORKER_ID" "$QUEUE_FILE" "$QUEUE_ITEM_LOCK_DIR" "$TASK_WORKER_SCRIPT" &
    ((QUEUE_WORKER_ID++))
done

wait

echo "All workers have completed processing the queue."

rm -rf "$QUEUE_ITEM_LOCK_DIR"