#!/usr/bin/env bash
#
# bbq - Simple Bash message queues using Linux named pipes.
#
# bbq.sh provides a few convenient Bash functions for creating and working with
# Linux named pipes (FIFO). Its primary use case is to set up multiple worker
# processes getting work from a single job queue in your shell script.
#
# Please see the examples/basic script for an example usage of bbq.sh.
#
# Requirements:
# - Linux, Bash, GNU coreutils, and flock from util-linux.
#
if [[ ${bbq_SOURCED:-} ]]; then
return 0
fi
bbq_error () { echo "$@" >&2; }
bbq_debug () { if [[ ${bbq_DEBUG:-} ]]; then echo "$@" >&2; fi; }
if [[ $BASH_SOURCE == "$0" ]]; then
bbq_error "Please source bbq.sh instead of running it!"
exit 1
fi
# Set it to enable some debugging outputs.
bbq_DEBUG=
# The default number of worker processes to run by bbq-start
bbq_WORKER_COUNT=4
# The named pipe to be used by default to serve as a message queue.
bbq_FIFO=
# Each "job" in bbq is a ${bbq_CHUNK_SIZE}-byte chunk of bash commands
# written to $bbq_FIFO that is used as a message queue.
#
bbq_CHUNK_SIZE=1024 # bytes; on Linux the limit is 4k.
# Create a named pipe (FIFO) and set it as the default FIFO ($bbq_FIFO).
#
# If an argument is provided it's taken to be the path of the FIFO to be
# created; otherwise, a random file name will be chosen for the FIFO in the
# current directory.
#
bbq-new () { # [queue]
local queue=${1:-}
[[ $queue ]] || queue=bbq-$$-$RANDOM
[[ $queue == /* ]] || queue=$PWD/$queue
mkfifo -m 0600 "$queue" || return $?
bbq_FIFO=$queue
}
# Start worker processes in the background to process the queue and wait for
# them to end.
#
# Arguments:
#
# queue - Optional. Path to the named pipe for the workers to read
# messages/commands from. If ommitted then the queue is
# taken to be $bbq_FIFO, and in which case, it will also be
# deleted at the end.
#
# Options:
#
# -w COUNT - Number of worker processes to create that work on the queue.
# Default is 4 workers.
#
bbq-start () { # [-w COUNT] [queue]
(declare -A workers # pid -> exit code
bbq_owns_the_pipe=
trap '
kill ${!workers[*]} >/dev/null 2>&1 || true
if [[ $bbq_owns_the_pipe ]]; then rm -f "$bbq_FIFO"; fi
' EXIT
_bbq_start_workers "$@"
) &
}
# Enqueue arbitrary Bash commands as a fixed length string into the $bbq_FIFO.
#
# Arguments:
#
# command - Arbitrary Bash command to add to the queue.
#
# queue - Optional. If ommitted, $bbq_FIFO is assumed; otherwise, it
# should be a path to a message queue created by bbq-new.
#
# This command MUST ONLY be run after bbq-start.
#
# NOTE: This command may block on writing to the pipe if the pipe buffer is full.
# Therefore, it's recommended that you run it as the last part of your
# script. Alternatively, you can also run it in the background (i.e.,
# with '&') directly or indirectly to avoid blocking the flow of your
# script.
#
bbq () { # <command> [queue]
local code=$1 queue=${2:-$bbq_FIFO}
if (( ${#code} > bbq_CHUNK_SIZE )); then
bbq_error "Encoded message exceeds allowed chunk size ($bbq_CHUNK_SIZE): $code"
return 1
fi
[[ -p $queue ]] || { bbq_error "'$queue' must be a named pipe!"; return 1; }
# Accoridng to docs and google, on linux, read/write less than PIPE_BUF (4k
# bytes) on a FIFO is atomic. So, we don't need to lock before writes.
#
# See also 'man fifo' and 'man 7 pipe'.
#
printf "%-${bbq_CHUNK_SIZE}s" "${code:0:$bbq_CHUNK_SIZE}" >"$queue"
}
# Internal implementation for bbq-start()
#
_bbq_start_workers () { # [-w COUNT] [queue]
local option; OPTIND=1
while getopts ':w:' option "$@"; do
case $option in
w) bbq_WORKER_COUNT=$OPTARG ;;
:) bbq_error "$FUNCNAME: Missing option argument for -$OPTARG"; return 1 ;;
\?) bbq_error "$FUNCNAME: Unknown option: -$OPTARG"; return 1 ;;
esac
done
shift $((OPTIND - 1))
bbq_WORKER_COUNT=${bbq_WORKER_COUNT:-4}
printf "%d" "$bbq_WORKER_COUNT" >/dev/null 2>&1 \
&& (( bbq_WORKER_COUNT > 0 )) || {
bbq_error "Worker count should be > 0"
return 1
}
local queue=${1:-}
if [[ $queue ]]; then
[[ $queue == /* ]] || queue=$PWD/$queue
bbq_FIFO=$queue
else
bbq_owns_the_pipe=1
fi
[[ -p ${bbq_FIFO:?required} ]] || {
bbq_error "$bbq_FIFO must be a named pipe!"
return 1
}
# Fork the workers to do work.
# This needs to be done before we can enqueue anything without blocking.
#
local i
for ((i=0; i < $bbq_WORKER_COUNT; i++)); do
_bbq_worker & workers[$!]=
done
# Keep a write FD open to the FIFO to prevent the read ends from
# getting EOFs, which happens when all write FDs are closed.
#
local write_fd; exec {write_fd}>"$bbq_FIFO" || return $?
# Wait for the worker processes to exit
local pid failed=
for pid in ${!workers[*]}; do
wait $pid && workers[$pid]=0 || {
workers[$pid]=$?
bbq_error "Worker $pid exited! (rc=${workers[$pid]})"
failed=1
}
done
[[ ! $failed ]]
}
# Represents a background worker sub process that takes commands from FIFO.
#
_bbq_worker () {
local read_fd
exec {read_fd}<"${bbq_FIFO:?required}" || return $?
local code
_bbq_pop () {
local rc
flock $read_fd || return $?
read -u $read_fd -rN "$bbq_CHUNK_SIZE" code || rc=$?
flock -u $read_fd || return $?
return $rc
}
while true; do
code=; _bbq_pop || {
bbq_debug "Worker $BASHPID: Failed dequeuing! code=$code"
continue
}
eval "$code"
# NOTE:
# - If you 'set -e' then the worker could die due to a non-zero
# exit from evaluating $code.
# - If all your workers exited (i.e., no more FDs reading on the
# pipe) then the enqueue operation will either block on opening the write end,
# or you get a SIGPIPE when writing to the pipe.
done
bbq_debug "Worker $BASHPID quit"
eval "exec $read_fd>&-"
}
bbq_SOURCED=1