fpsync

#!/bin/sh

# Copyright (c) 2014-2024 Ganael LAPLANCHE <ganael.laplanche@martymac.org>
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions
# are met:
# 1. Redistributions of source code must retain the above copyright
#    notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
#    notice, this list of conditions and the following disclaimer in the
#    documentation and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE AUTHORS AND CONTRIBUTORS ``AS IS'' AND
# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHORS OR CONTRIBUTORS BE LIABLE
# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
# OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
# HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
# OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
# SUCH DAMAGE.

# This script is a simple wrapper showing how fpart can be used to migrate data.
# It uses fpart and a copy tool to spawn multiple instances to migrate data from
# src_dir/ to dst_url/. Jobs can execute either locally or over SSH.

FPSYNC_VERSION="1.6.1"

########## Default values for options

# External tool used to copy files
OPT_TOOL_NAME="rsync"
# External tool path
OPT_TOOL_PATH=""
# Number of sync jobs to run in parallel ("workers", -n)
OPT_JOBS=2
# Same, but autodetected
#OPT_JOBS=$(sysctl -n hw.ncpu)  # On FreeBSD
#OPT_JOBS=$(nproc)              # On Linux
# Maximum files or directories per sync job (-f)
OPT_FPMAXPARTFILES="2000"
# Maximum bytes per sync job (-s)
OPT_FPMAXPARTSIZE="$((4 * 1024 * 1024 * 1024))" # 4 GB
# Work on a per-directory basis (disabled by default)
OPT_DIRSONLY=""
# Pack erroneous dirs apart and enable recursive rsync
OPT_AGGRESSIVE=""
# SSH workers (execute jobs locally if not defined, -w)
OPT_WRKRS=""
# Shared dir (must be shared amongst all workers, -d)
OPT_SHDIR=""
# Temporary dir (local, used for queue management, -t)
OPT_TMPDIR="/tmp/fpsync"
# E-mail report option (-M)
OPT_MAIL=""
# Prepare mode (-p)
OPT_PREPARERUN=""
# List runs (-l)
OPT_LISTRUNS=""
# Run ID for resume mode (-r)
OPT_RUNID=""
# Replay mode (-R)
OPT_REPLAYRUN=""
# Archive run (-a)
OPT_ARCHIVERUN=""
# Delete run (-D)
OPT_DELETERUN=""
# User-settable tool options (-o)
OPT_TOOL=""
# Fpart options (-O)
OPT_FPART="-x|.zfs|-x|.snapshot*|-x|.ckpt"
# Sudo mode (-S)
OPT_SUDO=""
# Verbose mode (-v)
OPT_VERBOSE="0"
# Source directory
OPT_SRCDIR=""
# Destination directory
OPT_DSTURL=""

########## Internal variables (cannot be set through CLI yet)

# Force color usage (even if stdout and stderr are *not* associated with a terminal)
OPT_FORCECOLORS=""
# POSIX-compliant shell used within generated jobs. It must exist locally
# and remotely (when using workers), as well as support '-c' (commands) and
# '-s' (stdin) options.
# That option may be used to work-around missing 'pipefail' option from '/bin/sh'
# on certain systems (e.g. Debian)
OPT_JOBSSHELL="/bin/sh"
#OPT_JOBSSHELL="/bin/bash"

########## Various functions

#set -o errexit
#set -o nounset
LC_ALL=C

# Our color constants
# See: https://en.wikipedia.org/wiki/ANSI_escape_code
COLOR_BLUE=$(tput setaf 4)
COLOR_GREEN=$(tput setaf 2)
COLOR_ORANGE=$(tput setaf 3)
COLOR_RED=$(tput setaf 1)
COLOR_WHITE=$(tput setaf 7)
COLOR_STOP=$(tput sgr0)

# Disable 'pipefail' option if not supported
QUIRK_PIPEFAIL='set -o pipefail'
${OPT_JOBSSHELL} -c "${QUIRK_PIPEFAIL}" 2>/dev/null || \
    QUIRK_PIPEFAIL=':'

# Print help
usage () {
    cat << EOF
fpsync v${FPSYNC_VERSION} - Sync directories in parallel using fpart
Copyright (c) 2014-2024 Ganael LAPLANCHE <ganael.laplanche@martymac.org>
WWW: http://contribs.martymac.org
Usage: $0 [-p] [OPTIONS...] src_dir/ dst_url/
       $0 -l
       $0 -r runid [-R] [OPTIONS...]
       $0 -a runid
       $0 -D runid

COMMON OPTIONS:
  -t /dir/    set fpsync temp dir to </dir/> (absolute path)
  -d /dir/    set fpsync shared dir to </dir/> (absolute path)
              This option is mandatory when using SSH workers.
  -M mailaddr send an e-mail to mailaddr after a run. Multiple
              -space-separated- addresses can be specified.
  -v          verbose mode (default: quiet)
              This option can be be specified several times to
              increase verbosity level.
  -h          this help

SYNCHRONIZATION OPTIONS:
  -m tool     external copy tool to use: $(tool_print_supported)
              (default: 'rsync')
  -T path     absolute path of copy tool (default: guessed)
  -f y        transfer at most <y> files or directories per sync job
  -s z        transfer at most <z> bytes per sync job
  -E          work on a per-directory basis ('rsync' tool only)
              (WARNING!!! Enables rsync(1)'s --delete option!)
              Specify twice to enable "aggressive" mode that will isolate
              erroneous directories and enable recursive synchronization for
              them ('rsync' tool only)
  -o options  override default copy tool options with <options>
              See fpsync(1) for more details.
  -O options  override default fpart options with pipe-separated <options>
              See fpsync(1) for more details.
  -S          use sudo for filesystem crawling and synchronizations
  src_dir/    source directory (absolute path)
  dst_url/    destination directory (or URL, when using 'rsync' tool)

JOB HANDLING AND DISPATCHING OPTIONS:
  -n x        start <x> concurrent sync jobs per run
  -w wrks     space-separated list of SSH workers
              e.g.: -w 'login@host1 login@host2 login@host3'
              or:   -w 'login@host1' -w 'login@host2' -w 'login@host3'
              Jobs are executed locally if not specified (default).

RUN HANDLING OPTIONS:
  -p          prepare mode: prepare target(s) and create a resumable
              run by crawling filesystem but do not actually start
              synchronization jobs.
  -l          list previous runs and their status.
  -r runid    resume run <runid>
              (options -m, -T, -f, -s, -E, -o, -O, -S, /src_dir/ and
              /dst_url/ are ignored when resuming a previous run)
  -R          replay mode (needs option -r): re-synchronize all
              partitions from run <runid> instead of working on
              remaining ones only.
  -a runid    archive run <runid> to temp dir
  -D runid    delete run <runid>

See fpsync(1) for more details.
EOF
}

# Print a message to stdout and exit with normal exit code
end_ok () {
    [ -n "$1" ] && echo "$1"
    exit 0
}

# Print a message to stderr and exit with error code 1
end_die () {
    [ -n "$1" ] && echo "$1" 1>&2
    exit 1
}

# Print (to stdout) and log a message
# $1 = level (0 = quiet, 1 = verbose, >=2 more verbose)
# $2 = message to log
# $3 = color
echo_log () {
    local _log_ts=$(date '+%s')
    local _color_start=''
    local _color_stop=''

    # Prepare color
    if { [ -t 1 ] && [ -t 2 ] ;} || [ -n "${OPT_FORCECOLORS}" ]
    then
        case "$3" in
        "blue")
            _color_start="${COLOR_BLUE}"
            _color_stop="${COLOR_STOP}"
            ;;
        "green")
            _color_start="${COLOR_GREEN}"
            _color_stop="${COLOR_STOP}"
            ;;
        "orange")
            _color_start="${COLOR_ORANGE}"
            _color_stop="${COLOR_STOP}"
            ;;
        "red")
            _color_start="${COLOR_RED}"
            _color_stop="${COLOR_STOP}"
            ;;
        "white")
            _color_start="${COLOR_WHITE}"
            _color_stop="${COLOR_STOP}"
            ;;
        esac
    fi

    is_num "$1" && [ ${OPT_VERBOSE} -ge $1 ] && [ -n "$2" ] && \
        printf '%s\n' "${_log_ts} ${_color_start}$2${_color_stop}"
    [ -n "$2" ] && \
        echo "${_log_ts} $2" >> "${FPSYNC_LOGFILE}"
}

# Check if $1 is an absolute path
is_abs_path() {
    echo "$1" | grep -qE '^/'
}

# Check if $1 is a valid rsync URL
# Cf. rsync(1) :
#   SSH:   [USER@]HOST:DEST
#   Rsync: [USER@]HOST::DEST
#   Rsync: rsync://[USER@]HOST[:PORT]/DEST
# Simplified as: "anything but slash" followed by at least one ":"
is_remote_path() {
    echo "$1" | grep -qE '^[^/]+:'
}

# Check if $1 is a number
is_num () {
    echo "$1" | grep -qE '^[0-9]+$'
}

# Check if $1 is an acceptable size argument
# - must be greater than 0
# - may contain 'kKmMgGtTpP' suffix
is_size () {
    echo "$1" | grep -qE '^0*[1-9][0-9]*[kKmMgGtTpP]?$'
}

# Check if $1 contains (at least) a valid e-mail address
is_mailaddr () {
    echo "$1" | grep -qE '^[a-zA-Z0-9+._-]+@[a-zA-Z0-9-]+\.'
}

# Check if $1 is a valid run ID
is_runid () {
    echo "$1" | grep -qE '^[0-9]+-[0-9]+$'
}

########## Results handling

# Get files from specified run's logdir, by extension
# $1 = run ID
# $2 = file extension
run_logs_list_by_ext () {
    [ -n "$1" ] && [ -n "$2" ] && \
        find "${FPSYNC_LOGDIR_BASE}/${1}/" -type f -name "*.${2}" ! -size 0 2>/dev/null
}

# Check if file $1 exists and contains something
# $1 = file path
file_is_not_empty () {
    [ -s "$1" ]
}

# Return contents of .ret file $1
# $1 = .ret file path
retfile_get_code () {
    cat "$1" 2>/dev/null
}

# Check if .ret file $1 exists and contains a non-error return code
# $1 = .ret file path
retfile_is_success () {
    grep -q '^0$' "$1" 2>/dev/null
}

# Check if .ret file $1 exists and contains an error return code
# $1 = .ret file path
retfile_is_error () {
    grep -q -v '^0$' "$1" 2>/dev/null
}

# Return corresponding .stderr files from .ret file list
# Only non-empty log files are returned
logfiles_filter_ret_to_stderr () {
    while read _line
    do
        case "${_line}" in
        *.ret)
            file_is_not_empty "${_line%.ret}.stderr" && \
                echo "${_line%.ret}.stderr"
            ;;
        esac
    done
}

# Return .ret files containing an error return code from .ret file list
logfiles_filter_ret_with_errors () {
    while read _line
    do
        case "${_line}" in
        *.ret)
            retfile_is_error "${_line}" && \
                echo "${_line}"
            ;;
        esac
    done
}

# Return .stderr files with no error associated
# Those logs may contain interesting additional warnings
logfiles_filter_stderr_without_ret () {
    while read _line
    do
        case "${_line}" in
        *.stderr)
            retfile_is_error "${_line%.stderr}.ret" || \
                echo "${_line}"
            ;;
        esac
    done
}

########## Tool handling

# Chek if a tool is supported
# $1 = tool name
tool_is_supported () {
    echo "$1" | grep -qE '^(rsync|cpio|pax|tar|tarify)$'
}

# Chek if TAR_BIN is GNU tar
# TAR_BIN must be initialized
tar_is_GNU_tar () {
    [ -n "${TAR_BIN}" ] &&
        "${TAR_BIN}" --version 2>/dev/null | head -n 1 | grep -q 'GNU tar'
}

# Print supported tools in a friendly manner
tool_print_supported () {
    echo "'rsync', 'cpio', 'pax', 'tar' or 'tarify'"
}

# Check if a tool supports a URL as sync target
# $1 = tool name
tool_supports_urls () {
    echo "$1" | grep -q '^rsync$'
}

# Check if a tool supports directory-only mode
# (requires the ability to sync a single-level directory tree)
# $1 = tool name
tool_supports_dirsonly () {
    echo "$1" | grep -q '^rsync$'
}

# Check if a tool supports aggressive mode
# (requires the ability to sync recursively)
# $1 = tool name
tool_supports_aggressive () {
    echo "$1" | grep -q '^rsync$'
}

# Get default tool-related options
# $1 = tool name
tool_get_base_opts () {
    [ "$1" = "rsync" ] &&
        printf '%s\n' '-q -lptgoD --numeric-ids'
}

# Get mode-specific complementary tool-related options
# (main call or left part of a pipeline)
# $1 = tool name
# $2 = dirs only mode (if string not empty)
tool_get_tool_mode_opts () {
    if [ -z "$2" ]
    then
        # File-based mode: recursion is usually disabled here
        # as we are working with leaf elements only
        case "$1" in
        "rsync")
            # Non-recursive (more exactly: single-depth) rsync(1)
            printf '%s\n' '-d'
            ;;
        "cpio")
            # Passthrough mode, create directories, preserve modification time
            printf '%s\n' '-pdm'
            ;;
        "pax")
            # Read/Write mode, do not copy directories recursively, preserve everything
            printf '%s\n' '-r -w -d -p e'
            ;;
        "tar"|"tarify")
            printf '%s\n' '--no-recursion'
            ;;
        *)
            ;;
        esac
    else
        # Dirs-only mode
        case "$1" in
        "rsync")
            # Single-depth rsync(1) + deletion
            # Postpone deletion to limit impacts of a user interruption
            # XXX Aggressive mode can set option -r, which takes precedence
            # over -d (in fact, aggressive mode *depends* on having option -r
            # overriding -d)
            printf '%s\n' '-d --relative --delete --delete-after'
            ;;
        *)
            ;;
        esac
    fi
}

# Get mode-specific complementary tool-related options
# (right part of a pipeline), if any)
# $1 = tool name
# $2 = dirs only mode (if string not empty)
tool_get_tool_mode_opts_2 () {
    if [ -z "$2" ]
    then
        # File-based mode: recursion is usually disabled here
        # as we are working with leaf elements only
        case "$1" in
        "tar")
            # GNU Tar needs --delay-directory-restore during extraction.
            if tar_is_GNU_tar
            then
                printf '%s\n' '--delay-directory-restore'
            fi
            ;;
        *)
            ;;
        esac
    fi
}

# Get recursive option for specified tool
# $1 = tool name
tool_get_tool_mode_opts_recursive () {
    [ "$1" = "rsync" ] &&
        printf '%s\n' '-r'
}

# Get file list separator option for fpart
# $1 = tool name
tool_get_fpart_separator_opt () {
    case "$1" in
    "pax")
        printf '%s\n' ''
        ;;
    *)
        # Every other tool supports a null-separated list of files
        printf '%s\n' '-0'
        ;;
    esac
}

# Get mode-specific fpart options
# $1 = tool name
# $2 = dirs only mode (if string not empty)
# $3 = aggressive mode (if string not empty)
tool_get_fpart_mode_opts () {
    if [ -z "$2" ]
    then
        # File-based mode
        case "$1" in
        "rsync")
            printf '%s\n' '-zz'
            ;;
        "cpio"|"pax"|"tar"|"tarify")
            # We want empty directory entries with those tools
            # and fix parent dirs' timestamps with option -P
            # to re-apply correct metadata
            printf '%s\n' '-zzzP'
            ;;
        *)
            ;;
        esac
    else
        # Dirs-only mode
        case "$1" in
        "rsync")
            if [ -z "$3" ]
            then
                # Regular mode
                #
                # We do *not* want fpart option -zz in regular dirs-only mode
                # because unreadable directories will be created when sync'ing
                # the parent
                printf '%s\n' '-E'
            else
                # Aggressive mode
                #
                # Erroneous dirs are -from fpart's point of view-, mostly leaf
                # dirs (no subdirs should have been packed before, except
                # maybe in case of partially-read directories containing
                # subdirs). Pack erroneous dirs separately and enable recursive
                # rsync for them to try to overcome transcient errors such as
                # Linux SMB client deferring opendir() to support compound SMB
                # requests. See: https://github.com/martymac/fpart/pull/37
                printf '%s\n' '-E|-zz|-Z'
            fi
            ;;
        *)
            ;;
        esac
    fi
}

# Init tool-specific fpart hooks (black magic is here !)
# $1 = tool name
# $2 = aggressive mode (if string not empty)
# XXX That function modifies a global variable to avoid too many escape
# characters when returning values through stdout
tool_init_fpart_job_command () {
    case "$1" in
    "rsync")
        if [ -z "$2" ]
        then
            # Regular mode
            FPART_JOBCOMMAND="${OPT_JOBSSHELL} -c '${SUDO} ${TOOL_BIN} ${OPT_TOOL} \
                ${TOOL_MODEOPTS} --files-from=\\\"\${FPART_PARTFILENAME}\\\" --from0 \
                \\\"${OPT_SRCDIR}/\\\" \
                \\\"${OPT_DSTURL}/\\\"' \
                1>\"${FPSYNC_LOGDIR}/\${FPART_PARTNUMBER}.stdout\" \
                2>\"${FPSYNC_LOGDIR}/\${FPART_PARTNUMBER}.stderr\""
        else
            # Aggressive mode: enable recursivity for erroneous partitions
            # (i.e. where: (errno != 0) && (errno != EACCESS))
            # Also, skip sync for errno==EACCESS as this is a legitimate error:
            # unaccessible dirs will be re-created by parents, anyway
            FPART_JOBCOMMAND="${OPT_JOBSSHELL} -c ' \
                FPART_PARTERRNO=\\\"\${FPART_PARTERRNO}\\\"; \
                TOOL_MODEOPTS_R=; \
                if [ \\\${FPART_PARTERRNO} -ne 0 ]; \
                then \
                    if [ \\\${FPART_PARTERRNO} -eq 13 ]; \
                    then \
                        exit 0; \
                    else \
                        TOOL_MODEOPTS_R=\\\"${TOOL_MODEOPTS_R}\\\"; \
                    fi; \
                fi; \
                ${SUDO} ${TOOL_BIN} ${OPT_TOOL} ${TOOL_MODEOPTS} \
                \\\${TOOL_MODEOPTS_R} --files-from=\\\"\${FPART_PARTFILENAME}\\\" --from0 \
                \\\"${OPT_SRCDIR}/\\\" \
                \\\"${OPT_DSTURL}/\\\"' \
                1>\"${FPSYNC_LOGDIR}/\${FPART_PARTNUMBER}.stdout\" \
                2>\"${FPSYNC_LOGDIR}/\${FPART_PARTNUMBER}.stderr\""
        fi
        ;;
    "cpio")
        # XXX Warning: -0 and --quiet are non-standard
        # (not supported on Solaris), see:
        # http://pubs.opengroup.org/onlinepubs/7908799/xcu/cpio.html
        # XXX Exec whole shell cmd as root, because we need to cwd first
        FPART_JOBCOMMAND="${SUDO} ${OPT_JOBSSHELL} -c '${QUIRK_PIPEFAIL} ; \
            cd \\\"${OPT_SRCDIR}/\\\" && \
            cat \\\"\${FPART_PARTFILENAME}\\\" | \
            ${TOOL_BIN} ${OPT_TOOL} -0 --quiet ${TOOL_MODEOPTS} \
            \\\"${OPT_DSTURL}/\\\"' \
            1>\"${FPSYNC_LOGDIR}/\${FPART_PARTNUMBER}.stdout\" \
            2>\"${FPSYNC_LOGDIR}/\${FPART_PARTNUMBER}.stderr\""
        ;;
    "pax")
        # https://pubs.opengroup.org/onlinepubs/9699919799/utilities/pax.html
        # XXX Exec whole shell cmd as root, because we need to cwd first
        FPART_JOBCOMMAND="${SUDO} ${OPT_JOBSSHELL} -c '${QUIRK_PIPEFAIL} ; \
            cd \\\"${OPT_SRCDIR}/\\\" && \
            cat \\\"\${FPART_PARTFILENAME}\\\" | \
            ${TOOL_BIN} ${OPT_TOOL} ${TOOL_MODEOPTS} \
            \\\"${OPT_DSTURL}/\\\"' \
            1>\"${FPSYNC_LOGDIR}/\${FPART_PARTNUMBER}.stdout\" \
            2>\"${FPSYNC_LOGDIR}/\${FPART_PARTNUMBER}.stderr\""
        ;;
    "tar")
        FPART_JOBCOMMAND="${OPT_JOBSSHELL} -c '${QUIRK_PIPEFAIL} ; \
            ${SUDO} ${TOOL_BIN} cf - \
            ${OPT_TOOL} -C \\\"${OPT_SRCDIR}/\\\" ${TOOL_MODEOPTS} \
            --null -T \\\"\${FPART_PARTFILENAME}\\\" | \
            ${SUDO} ${TOOL_BIN} xpf - \
            ${OPT_TOOL} -C \\\"${OPT_DSTURL}/\\\" ${TOOL_MODEOPTS_2}' \
            1>\"${FPSYNC_LOGDIR}/\${FPART_PARTNUMBER}.stdout\" \
            2>\"${FPSYNC_LOGDIR}/\${FPART_PARTNUMBER}.stderr\""
        ;;
    "tarify")
        FPART_JOBCOMMAND="${OPT_JOBSSHELL} -c '${SUDO} ${TOOL_BIN} c \
            -f \\\"${OPT_DSTURL}/\${FPART_PARTNUMBER}.tar\\\" \
            ${OPT_TOOL} -C \\\"${OPT_SRCDIR}/\\\" ${TOOL_MODEOPTS} \
            --null -T \\\"\${FPART_PARTFILENAME}\\\"' \
            1>\"${FPSYNC_LOGDIR}/\${FPART_PARTNUMBER}.stdout\" \
            2>\"${FPSYNC_LOGDIR}/\${FPART_PARTNUMBER}.stderr\""
        ;;
    *)
        ;;
    esac
}

# Check if $2 contains invalid options regarding tool $1
# $1 = tool name
# $2 = tool options
tool_uses_forbidden_option () {
    # For rsync, prevent usage of :
    # --delete
    # --recursive, -r
    # -a (implies -r)
    # and leave fpsync handle them internally.
    [ "$1" = "rsync" ] && \
        { printf '%s\n' "$2" | grep -q -- '--delete' || \
          printf '%s\n' "$2" | grep -q -- '--recursive' || \
          printf '%s\n' "$2" | grep -qE -- '(^|[[:space:]])-[^[:space:]-]*r' || \
          printf '%s\n' "$2" | grep -qE -- '(^|[[:space:]])-[^[:space:]-]*a' ;}
}

########## Options handling

# Parse user options and initialize OPT_* global variables
parse_opts () {
    local opt OPTARG OPTIND

    while getopts "m:T:n:f:s:Ew:d:t:M:plr:Ra:D:o:O:Svh" opt
    do
        case "${opt}" in
        "m")
            if tool_is_supported "${OPTARG}"
            then
                OPT_TOOL_NAME=${OPTARG}
            else
                end_die "Unsupported tool, please specify $(tool_print_supported)"
            fi
            ;;
        "T")
            if is_abs_path "${OPTARG}"
            then
                OPT_TOOL_PATH="${OPTARG}"
            else
                end_die "Please supply an absolute path for tool path"
            fi
            ;;
        "n")
            if is_num "${OPTARG}" && [ ${OPTARG} -ge 1 ]
            then
                OPT_JOBS=${OPTARG}
            else
                end_die "Option -n expects a numeric value >= 1"
            fi
            ;;
        "f")
            if is_num "${OPTARG}" && [ ${OPTARG} -ge 0 ]
            then
                OPT_FPMAXPARTFILES=${OPTARG}
            else
                end_die "Option -f expects a numeric value >= 0"
            fi
            ;;
        "s")
            if { is_num "${OPTARG}" && [ ${OPTARG} -ge 0 ] ;} || is_size "${OPTARG}"
            then
                OPT_FPMAXPARTSIZE=${OPTARG}
            else
                end_die "Option -s expects a numeric value >= 0"
            fi
            ;;
        "E")
            if [ "${OPT_DIRSONLY}" = "yes" ]
            then
                OPT_AGGRESSIVE="yes"
            fi
            OPT_DIRSONLY="yes"
            ;;
        "w")
            if [ -n "${OPTARG}" ]
            then
                OPT_WRKRS="${OPT_WRKRS} ${OPTARG}"
            else
                end_die "Invalid workers list supplied"
            fi
            ;;
        "d")
            if is_abs_path "${OPTARG}"
            then
                OPT_SHDIR="${OPTARG}"
            else
                end_die "Please supply an absolute path for shared dir"
            fi
            ;;
        "t")
            if is_abs_path "${OPTARG}"
            then
                OPT_TMPDIR="${OPTARG}"
            else
                end_die "Please supply an absolute path for temp dir"
            fi
            ;;
        "M")
            if [ -n "${OPTARG}" ] && is_mailaddr "${OPTARG}"
            then
                OPT_MAIL="${OPTARG}"
            else
                end_die "Please supply a valid e-mail address"
            fi
            ;;
        "p")
            OPT_PREPARERUN="yes"
            ;;
        "l")
            OPT_LISTRUNS="yes"
            ;;
        "r")
            if [ -n "${OPTARG}" ] && is_runid "${OPTARG}"
            then
                OPT_RUNID="${OPTARG}"
            else
                end_die "Invalid run ID supplied"
            fi
            ;;
        "R")
            OPT_REPLAYRUN="yes"
            ;;
        "a")
            if [ -n "${OPTARG}" ] && is_runid "${OPTARG}"
            then
                OPT_ARCHIVERUN="${OPTARG}"
            else
                end_die "Invalid run ID supplied"
            fi
            ;;
        "D")
            if [ -n "${OPTARG}" ] && is_runid "${OPTARG}"
            then
                OPT_DELETERUN="${OPTARG}"
            else
                end_die "Invalid run ID supplied"
            fi
            ;;
        "o")
            if [ -n "${OPTARG}" ]
            then
                OPT_TOOL="${OPTARG}"
            else
                end_die "Invalid tool options supplied"
            fi
            ;;
        "O")
            if [ -n "${OPTARG}" ]
            then
                OPT_FPART="${OPTARG}"
            else
                end_die "Invalid fpart options supplied"
            fi
            ;;
        "S")
            OPT_SUDO="yes"
            ;;
        "v")
            OPT_VERBOSE="$((${OPT_VERBOSE} + 1))"
            ;;
        "h")
            usage
            end_ok
            ;;
        *)
            usage
            end_die "Invalid option specified"
            ;;
        esac
    done
    shift $((${OPTIND} - 1))

    # Validate OPT_SHDIR (shared directory)
    if [ -z "${OPT_WRKRS}" ]
    then
        # For local jobs, set shared directory to temporary directory
        [ -z "${OPT_SHDIR}" ] && \
            OPT_SHDIR="${OPT_TMPDIR}"
    else
        # For remote ones, specifying a shared directory is mandatory
        [ -z "${OPT_SHDIR}" ] && \
            end_die "Please supply a shared dir when specifying workers"
    fi

    # Run handling constraints
    _err_msg="Please specify only a single option from: -p, -l -r, -a or -D"
    [ -n "${OPT_PREPARERUN}" ] && \
        { [ -n "${OPT_LISTRUNS}" ] || \
          [ -n "${OPT_RUNID}" ] || \
          [ -n "${OPT_ARCHIVERUN}" ] || \
          [ -n "${OPT_DELETERUN}" ] ;} && \
            end_die "${_err_msg}"
    [ -n "${OPT_LISTRUNS}" ] && \
        { [ -n "${OPT_RUNID}" ] || \
          [ -n "${OPT_ARCHIVERUN}" ] || \
          [ -n "${OPT_DELETERUN}" ] ;} && \
            end_die "${_err_msg}"
    [ -n "${OPT_RUNID}" ] && \
        { [ -n "${OPT_ARCHIVERUN}" ] || \
          [ -n "${OPT_DELETERUN}" ] ;} && \
            end_die "${_err_msg}"
    [ -n "${OPT_ARCHIVERUN}" ] && \
        [ -n "${OPT_DELETERUN}" ] && \
        end_die "${_err_msg}"
    _err_msg=

    [ -n "${OPT_REPLAYRUN}" ] && [ -z "${OPT_RUNID}" ] && \
        end_die "Replay (-R) option can only be used with resume (-r) option"

    # Validate partitions' constraints
    if is_num "${OPT_FPMAXPARTFILES}" && [ ${OPT_FPMAXPARTFILES} -eq 0 ] && \
        is_num "${OPT_FPMAXPARTSIZE}" && [ ${OPT_FPMAXPARTSIZE} -eq 0 ]
    then
        end_die "Please specify a least a file (-f) or size (-s) limit for partitions"
    fi

    # Check for src_dir and dst_url presence and validity
    if [ -z "${OPT_RUNID}" ] && [ -z "${OPT_LISTRUNS}" ] && \
        [ -z "${OPT_ARCHIVERUN}" ] && [ -z "${OPT_DELETERUN}" ]
    then
        # Check src dir, must be an absolute path
        if is_abs_path "$1"
        then
            OPT_SRCDIR="$1"
        else
            usage
            end_die "Please supply an absolute path for src_dir/"
        fi
        # Check dst_url, must be either an absolute path or a URL
        if is_abs_path "$2" || is_remote_path "$2"
        then
            is_remote_path "$2" && ! tool_supports_urls "${OPT_TOOL_NAME}" && \
                end_die "URLs are not supported when using ${OPT_TOOL_NAME}"
            OPT_DSTURL="$2"
        else
            usage
            if tool_supports_urls "${OPT_TOOL_NAME}"
            then
                end_die "Please supply either an absolute path or a rsync URL for dst_url/"
            else
                end_die "Please supply an absolute path for dst_url/"
            fi
        fi
    fi

    # Handle tool-related options
    if [ "${OPT_DIRSONLY}" = "yes" ] && ! tool_supports_dirsonly "${OPT_TOOL_NAME}"
    then
        end_die "Option -E is invalid when using ${OPT_TOOL_NAME} tool"
    fi
    if [ "${OPT_AGGRESSIVE}" = "yes" ] && ! tool_supports_aggressive "${OPT_TOOL_NAME}"
    then
        end_die "Aggressive mode is invalid when using ${OPT_TOOL_NAME} tool"
    fi
    if [ -z "${OPT_TOOL}" ]
    then
        OPT_TOOL=$(tool_get_base_opts "${OPT_TOOL_NAME}")
    else
        tool_uses_forbidden_option "${OPT_TOOL_NAME}" "${OPT_TOOL}" && \
            end_die "Incompatible option(s) detected within toolopts (option -o)"
    fi
}

########## Work-related functions (in-memory, running-jobs handling)

# Initialize WORK_FREEWORKERS by expanding OPT_WRKRS up to OPT_JOBS elements,
# assigning a fixed number of slots to each worker.
# Sanitize OPT_WRKRS if necessary.
work_list_free_workers_init () {
    local _OPT_WRKRS_NUM=$(echo ${OPT_WRKRS} | awk '{print NF}')
    if [ ${_OPT_WRKRS_NUM} -gt 0 ]
    then
        local _i=0
        while [ ${_i} -lt ${OPT_JOBS} ]
        do
            local _OPT_WRKRS_IDX="$((${_i} % ${_OPT_WRKRS_NUM} + 1))"
            WORK_FREEWORKERS="${WORK_FREEWORKERS} $(echo ${OPT_WRKRS} | awk '{print $'${_OPT_WRKRS_IDX}'}')"
            _i=$((${_i} + 1))
        done
    else
        OPT_WRKRS=""
        WORK_FREEWORKERS="local"
    fi
}

# Pick-up next worker
work_list_pick_next_free_worker () {
    echo "${WORK_FREEWORKERS}" | awk '{print $1}'
}

# Remove next worker from list
work_list_trunc_next_free_worker () {
    WORK_FREEWORKERS="$(echo ${WORK_FREEWORKERS} | sed -E 's/^[[:space:]]*[^[:space:]]+[[:space:]]*//')"
}

# Push a work to the list of currently-running ones
work_list_push () {
    if [ -n "$1" ]
    then
        WORK_LIST="${WORK_LIST} $1"
        WORK_NUM="$((${WORK_NUM} + 1))"
    fi
}

# Rebuild the currently-running jobs' list by examining each process' state
work_list_refresh () {
    local _WORK_LIST=""
    local _WORK_NUM=0
    local _JOB_PID=""
    local _JOB_PART=""
    local _JOB_HOST=""
    for _JOB in ${WORK_LIST}
    do
        # Extract job info
        _JOB_PID="${_JOB%%:*}"
        _JOB_PART="${_JOB#*:}" ; _JOB_PART="${_JOB_PART%:*}"
        _JOB_HOST="${_JOB##*:}"

        if ps -p "${_JOB_PID}" 1>/dev/null 2>&1
        then
            # The process is still alive, keep it
            _WORK_LIST="${_WORK_LIST} ${_JOB}"
            _WORK_NUM="$((${_WORK_NUM} + 1))"
        else
            # Job exited (either naturally or by signal)
            if [ -n "${OPT_WRKRS}" ]
            then
                WORK_FREEWORKERS="${WORK_FREEWORKERS} ${_JOB_HOST}"
            fi

            # Not finding a .ret file is abnormal
            # Main job script (the one that generates the .ret file
            # itself) must have been killed. We mark it as erroneous by
            # simulating a SIGTERM with a return code of 143 (128 + SIGTERM).
            # See: https://pubs.opengroup.org/onlinepubs/9699919799/utilities/V3_chap02.html#tag_18_08_02
            if ! file_is_not_empty "${FPSYNC_LOGDIR}/${_JOB_PART}.ret"
            then
                echo "143" 2>/dev/null >"${FPSYNC_LOGDIR}/${_JOB_PART}.ret" || \
                    end_die "Error writing to .ret file: ${FPSYNC_LOGDIR}/${_JOB_PART}.ret"
                echo "Job killed (SIGTERM simulated by fpsync)" 2>/dev/null >>"${FPSYNC_LOGDIR}/${_JOB_PART}.stderr" || \
                    end_die "Error writing to .stderr file: ${FPSYNC_LOGDIR}/${_JOB_PART}.stderr"
            fi

            _retcode=$(retfile_get_code "${FPSYNC_LOGDIR}/${_JOB_PART}.ret")
            is_num "${_retcode}" || \
                end_die "Invalid return code fetched for job: ${_JOB_PART}"

            if [ "${_retcode}" -le 128 ] && \
                # XXX Workaround for rsync(1) returning 20 when killed
                { [ "${OPT_TOOL_NAME}" != "rsync" ] || [ "${_retcode}" -ne 20 ] ;}
            then
                # Job finished naturally, check return value
                if retfile_is_success "${FPSYNC_LOGDIR}/${_JOB_PART}.ret"
                then
                    echo_log "2" "<= [QMGR] Job ${_JOB_PART} (${_JOB}) exited (success)" "green"
                else
                    echo_log "2" "<= [QMGR] Job ${_JOB_PART} (${_JOB}) exited (error: ${_retcode})" "red"
                fi

                # Update status counters
                _run_done_jobs=$(( ${_run_done_jobs} + 1 ))
                _run_done_files=$(( \
                    ${_run_done_files} + $( \
                        job_files=0 ; \
                        . "${FPSYNC_PARTSTMPL}.${_JOB_PART}.meta" 2>/dev/null ; \
                        echo "${job_files}" \
                    ) \
                ))
                _run_done_size=$(( \
                    ${_run_done_size} + $( \
                        job_size=0 ; \
                        . "${FPSYNC_PARTSTMPL}.${_JOB_PART}.meta" 2>/dev/null ; \
                        echo "${job_size}" \
                    ) \
                ))
                # Mark job as done
                mv "${JOBS_WORKDIR}/${_JOB_PART}" "${JOBS_DONEDIR}" 2>/dev/null || \
                    end_die "Error moving job ${_JOB_PART} to done dir"
            else
                # Job killed, do *not* mark the job as done
                # to make the run resumable
                if [ ${SIGINT_COUNT} -le 1 ]
                then
                    echo_log "2" "<= [QMGR] Job ${_JOB_PART} (${_JOB}) exited (killed)" "red"
                else
                    echo_log "2" "<= [QMGR] Job ${_JOB_PART} (${_JOB}) exited (killed by user)" "red"
                fi
            fi

            # Remove useless log and result files
            file_is_not_empty "${FPSYNC_LOGDIR}/${_JOB_PART}.stdout" || \
                rm -f "${FPSYNC_LOGDIR}/${_JOB_PART}.stdout" 2>/dev/null
            file_is_not_empty "${FPSYNC_LOGDIR}/${_JOB_PART}.stderr" || \
                rm -f "${FPSYNC_LOGDIR}/${_JOB_PART}.stderr" 2>/dev/null
            retfile_is_success "${FPSYNC_LOGDIR}/${_JOB_PART}.ret" && \
                rm -f "${FPSYNC_LOGDIR}/${_JOB_PART}.ret" 2>/dev/null
        fi
    done
    WORK_LIST=${_WORK_LIST}
    WORK_NUM=${_WORK_NUM}
}

########## Jobs-related functions (on-disk, jobs' queue handling)

# Initialize job queue and work directories
job_queue_init () {
    mkdir -p "${JOBS_QUEUEDIR}" 2>/dev/null || \
        end_die "Cannot create job queue directory ${JOBS_QUEUEDIR}"
    mkdir -p "${JOBS_WORKDIR}" 2>/dev/null || \
        end_die "Cannot create job work directory ${JOBS_WORKDIR}"
    mkdir -p "${JOBS_DONEDIR}" 2>/dev/null || \
        end_die "Cannot create job done directory ${JOBS_DONEDIR}"
}

# Dump job queue information to allow later resuming
job_queue_info_dump () {
    # Create "info" file
    local _TMPMASK="$(umask)"
    umask "0077"
    touch "${JOBS_QUEUEDIR}/info" 2>/dev/null
    umask "${_TMPMASK}"

    # Dump necessary information to resume a run
    # XXX OPT_TOOL_NAME and OPT_TOOL_PATH are technically ignored when resuming
    # because fpart pass has finished and job scripts have already been written.
    # Anyway, we record them for 2 reasons:
    #   - to display them to the user
    #   - to check if OPT_TOOL_PATH exists on all workers
    cat << EOF > "${JOBS_QUEUEDIR}/info" || \
        end_die "Cannot record run information"
# Run information used for resuming, do not edit !
OPT_SRCDIR="${OPT_SRCDIR}"
OPT_DSTURL="${OPT_DSTURL}"
OPT_TOOL_NAME="${OPT_TOOL_NAME}"
OPT_TOOL_PATH="${OPT_TOOL_PATH}"
EOF
}

job_queue_info_load () {
    # Source info file and initialize a few variables
    . "${JOBS_QUEUEDIR}/info" || \
        end_die "Cannot read run information"

    # Validate loaded options
    ! tool_is_supported "${OPT_TOOL_NAME}" && \
        end_die "Invalid option value loaded from resumed run: OPT_TOOL_NAME"

    ! is_abs_path "${OPT_TOOL_PATH}" && \
        end_die "Invalid option value loaded from resumed run: OPT_TOOL_PATH"
    TOOL_BIN="${OPT_TOOL_PATH}"

    is_abs_path "${OPT_SRCDIR}" || \
        end_die "Invalid options value loaded from resumed run: OPT_SRCDIR"

    if ! tool_supports_urls "${OPT_TOOL_NAME}"
    then
        is_abs_path "${OPT_DSTURL}" || \
            end_die "Invalid options value loaded from resumed run: OPT_DSTURL"
    else
        is_abs_path "${OPT_DSTURL}" || is_remote_path "${OPT_DSTURL}" || \
            end_die "Invalid options value loaded from resumed run: OPT_DSTURL"
    fi
}

# Set the "fp_done" (fpart done) flag within job queue
job_queue_fp_done () {
    sleep 1 # Ensure this very last file gets created within the next second of
            # last job file's mtime. Necessary for filesystems that don't get
            # below the second for mtime precision (msdosfs).
    touch "${JOBS_QUEUEDIR}/fp_done"
}

# Set the "sl_stop" (sync loop stop) flag within job queue
job_queue_sl_stop () {
    touch "${JOBS_QUEUEDIR}/sl_stop"
}

########## Status functions

# Count total jobs to perform in specified run
# $1 = run ID
# FPSYNC_PARTSDIR can host several types of file :
# part.<job_number>: file listing for a sync job
# part.<job_number>.meta: job meta-information (number of files, size)
# run.meta: run meta-information (number of jobs, files, size)
run_total_jobs_count () {
    (
        total_num_parts=0 ; \
        [ -f "${FPSYNC_PARTSDIR_BASE}/${1}/run.meta" ] && \
            . "${FPSYNC_PARTSDIR_BASE}/${1}/run.meta" 2>/dev/null ; \
        echo "${total_num_parts}"
    )
}

# Count total files to transfer in specified run
# $1 = run ID
run_total_files_count () {
    (
        total_num_files=0 ; \
        [ -f "${FPSYNC_PARTSDIR_BASE}/${1}/run.meta" ] && \
            . "${FPSYNC_PARTSDIR_BASE}/${1}/run.meta" 2>/dev/null ; \
        echo "${total_num_files}"
    )
}

## Count total bytes to transfer in specified run
# $1 = run ID
run_total_size_count () {
    (
        total_size=0 ; \
        [ -f "${FPSYNC_PARTSDIR_BASE}/${1}/run.meta" ] && \
            . "${FPSYNC_PARTSDIR_BASE}/${1}/run.meta" 2>/dev/null ; \
        echo "${total_size}"
    )
}

# List incomplete jobs in specified run
# Incomplete jobs are those that remain in work dir
# $1 = run ID
# JOBS_WORKDIR can host several types of file :
# <job_number>: a sync job to perform
# 'fp_done': present when every job has been scheduled
run_incomplete_jobs_list () {
    [ -n "${1}" ] && \
      find "${JOBS_WORKDIR_BASE}/${1}" \
          -type f -a \
          '(' ! -name 'fp_done' ')' \
              -exec basename {} \; 2>/dev/null
}

# Count incomplete jobs in specified run
# $1 = run ID
run_incomplete_jobs_count () {
    run_incomplete_jobs_list "${1}" | wc -l | awk '{print $1}'
}

# List complete jobs in specified run
# $1 = run ID
run_complete_jobs_list () {
    [ -n "${1}" ] && \
        find "${JOBS_DONEDIR_BASE}/${1}" \
            -type f \
            -exec basename {} \; 2>/dev/null
}

# Count complete jobs in specified run
# $1 = run ID
run_complete_jobs_count () {
    run_complete_jobs_list "${1}" | wc -l | awk '{print $1}'
}

# Count files transferred in specified run
# $1 = run ID
run_complete_files_count () {
    local _jobs_files=0
    for _file in $(run_complete_jobs_list "${1}")
    do
        _jobs_files=$(( \
            ${_jobs_files} + $( \
                job_files=0 ; \
                . "${FPSYNC_PARTSDIR_BASE}/${1}/part.${_file}.meta" 2>/dev/null ; \
                echo "${job_files}" \
            ) \
        ))
    done
    echo "${_jobs_files}"
}

# Count bytes transferred in specified run
# $1 = run ID
run_complete_size_count () {
    local _jobs_size=0
    for _file in $(run_complete_jobs_list "${1}")
    do
        _jobs_size=$(( \
            ${_jobs_size} + $( \
                job_size=0 ; \
                . "${FPSYNC_PARTSDIR_BASE}/${1}/part.${_file}.meta" 2>/dev/null ; \
                echo "${job_size}" \
            ) \
        ))
    done
    echo "${_jobs_size}"
}

# Check if specified run is resumable
# $1 = run ID
run_is_resumable () {
    [ -f "${JOBS_QUEUEDIR_BASE}/${1}/fp_done" ] || \
        { [ -f "${JOBS_WORKDIR_BASE}/${1}/fp_done" ] && \
          [ $(run_incomplete_jobs_count "${1}") -gt 0 ] ;}
}

# Check if specified run is replayable
# $1 = run ID
run_is_replayable () {
    [ -f "${JOBS_QUEUEDIR_BASE}/${1}/fp_done" ] || \
        [ -f "${JOBS_WORKDIR_BASE}/${1}/fp_done" ]
}

# Print a specific run's status to stdout
# $1 = run ID
run_status_print () {
    if run_is_resumable "${1}"
    then
        echo "resumable (synchronization not complete, use -r to resume)"
    else
        if run_is_replayable "${1}"
        then
            echo "replayable (use -R to replay)"
        else
            echo "not resumable (fpart pass not complete)"
        fi
    fi
}

sigint_handler () {
    SIGINT_COUNT="$((${SIGINT_COUNT} + 1))"

    # Handle first ^C:
    # stop queue processing by setting the "sl_stop" flag
    if [ ${SIGINT_COUNT} -le 1 ]
    then
        job_queue_sl_stop
        echo_log "1" "===> Interrupted. Waiting for running jobs to complete..."
        echo_log "1" "===> (hit ^C again to kill them and exit)"
    # Handle subsequent ^C:
    # kill sync processes to fast-unlock the main process
    else
        [ ${SIGINT_COUNT} -le 2 ] && \
            echo_log "1" "===> Interrupted again, killing remaining jobs"
        for _KJOB in ${WORK_LIST}
        do
            # Extract job info
            _KJOB_PID="${_KJOB%%:*}"
            [ ${SIGINT_COUNT} -le 2 ] && \
                echo_log "2" "=> [QMGR] Killing job ${_KJOB}" "orange"
            kill -s INT -- "-${_KJOB_PID}" 1>/dev/null 2>&1
        done
    fi
}

# Handle ^T: print info about queue status
# (hint: on FreeBSD, SIGINFO verbosity can be reduced
# by setting sysctl 'kern.tty_info_kstacks' to 0)
siginfo_handler () {
    #### Jobs
    local _jobs_total=$(run_total_jobs_count "${FPSYNC_RUNID}")
    local _jobs_running="${WORK_NUM}"
    local _jobs_done="${_run_done_jobs}"
    # Jobs remaining, including running ones
    local _jobs_remaining=$(( ${_jobs_total} - ${_jobs_done} ))
    local _jobs_percent="??"
    [ ${_jobs_total} -ge 1 ] && \
        _jobs_percent=$(( (${_jobs_done} * 100) / ${_jobs_total} ))

    #### Files
    local _files_total=$(run_total_files_count "${FPSYNC_RUNID}")
    local _files_done="${_run_done_files}"
    local _files_remaining=$(( ${_files_total} - ${_files_done} ))
    local _files_percent="??"
    [ ${_files_total} -ge 1 ] && \
        _files_percent=$(( (${_files_done} * 100) / ${_files_total} ))

    #### Size
    local _size_total=$(run_total_size_count "${FPSYNC_RUNID}")
    local _size_done="${_run_done_size}"
    local _size_remaining=$(( ${_size_total} - ${_size_done} ))
    local _size_percent="??"
    [ ${_size_total} -ge 1 ] && \
        _size_percent=$(( (${_size_done} * 100) / ${_size_total} ))

    # Time counters, relative to the current run (resume-aware)
    local _siginfo_ts=$(date '+%s')
    local _run_elapsed_time=$(( ${_siginfo_ts} - ${_run_start_time} ))
    local _run_jobs_done=$(( ${_jobs_done} - ${_run_start_jobs} ))
    local _run_size_done=$(( ${_size_done} - ${_run_start_size} ))

    local _run_time_per_job="??"
    local _run_time_remaining="??"
    if [ ${_run_jobs_done} -ge 1 ]
    then
        _run_time_per_job=$(( ${_run_elapsed_time} / ${_run_jobs_done} ))
        _run_time_remaining=$(( ${_run_time_per_job} * ${_jobs_remaining} ))
    fi

    local _run_bytes_per_second="??"
    if [ ${_run_elapsed_time} -ge 1 ]
    then
        _run_bytes_per_second=$(( ${_run_size_done} / ${_run_elapsed_time} ))
    fi

    local _color_start=''
    local _color_stop=''
    if { [ -t 1 ] && [ -t 2 ] ;} || [ -n "${OPT_FORCECOLORS}" ]
    then
        _color_start="${COLOR_WHITE}"
        _color_stop="${COLOR_STOP}"
    fi
    printf '%s\n' "${_siginfo_ts} <=== ${_color_start}Parts done: ${_jobs_done}/${_jobs_total} (${_jobs_percent}%), remaining: ${_jobs_remaining} (${_jobs_running} running)${_color_stop}"
    printf '%s\n' "${_siginfo_ts} <=== ${_color_start}Files done: ${_files_done}/${_files_total} (${_files_percent}%), remaining: ${_files_remaining}${_color_stop}"
    printf '%s\n' "${_siginfo_ts} <=== ${_color_start}Bytes done: ${_size_done}/${_size_total} (${_size_percent}%), remaining: ${_size_remaining}${_color_stop}"
    printf '%s\n' "${_siginfo_ts} <=== ${_color_start}Time elapsed: ${_run_elapsed_time}s, remaining: ~${_run_time_remaining}s (~${_run_time_per_job}s/job, ~${_run_bytes_per_second}B/sec)${_color_stop}"
}

# Get next job name relative to ${JOBS_WORKDIR}/
# Returns empty string if no job is available
# JOBS_QUEUEDIR can host several types of file :
# <job_number>: a sync job to perform
# 'info': info file regarding this fpsync run
# 'sl_stop': the 'immediate stop' flag, set when ^C is hit
# 'fp_done': set when fpart has finished crawling src_dir/ and generated jobs
job_queue_next () {
    local _next=""
    if [ -f "${JOBS_QUEUEDIR}/sl_stop" ]
    then
        echo "sl_stop"
    else
        _next=$(cd "${JOBS_QUEUEDIR}" && ls -rt1 2>/dev/null | grep -v 'info' | head -n 1)
        if [ -n "${_next}" ]
        then
            mv "${JOBS_QUEUEDIR}/${_next}" "${JOBS_WORKDIR}" || \
                end_die "Cannot dequeue next job"
            echo "${_next}"
        fi
    fi
}

########## Program start (main() !)

# Parse command-line options
parse_opts "$@"

# Base paths, common to all runs
JOBS_QUEUEDIR_BASE="${OPT_TMPDIR}/queue"
JOBS_WORKDIR_BASE="${OPT_TMPDIR}/work"
JOBS_DONEDIR_BASE="${OPT_TMPDIR}/done"
FPSYNC_PARTSDIR_BASE="${OPT_SHDIR}/parts"
FPSYNC_LOGDIR_BASE="${OPT_SHDIR}/log"

## Paths to 3rd party binaries

# Tar version to use for 'tar' and 'tarify' tools
# Use 'gtar' on Solaris-based OS
TAR_NAME="tar"

# Paths to executables that must exist locally
FPART_BIN="$(command -v fpart)"
SSH_BIN="$(command -v ssh)"
MAIL_BIN="$(command -v mail)"
TAR_BIN="$(command -v ${TAR_NAME})"

# Paths to executables that must exist both locally and remotely
SUDO_BIN="$(command -v sudo)"
SUDO=""
[ -n "${OPT_SUDO}" ] && \
    SUDO="${SUDO_BIN}"

# Paths to executables that must exist either locally or remotely (depending
# on if you use SSH or not). When using SSH, the following binaries must be
# present at those paths on each worker.
#
# XXX OPT_TOOL_NAME, OPT_TOOL_PATH and TOOL_BIN may be overridden
# later when resuming a run, see job_queue_info_dump() and job_queue_info_load()
if [ -n "${OPT_TOOL_PATH}" ]
then
    # Use the path to the tool if provided on the command line
    TOOL_BIN="${OPT_TOOL_PATH}"
else
    # Provide defaults regarding tool name
    case "${OPT_TOOL_NAME}" in
    "tar"|"tarify")
        TOOL_BIN="${TAR_BIN}"
        ;;
    *)
        TOOL_BIN=$(command -v "${OPT_TOOL_NAME}")
        ;;
    esac
    # Set this for job_queue_info_dump and job_queue_info_load to work correctly
    OPT_TOOL_PATH="${TOOL_BIN}"
fi

## Simple run-related -independent- actions, not requiring FPSYNC_RUNID

# List runs and exit
if [ -n "${OPT_LISTRUNS}" ]
then
    echo "<=== Listing runs"
    for _run in \
        $(cd "${FPSYNC_PARTSDIR_BASE}" 2>/dev/null && ls -1)
    do
        echo "===> Run ID: ${_run}, status: $(run_status_print "${_run}")"
        if [ ${OPT_VERBOSE} -ge 1 ]
        then
            if run_is_resumable "${_run}"
            then
                _jobs_total=$(run_total_jobs_count "${_run}")
                _jobs_done=$(run_complete_jobs_count "${_run}")
                _jobs_remaining=$(( ${_jobs_total} - ${_jobs_done} ))
                _jobs_percent="??"
                [ ${_jobs_total} -ge 1 ] && \
                    _jobs_percent=$(( (${_jobs_done} * 100) / ${_jobs_total} ))
                echo "     Jobs done: ${_jobs_done}/${_jobs_total} (${_jobs_percent}%), remaining: ${_jobs_remaining}"
            else
                if run_is_replayable "${_run}"
                then
                    _report_logs_fatal=$(run_logs_list_by_ext "${_run}" "ret" | logfiles_filter_ret_with_errors)
                    if [ -z "${_report_logs_fatal}" ]
                    then
                        echo "     Run completed without error."
                    else
                        echo "     Run completed with errors."
                    fi
                fi
            fi
        fi
    done
    end_ok
fi

# Delete run and exit
if [ -n "${OPT_DELETERUN}" ]
then
    { [ -d "${FPSYNC_PARTSDIR_BASE}/${OPT_DELETERUN}" ] && \
        rm -rf "${FPSYNC_PARTSDIR_BASE}/${OPT_DELETERUN}" && \
        rm -rf "${FPSYNC_LOGDIR_BASE}/${OPT_DELETERUN}" && \
        rm -rf "${JOBS_QUEUEDIR_BASE}/${OPT_DELETERUN}" && \
        rm -rf "${JOBS_WORKDIR_BASE}/${OPT_DELETERUN}" && \
        rm -rf "${JOBS_DONEDIR_BASE}/${OPT_DELETERUN}" && \
        end_ok "Successfully deleted run ${OPT_DELETERUN}" ;} || \
            end_die "Error deleting run ${OPT_DELETERUN}"
fi

# Archive run and exit
if [ -n "${OPT_ARCHIVERUN}" ]
then
    [ ! -x "${TAR_BIN}" ] && \
        end_die "Tar is missing locally, check your configuration"

    { [ -d "${FPSYNC_PARTSDIR_BASE}/${OPT_ARCHIVERUN}" ] && \
        ( cd "${OPT_TMPDIR}" 2>/dev/null && \
            "${TAR_BIN}" czf fpsync-run-${OPT_ARCHIVERUN}.tgz \
                "${FPSYNC_PARTSDIR_BASE}/${OPT_ARCHIVERUN}" \
                "${FPSYNC_LOGDIR_BASE}/${OPT_ARCHIVERUN}" \
                "${JOBS_QUEUEDIR_BASE}/${OPT_ARCHIVERUN}" \
                "${JOBS_WORKDIR_BASE}/${OPT_ARCHIVERUN}" \
                "${JOBS_DONEDIR_BASE}/${OPT_ARCHIVERUN}" 2>/dev/null ) && \
            end_ok "Successfully created ${OPT_TMPDIR}/fpsync-run-${OPT_ARCHIVERUN}.tgz" ;} || \
                end_die "Error archiving run ${OPT_ARCHIVERUN}"
fi

## Options' post-processing section, advanced variables' initialization

# Run ID initialization
if [ -n "${OPT_RUNID}" ]
then
    # Resume mode, check if run ID exists
    if [ -d "${JOBS_QUEUEDIR_BASE}/${OPT_RUNID}" ] && \
        [ -d "${JOBS_WORKDIR_BASE}/${OPT_RUNID}" ] && \
        [ -d "${JOBS_DONEDIR_BASE}/${OPT_RUNID}" ]
    then
        FPSYNC_RUNID="${OPT_RUNID}"
    else
        end_die "Could not find specified run's queue/work/done directories"
    fi
else
    # Generate a unique run ID. This run ID *must* remain
    # unique from one run to another.
    FPSYNC_RUNID="$(date '+%s')-$$"
fi

# Queue manager configuration for current run.
# Those queues remain local, even when using SSH.
JOBS_QUEUEDIR="${JOBS_QUEUEDIR_BASE}/${FPSYNC_RUNID}" # Sync jobs' queue dir
JOBS_WORKDIR="${JOBS_WORKDIR_BASE}/${FPSYNC_RUNID}"   # Currently syncing jobs' dir
JOBS_DONEDIR="${JOBS_DONEDIR_BASE}/${FPSYNC_RUNID}"   # Done jobs' dir

# Shared paths for current run.
# Those ones must be shared amongst all nodes when using SSH
# (e.g. through a NFS share mounted on *every* single node, including the master
# 'job submitter').
FPSYNC_PARTSDIR="${FPSYNC_PARTSDIR_BASE}/${FPSYNC_RUNID}"
FPSYNC_PARTSTMPL="${FPSYNC_PARTSDIR}/part"
FPSYNC_LOGDIR="${FPSYNC_LOGDIR_BASE}/${FPSYNC_RUNID}"
FPSYNC_LOGFILE="${FPSYNC_LOGDIR}/fpsync.log"
FPSYNC_LOGFIFO="${FPSYNC_LOGDIR}/.fpsync.log.fifo"

# Prepare mode-specific tool and fpart options
# used when starting a new run (only)
TOOL_MODEOPTS=$(tool_get_tool_mode_opts "${OPT_TOOL_NAME}" "${OPT_DIRSONLY}")
TOOL_MODEOPTS_2=$(tool_get_tool_mode_opts_2 "${OPT_TOOL_NAME}" "${OPT_DIRSONLY}")
TOOL_MODEOPTS_R=$(tool_get_tool_mode_opts_recursive "${OPT_TOOL_NAME}")
FPART_SEPARATOROPT=$(tool_get_fpart_separator_opt "${OPT_TOOL_NAME}")
FPART_MODEOPTS=$(tool_get_fpart_mode_opts "${OPT_TOOL_NAME}" "${OPT_DIRSONLY}" "${OPT_AGGRESSIVE}")
FPART_JOBCOMMAND= ; tool_init_fpart_job_command "${OPT_TOOL_NAME}" "${OPT_AGGRESSIVE}"
FPART_POSTHOOK="echo \"${FPART_JOBCOMMAND}\" > \
    \"${JOBS_QUEUEDIR}/\${FPART_PARTNUMBER}\" && \
    printf '%s\n%s\n' job_size=\"\${FPART_PARTSIZE}\" job_files=\"\${FPART_PARTNUMFILES}\" > \
    \"${FPSYNC_PARTSTMPL}.\${FPART_PARTNUMBER}.meta\" && \
    printf '%s\n%s\n%s\n' total_num_parts=\"\${FPART_TOTALNUMPARTS}\" total_size=\"\${FPART_TOTALSIZE}\" total_num_files=\"\${FPART_TOTALNUMFILES}\" > \
    \"${FPSYNC_PARTSDIR}/run.meta\" && \
    [ ${OPT_VERBOSE} -ge 2 ] && \
    echo \"\$(date '+%s') ==> [FPART] Partition \${FPART_PARTNUMBER} written\"" # [1]

# [1] Be careful to host the job queue on a filesystem that can handle
# fine-grained mtime timestamps (i.e. with a sub-second precision) if you want
# the queue to be processed in order when fpart generates several job files per
# second.
# On FreeBSD, vfs timestamps' precision can be tuned using the
# vfs.timestamp_precision sysctl. See vfs_timestamp(9).

## End of options' post-processing section, let's start for real now !

SIGINT_COUNT=0      # ^C counter

WORK_NUM=0          # Current number of running processes
WORK_LIST=""        # Work PID:PART:WORKER list
WORK_FREEWORKERS="" # Free workers' list

# Check for local binaries' presence
[ ! -x "${FPART_BIN}" ] && \
    end_die "Fpart is missing locally, check your configuration"
[ -n "${OPT_WRKRS}" ] && [ ! -x "${SSH_BIN}" ] && \
    end_die "SSH is missing locally, check your configuration"
[ -n "${OPT_MAIL}" ] && [ ! -x "${MAIL_BIN}" ] && \
    end_die "Mail is missing locally, check your configuration"
[ -n "${OPT_SUDO}" ] && [ ! -x "${SUDO}" ] && \
    end_die "Sudo is missing locally, check your configuration"
# Check for common commands/tools (we accept builtins here)
for _tool in 'ps' 'grep' 'printf' 'head' 'cut' 'awk' 'sed' 'wc'
do
    [ -z "$(command -v ${_tool})" ] && \
        end_die "'${_tool}' is missing locally, check your configuration"
done

# Create / check for fpart shared directories
if [ -z "${OPT_RUNID}" ]
then
    # For a new run, create those directories
    mkdir -p "${FPSYNC_PARTSDIR}" 2>/dev/null || \
        end_die "Cannot create partitions' output directory: ${FPSYNC_PARTSDIR}"
    mkdir -p "${FPSYNC_LOGDIR}" 2>/dev/null || \
        end_die "Cannot create log directory: ${FPSYNC_LOGDIR}"
else
    # In resume mode, FPSYNC_PARTSDIR and FPSYNC_LOGDIR must already exist
    if [ ! -d "${FPSYNC_PARTSDIR}" ] || [ ! -d "${FPSYNC_LOGDIR}" ]
    then
        end_die "Could not find specified run's 'parts' and 'log' directories"
    else
        # When replaying a run, cleanup previous logs to avoid leftovers
        # if replay gets canceled. Keep fpsync.log file to retain fpart
        # pass logs. Replay/rerun logs wil be appended to that file.
        if [ -n "${OPT_REPLAYRUN}" ]
        then
            find "${FPSYNC_LOGDIR}/" -type f -a '(' \
                    -name '*.ret' -o -name '*.stdout' -o -name '*.stderr' \
                ')' -exec rm {} \; 2>/dev/null || \
                    end_die "Error replaying specified run (could not cleanup previous logs)"
        fi
    fi
fi

# Create or update log file
{ touch "${FPSYNC_LOGFILE}" 2>/dev/null && [ -w "${FPSYNC_LOGFILE}" ] ;} || \
    end_die "Cannot create log file: ${FPSYNC_LOGFILE}"

# Create / check for run's queue and work dirs
if [ -z "${OPT_RUNID}" ]
then
    # For a new run, create those directories
    job_queue_init
else
    # When resuming a run, check if :
    # - the queue, work and done dirs exist
    # - we can get the number of workers previously implied
    #   (the 'info' flag is present)
    # + For resume mode, strictly check if :
    #   - fpart pass has completed
    #     (the 'fp_done' flag has been created in the job queue dir)
    #   - fpsync run has *not* completed
    #     (the 'fp_done' flag is *still* present in the job queue dir, or
    #      the 'fp_done' flag is present in the work queue dir and jobs remain
    #      incomplete because of a crash or kill)
    # + For replay mode, we only check that :
    #   - fpart pass has completed
    #     (the 'fp_done' flag is present either in the job queue dir or
    #      in the work queue dir)
    [ ! -f "${JOBS_QUEUEDIR}/info" ] && \
        end_die "Specified run is not resumable ('info' flag missing)"
    [ ! -d "${JOBS_WORKDIR}" ] && \
        end_die "Specified run is not resumable (work dir missing)"
    [ ! -d "${JOBS_DONEDIR}" ] && \
        end_die "Specified run is not resumable (done dir missing)"

    # Simple resume
    if [ -z "${OPT_REPLAYRUN}" ]
    then
        if ! run_is_resumable "${OPT_RUNID}"
        then
            end_die "Specified run is not resumable (try -R to replay ?)"
        fi
    # Full replay (option -R used)
    else
        if ! run_is_replayable "${OPT_RUNID}"
        then
            end_die "Specified run is not replayable (fpart pass not complete)"
        fi
    fi

    # Run is resumable, try to reload info and prepare queues
    job_queue_info_load

    # Remove the "sl_stop" flag, if any
    rm -f "${JOBS_QUEUEDIR}/sl_stop" 2>/dev/null

    # Move the 'fp_done' flag (if any) to the jobs queue
    mv "${JOBS_WORKDIR}/fp_done" "${JOBS_QUEUEDIR}" 2>/dev/null

    # Move potentially-incomplete jobs to the jobs queue
    for _file in $(run_incomplete_jobs_list "${FPSYNC_RUNID}")
    do
        echo_log "3" "=> [QMGR] Rescheduling job ${_file} (incomplete)"
        mv "${JOBS_WORKDIR}/${_file}" "${JOBS_QUEUEDIR}" 2>/dev/null || \
            end_die "Error resuming specified run (could not re-schedule jobs)"
    done

    # When replaying, also reschedule already done jobs to the jobs queue
    if [ -n "${OPT_REPLAYRUN}" ]
    then
        for _file in \
            $(cd "${JOBS_DONEDIR}" && ls -1)
        do
            echo_log "3" "=> [QMGR] Rescheduling job ${_file}"
            mv "${JOBS_DONEDIR}/${_file}" "${JOBS_QUEUEDIR}" 2>/dev/null || \
                end_die "Error replaying specified run (could not re-schedule jobs)"
        done
    fi
fi

# Validate src_dir/ locally (needed for fpart) for first runs or local ones
if [ -z "${OPT_RUNID}" ] || [ -z "${OPT_WRKRS}" ]
then
    [ ! -d "${OPT_SRCDIR}" ] && \
        end_die "Source directory does not exist (or is not a directory): ${OPT_SRCDIR}"
fi

# When using SSH, validate src_dir/ and dst_url/ remotely and check for tool
# presence (this also allows checking SSH connectivity to each declared host)
if [ -n "${OPT_WRKRS}" ]
then
    echo_log "2" "=====> Validating requirements on SSH nodes..."

    _FIRST_HOST="$(echo ${OPT_WRKRS} | awk '{print $1}')"
    for _host in ${OPT_WRKRS}
    do
        # Check for sudo presence (it must be passwordless)
        if [ -n "${OPT_SUDO}" ]
        then
            "${SSH_BIN}" "${_host}" "${SUDO} ${OPT_JOBSSHELL} -c ':' 2>/dev/null" || \
                end_die "Sudo executable not found or requires password on target ${_host}"
        fi

        # When using a local (or NFS-mounted) dest dir...
        if is_abs_path "${OPT_DSTURL}"
        then
            # ...blindly try to create dst_url/ as well as a witness file on
            # the first node. Using a witness file will allow us to check for
            # its presence/visibility from other nodes, avoiding "split-brain"
            # situations where dst_url/ exists but is not shared amongst all
            # nodes (typically a local mount point where the shared storage
            # area *should* be mounted but isn't, for any reason).
            [ "${_host}" = "${_FIRST_HOST}" ] && \
                "${SSH_BIN}" "${_host}" "${OPT_JOBSSHELL} -c 'mkdir -p \"${OPT_DSTURL}\" && \
                    ${SUDO} touch \"${OPT_DSTURL}/${FPSYNC_RUNID}\"' 2>/dev/null"
        fi

        # Check for src_dir/ presence
        "${SSH_BIN}" "${_host}" "${OPT_JOBSSHELL} -c '[ -d \"${OPT_SRCDIR}\" ]'" || \
            end_die "Source directory does not exist on target ${_host} (or is not a directory): ${OPT_SRCDIR}"

        # Check for dst_url/ presence (witness file)
        if is_abs_path "${OPT_DSTURL}"
        then
            "${SSH_BIN}" "${_host}" "${OPT_JOBSSHELL} -c '[ -f \"${OPT_DSTURL}/${FPSYNC_RUNID}\" ]'" || \
                end_die "Destination directory (shared) is not available on target ${_host}: ${OPT_DSTURL}"
        fi

        # Finally, check for tool presence
        "${SSH_BIN}" "${_host}" "${OPT_JOBSSHELL} -c '[ -x \"${TOOL_BIN}\" ]'" || \
            end_die "Tool ${OPT_TOOL_NAME} not useable on target ${_host}: ${TOOL_BIN} not found"

        echo_log "2" "<=== ${_host}: OK"
    done

    # Remove witness file
    if is_abs_path "${OPT_DSTURL}"
    then
        "${SSH_BIN}" "${_FIRST_HOST}" \
            "${OPT_JOBSSHELL} -c '${SUDO} rm -f \"${OPT_DSTURL}/${FPSYNC_RUNID}\"' 2>/dev/null"
    fi

    unset _FIRST_HOST
else
    # Local usage - create dst_url/ and check for tool presence
    if is_abs_path "${OPT_DSTURL}" && [ ! -d "${OPT_DSTURL}" ]
    then
        mkdir -p "${OPT_DSTURL}" 2>/dev/null || \
            end_die "Cannot create destination directory: ${OPT_DSTURL}"
    fi
    [ ! -x "${TOOL_BIN}" ] && \
        end_die "Tool ${OPT_TOOL_NAME} not useable locally: ${TOOL_BIN} not found"
fi

# Dispatch OPT_WRKRS into WORK_FREEWORKERS
work_list_free_workers_init

# Let's rock !
echo_log "4" "----"
echo_log "1" "Info: Fpsync started (syncing ${OPT_SRCDIR} => ${OPT_DSTURL}, pid: $$)" "blue"
echo_log "1" "Info: Run ID: ${FPSYNC_RUNID}$({ [ -n "${OPT_REPLAYRUN}" ] && echo ' (replayed)' ;} || { [ -n "${OPT_RUNID}" ] && echo ' (resumed)' ;})" "blue"
echo_log "2" "Info: Start time: $(date)" "blue"
echo_log "2" "Info: Concurrent sync jobs: ${OPT_JOBS}"
echo_log "2" "Info: Workers: $(echo "${OPT_WRKRS}" | sed -E -e 's/^[[:space:]]+//' -e 's/[[:space:]]+/ /g')$([ -z "${OPT_WRKRS}" ] && echo 'local')"
echo_log "2" "Info: Shared dir: ${OPT_SHDIR}"
echo_log "2" "Info: Temp dir: ${OPT_TMPDIR}"
# The following two options are just a recall to the user
# (they are technically ignored when resuming a run)
echo_log "2" "Info: Tool name: \"${OPT_TOOL_NAME}\""
echo_log "2" "Info: Tool path: \"${TOOL_BIN}\""
if [ -z "${OPT_RUNID}" ]
then
    # The following options are useless when resuming a run
    echo_log "2" "Info: Tool options: \"${OPT_TOOL}\""
    echo_log "2" "Info: Fpart options: \"${OPT_FPART}\""
    echo_log "2" "Info: Max files or directories per sync job: ${OPT_FPMAXPARTFILES}"
    echo_log "2" "Info: Max bytes per sync job: ${OPT_FPMAXPARTSIZE}"
fi
[ "${QUIRK_PIPEFAIL}" = ":" ] && \
    echo_log "1" "Warn: shell '${OPT_JOBSSHELL}' does not support the 'pipefail' option, sync errors may be missed" "orange"

# Record run information
job_queue_info_dump

# Initial status, required by siginfo_handler()
_run_start_time=$(date '+%s')
_run_start_jobs=$(run_complete_jobs_count "${FPSYNC_RUNID}")
_run_start_files=$(run_complete_files_count "${FPSYNC_RUNID}")
_run_start_size=$(run_complete_size_count "${FPSYNC_RUNID}")
# Current status (incremented after each job is finished)
_run_done_jobs="${_run_start_jobs}"
_run_done_files="${_run_start_files}"
_run_done_size="${_run_start_size}"

# When not resuming a previous run, start fpart
if [ -z "${OPT_RUNID}" ]
then
    # Set limited traps during FS crawling
    # They will possibly be overridden if starting the queue manager
    trap ':' 2
    trap '' 29

    # Prepare logger for sub-shell below
    rm -f "${FPSYNC_LOGFIFO}" 2>/dev/null
    mkfifo "${FPSYNC_LOGFIFO}" || \
        end_die "Cannot create named pipe: ${FPSYNC_LOGFIFO}"

    echo_log "2" "===> Starting Fpart" "blue"
    # Fpart process is forked in the same process group as its parent
    # (the main script) to allow SIGINT propagation and tty output
    # https://pubs.opengroup.org/onlinepubs/009604599/basedefs/xbd_chap11.html#tag_11_01_04
    (
        # Trap SIGINT to a noop to let sub-shell process
        # continue after fpart termination
        trap ':' 2

        # We use a FIFO for logging to avoid a pipe because we would need
        # a 'pipefail' option from the shell, which is not always available
        # (Debian's dash does not support it, for example)
        tee -a "${FPSYNC_LOGFILE}" < "${FPSYNC_LOGFIFO}" &

        # Allow passing special characters within fpart sub-options
        # (OPT_FPART, FPART_MODEOPTS)
        set -o noglob
        # Allow spaces within fpart sub-options
        IFS="|"

        echo_log "1" "===> Analyzing filesystem..." "blue"
        # Start fpart from src_dir/ directory and produce jobs within
        # ${JOBS_QUEUEDIR}/
        cd "${OPT_SRCDIR}" && \
            ${SUDO} "${FPART_BIN}" \
            $([ ${OPT_FPMAXPARTFILES} -gt 0 ] && printf '%s\n' "-f ${OPT_FPMAXPARTFILES}") \
            $({ { is_num "${OPT_FPMAXPARTSIZE}" && [ ${OPT_FPMAXPARTSIZE} -gt 0 ] ;} || is_size "${OPT_FPMAXPARTSIZE}" ;} && printf '%s\n' "-s ${OPT_FPMAXPARTSIZE}") \
            -o "${FPSYNC_PARTSTMPL}" ${FPART_SEPARATOROPT} -e ${OPT_FPART} ${FPART_MODEOPTS} -L \
            -W "${FPART_POSTHOOK}" . 1>"${FPSYNC_LOGFIFO}" 2>&1

        if [ $? -ne 0 ]
        then
            job_queue_sl_stop
            echo_log "1" "<=== Fpart exited with errors" "red"
        else
            # Tell job_queue_loop that crawling has finished
            job_queue_fp_done
            echo_log "1" "<=== Fpart crawling finished" "blue"
            echo_log "1" "<=== Generated $(run_total_jobs_count "${FPSYNC_RUNID}") parts ($(run_total_size_count "${FPSYNC_RUNID}")B, $(run_total_files_count "${FPSYNC_RUNID}") files)"
        fi

        # Remove FIFO (once tee exits and closes its fd)
        rm -f "${FPSYNC_LOGFIFO}"
    ) &
    echo_log "2" "<=== Fpart started (from sub-shell pid=$!)"
fi

# When not in prepare mode, start synchronization loop
if [ -z "${OPT_PREPARERUN}" ]
then
    # Set SIGINT and SIGINFO traps and start job_queue_loop
    trap 'sigint_handler' 2
    trap 'siginfo_handler' 29
    echo_log "2" "===> Use ^C to abort, ^T (SIGINFO) to display status"

    # Main jobs' loop: pick up jobs within the queue directory and start them
    echo_log "2" "===> [QMGR] Starting queue manager" "blue"

    _next_job=""
    while [ "${_next_job}" != "fp_done" ] && [ "${_next_job}" != "sl_stop" ]
    do
        _PID=""
        if [ ${WORK_NUM} -lt ${OPT_JOBS} ]
        then
            _next_job="$(job_queue_next)"
            if [ -n "${_next_job}" ] && \
                [ "${_next_job}" != "fp_done" ] && \
                [ "${_next_job}" != "sl_stop" ]
            then
                if [ -z "${OPT_WRKRS}" ]
                then
                    echo_log "2" "=> [QMGR] Starting job ${_next_job} locally (${JOBS_WORKDIR}/${_next_job})" "blue"
                    # We want a new process group for each sync process to
                    # prevent SIGINT propagation
                    set -m
                    (
                        . "${JOBS_WORKDIR}/${_next_job}"
                        echo "$?" > "${FPSYNC_LOGDIR}/${_next_job}.ret"
                    ) &
                    _next_pid="$!"
                    work_list_push "${_next_pid}:${_next_job}:local"
                    set +m
                    echo_log "3" "=> [QMGR] Job ${_next_job} (${_next_pid}:${_next_job}:local) started, PID: ${_next_pid}"
                else
                    _next_host="$(work_list_pick_next_free_worker)"
                    work_list_trunc_next_free_worker
                    echo_log "2" "=> [QMGR] Starting job ${_next_job} on ${_next_host} (${JOBS_WORKDIR}/${_next_job})" "blue"
                    set -m
                    (
                        "${SSH_BIN}" "${_next_host}" "${OPT_JOBSSHELL} -s" \
                            < "${JOBS_WORKDIR}/${_next_job}"
                        echo "$?" > "${FPSYNC_LOGDIR}/${_next_job}.ret"
                    ) &
                    _next_pid="$!"
                    work_list_push "${_next_pid}:${_next_job}:${_next_host}"
                    set +m
                    echo_log "3" "=> [QMGR] Job ${_next_job} (${_next_pid}:${_next_job}:${_next_host}) started, PID: ${_next_pid}"
                fi
            fi
        else
            work_list_refresh
            sleep 0.2
        fi
    done

    if [ "${_next_job}" = "fp_done" ]
    then
        echo_log "2" "<=== [QMGR] Done submitting jobs. Waiting for them to finish."
    else
        echo_log "2" "<=== [QMGR] Stopped. Waiting for jobs to finish."
    fi

    # Wait for remaining processes, if any
    # (use an active wait to display process status)
    while [ ${WORK_NUM} -gt 0 ]
    do
        work_list_refresh
        sleep 0.2
    done
    echo_log "2" "<=== [QMGR] Queue processed"
else
    # We are in prepare mode, just wait for fpart to finish FS crawling
    wait
fi

# Display final status
[ ${OPT_VERBOSE} -ge 1 ] && siginfo_handler

if [ -f "${JOBS_QUEUEDIR}/sl_stop" ]
then
    echo_log "1" "<=== Fpsync interrupted." "red"
    end_die
fi

# Examine results and prepare report
if [ -n "${OPT_PREPARERUN}" ]
then
    echo_log "0" "<=== Successfully prepared run: ${FPSYNC_RUNID}" "green"
    _report_subj="Fpsync run ${FPSYNC_RUNID} (prepared)"
else
    _report_subj="Fpsync run ${FPSYNC_RUNID}"
fi
_ts_now=$(date '+%s')
_total_run_elapsed_time="$(( ${_ts_now} - ${_run_start_time} ))"
_report_logs_fatal=$(run_logs_list_by_ext "${FPSYNC_RUNID}" "ret" | logfiles_filter_ret_with_errors)
_report_logs_fatal_stderr=$(echo "${_report_logs_fatal}" | logfiles_filter_ret_to_stderr)
_report_logs_additional_stderr=$(run_logs_list_by_ext "${FPSYNC_RUNID}" "stderr" | logfiles_filter_stderr_without_ret)
_report_body=$(
    if [ -z "${_report_logs_fatal}" ]
    then
        echo "Fpsync completed without error in ${_total_run_elapsed_time}s."
    else
        echo "Fpsync completed with errors in ${_total_run_elapsed_time}s"
        echo ""
        echo "Some jobs did NOT return 0 (success):"
        echo "${_report_logs_fatal}"
        if [ -n "${_report_logs_fatal_stderr}" ]
        then
            echo ""
            echo "You may find more information in the following logs:"
            echo "${_report_logs_fatal_stderr}"
        fi
    fi
    if [ -n "${_report_logs_additional_stderr}" ]
    then
        echo ""
        echo "Some jobs reported non-fatal problems to stderr:"
        echo "${_report_logs_additional_stderr}"
    fi
    if [ -n "${OPT_MAIL}" ]
    then
        echo ""
        echo "--"
        echo "Logs generated by fpsync(1) tool."
        echo "See https://www.fpart.org for more information."
    fi
)

# Print report
echo_log "1" "<=== ${_report_body}"
echo_log "2" "<=== End time: $(date)" "blue"

# Send mail if required
[ -n "${OPT_MAIL}" ] && \
    printf "Sync ${OPT_SRCDIR} => ${OPT_DSTURL}\n\n${_report_body}\n" | ${MAIL_BIN} -s "${_report_subj}" ${OPT_MAIL}

if [ -n "${_report_logs_fatal}" ]
then
    echo_log "1" "Info: Fpsync stopped (with errors)" "red"
    end_die
fi
echo_log "1" "Info: Fpsync stopped (with success)" "green"
end_ok