Collapse file tree

Files

 master

Search this repository

/

fpsync

Copy path

BlameMore file actions

BlameMore file actions

Latest commit

martymac

Update links (http -> https)

Sep 5, 2024

ff8228f · Sep 5, 2024

History

History

Open commit details

History

executable file

·

2050 lines (1869 loc) · 70.4 KB

/

fpsync

Top

File metadata and controls

• Code
• Blame

executable file

·

2050 lines (1869 loc) · 70.4 KB

Raw

Copy raw file

Download raw file

You must be signed in to make or propose changes

More edit options

Open symbols panel

Edit and raw actions

#!/bin/sh # Copyright (c) 2014-2025 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.7.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="" # Date format used for log prefix (-Z) OPT_DATEFORMAT="+%s" # 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 version version () { cat << EOF fpsync v${FPSYNC_VERSION} - Sync directories in parallel using fpart Copyright (c) 2014-2025 Ganael LAPLANCHE <ganael.laplanche@martymac.org> WWW: https://www.fpart.org, https://contribs.martymac.org EOF } # Print help usage () { version cat << EOF 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. -Z prefix logs with human-readable dates instead of timestamps -v verbose mode (default: quiet) This option can be be specified several times to increase verbosity level. -h this help -V print version 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 "${OPT_DATEFORMAT}") 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]?$' } # Convert $1 (in bytes) to a human-readable size bytes_to_human_size () { if ! is_num "$1" then echo "$1" return fi local _result=$(echo "$1" | bc 2>/dev/null) # Remove leading 0s local _sufx="B,KiB,MiB,GiB,TiB,PiB" local _round_max=$(( $(echo ${_sufx} | tr -cd ',' | \ wc -c | awk '{print $1}') + 1 )) local _round=1 while [ "${_round}" -lt "${_round_max}" ] && \ [ $(echo "${_result} >= 1024" | bc 2>/dev/null) -eq 1 ] do _result=$(echo "scale=2; ${_result} / 1024" | bc 2>/dev/null) _round=$(( ${_round} + 1 )) done echo "${_result}$(echo ${_sufx} | cut -d ',' -f ${_round} 2>/dev/null)" } # 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" ] } # Wait $1 rounds for non-empty file $2 to appear # $1 = sleep rounds to wait # $2 = file path wait_for_file () { local _retry="$1" local _file="$2" while [ "${_retry}" -gt 0 ] && ! file_is_not_empty "$2" do sleep 0.2 _retry="$((${_retry} - 1))" done [ "${_retry}" -gt 0 ] && \ return 0 return 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:Zplr:Ra:D:o:O:SvhV" 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 ;; "Z") OPT_DATEFORMAT="+%b %d %H:%M:%S" ;; "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 ;; "V") version 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 ! wait_for_file 5 "${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 # Display computed status local _log_ts=$(date "${OPT_DATEFORMAT}") printf '%s\n' "${_log_ts} <=== ${_color_start}Parts done: ${_jobs_done}/${_jobs_total} (${_jobs_percent}%), remaining: ${_jobs_remaining} (${_jobs_running} running)${_color_stop}" printf '%s\n' "${_log_ts} <=== ${_color_start}Files done: ${_files_done}/${_files_total} (${_files_percent}%), remaining: ${_files_remaining}${_color_stop}" printf '%s\n' "${_log_ts} <=== ${_color_start}Bytes done: $(bytes_to_human_size ${_size_done})/$(bytes_to_human_size ${_size_total}) (${_size_percent}%), remaining: $(bytes_to_human_size ${_size_remaining})${_color_stop}" printf '%s\n' "${_log_ts} <=== ${_color_start}Time elapsed: ${_run_elapsed_time}s, remaining: ~${_run_time_remaining}s (~${_run_time_per_job}s/job, ~$(bytes_to_human_size ${_run_bytes_per_second})/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 '${OPT_DATEFORMAT}') ==> [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' 'bc' 'tr' 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). if [ "${_host}" = "${_FIRST_HOST}" ] then "${SSH_BIN}" "${_host}" "${OPT_JOBSSHELL} -c 'mkdir -p \"${OPT_DSTURL}\" && \ ${SUDO} touch \"${OPT_DSTURL}/${FPSYNC_RUNID}\"' 2>/dev/null" || \ end_die "Cannot create witness file on target ${_host}: ${OPT_DSTURL}/${FPSYNC_RUNID}" fi 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 ($(bytes_to_human_size $(run_total_size_count "${FPSYNC_RUNID}")), $(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

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

1977

1978

1979

1980

1981

1982

1983

1984

1985

1986

1987

1988

1989

1990

1991

1992

1993

1994

1995

1996

1997

1998

1999

2000

2001

2002

2003

2004

2005

2006

2007

2008

2009

2010

2011

2012

2013

2014

2015

2016

2017

2018

2019

2020

2021

2022

2023

2024

2025

2026

2027

2028

2029

2030

2031

2032

2033

2034

2035

2036

2037

2038

2039

2040

2041

2042

2043

2044

2045

2046

2047

2048

2049

2050

#!/bin/sh

# Copyright (c) 2014-2025 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.7.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=""

# Date format used for log prefix (-Z)

OPT_DATEFORMAT="+%s"

# 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 version

version () {

cat << EOF

fpsync v${FPSYNC_VERSION} - Sync directories in parallel using fpart

Copyright (c) 2014-2025 Ganael LAPLANCHE <ganael.laplanche@martymac.org>

WWW: https://www.fpart.org, https://contribs.martymac.org

EOF

}

# Print help

usage () {

version

cat << EOF

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.

-Z prefix logs with human-readable dates instead of timestamps

-v verbose mode (default: quiet)

This option can be be specified several times to

increase verbosity level.

-h this help

-V print version

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 "${OPT_DATEFORMAT}")

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]?$'

}

# Convert $1 (in bytes) to a human-readable size

bytes_to_human_size () {

if ! is_num "$1"

then

echo "$1"

return

fi

local _result=$(echo "$1" | bc 2>/dev/null) # Remove leading 0s

local _sufx="B,KiB,MiB,GiB,TiB,PiB"

local _round_max=$(( $(echo ${_sufx} | tr -cd ',' | \

wc -c | awk '{print $1}') + 1 ))

local _round=1

while [ "${_round}" -lt "${_round_max}" ] && \

[ $(echo "${_result} >= 1024" | bc 2>/dev/null) -eq 1 ]

do

_result=$(echo "scale=2; ${_result} / 1024" | bc 2>/dev/null)

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