repl

#!/usr/bin/env bash

show_help() {
	cat <<EOF

$EXEC_NAME version $VERSION

Usage:
 $EXEC_NAME [options]

This program replicates data objects that were created before yesterday. It only
replicates objects to the DEST-RESC if a replica doesn't already exist there. It
writes progress to standard error and all messages, error or otherwise, to
standard out.

For streaming replication, the number of concurrent irepl calls will
approximately be MULTIPLIER*16 with at most 10 data objects being replicated per
call.

For threaded replication, the data objects are broken into cohorts based on the
number of threads iRODS will use to perform the transfer. Objects of size zero
and those with size at least 1500 MiB are handled separately. For a cohort
whose members will be transfered using N threads, the number of concurrent irepl
calls will be approximately MULTIPLIER*16/N with approximately 1000/N^2 objects
being replicated per call.

For data objects that have a size of at least 1500 MiB, the number of concurrent
irepl calls will approximately be MULTIPLIER*MAX with one object being
replicated per call.

A data object with size zero could either be empty or could still be in the
process of being created. Size zero data objects are handled last to allow for
any in-flight data to finish being written to storage. The number of concurrent
irepl calls will approximately be MULTIPLIER*8 with 500 objects being replicated
per call.

Options:
 -A, --age AGE                   how many days old a data object must be to be
                                 replicated, default: 1
 -C, --collection COLLECTION     only replicate the data objects in this
                                 collection
 -d, --debug                     don't clean up temp files
 -h, --help                      show help and exit
 -H, --host HOST                 connect to the ICAT's DBMS on the host HOST
                                 instead of the PostgreSQL default
 -N, --max-transfer-threads MAX  sets the maximum number of threads for parallel
                                 replication to MAX; if MAX is 0 replication
                                 will be streaming.
 -M, --multiplier MULTIPLIER     a multiplier on the number of processes to run
                                 at once, default: 1
 -P, --port PORT                 connect to the ICAT's DBMS listening on TCP
                                 port PORT instead of the PostgreSQL default
 -R, --dest-resc DEST-RESC       replicate data objects to resource DEST-RESC
                                 instead of default replication resource
 -S, --src-resc SRC-RESC         only replicate the data objects with a file on
                                 the resource SRC-RESC
 -u, --until STOP-TIME           the time to stop replication in a format
                                 understandable by \`date\`
 -U, --user USER                 authorize the DBMS connection as user USER
                                 instead of the default
 -v, --verison                   show version and exit

Environment Variables:
 PGHOST  the default PostgreSQL host
 PGPORT  the default PostgreSQL listening port
 PGUSER  the default PostgreSQL user for authorizating the ICAT DB connection

Output:
The program echos the output of the underlying \`irepl -v\` calls to stdout. It
writes overall progress to stderr in way suitable for display on a terminal.

Prerequisites:
 1) iRODS 4.2.8 or later
 2) The user must be initialized with iRODS as an admin user.
 3) The user must be able to connect to the ICAT DB without providing a
    password.

© 2022, The Arizona Board of Regents on behalf of The University of Arizona. For
license information, see https://cyverse.org/license.
EOF
}

readonly VERSION=8

set -o errexit -o nounset -o pipefail

export PGHOST PGPORT PGUSER

export TIMEDOUT=3

readonly DEFAULT_AGE=1
readonly DEFAULT_MAX_TRANSFER_THREADS=3
readonly DEFAULT_PROC_MULT=1
readonly LOG=3
readonly MAX_DATA_OBJS_PER_IREPL=1000
readonly MAX_THREADS=16
readonly STREAMING_IREPL_ARG_FRAC=100
readonly TRANSFER_BUF_SZ=100

EXEC_NAME="$(basename "$(realpath --canonicalize-missing "$0")")"
readonly EXEC_NAME

DEBUG=''

declare ObjCache

main() {
	declare -A optMap=(
		[age]="$DEFAULT_AGE"
		[collection]=''
		[dest-resc]=''
		[help]=''
		[max-transfer-threads]="$DEFAULT_MAX_TRANSFER_THREADS"
		[multiplier]="$DEFAULT_PROC_MULT"
		[src-resc]=''
		[until]=''
		[version]='' )

	if ! resolve_opts optMap "$@"; then
		show_help >&2
		return 1
	fi

	if [[ -n "${optMap[help]}" ]]; then
		show_help
	elif [[ -n "${optMap[version]}" ]]; then
		printf '%s\n' "$VERSION"
	else
		replicate \
			"${optMap[age]}"  \
			"${optMap[max-transfer-threads]}" \
			"${optMap[multiplier]}" \
			"${optMap[collection]}" \
			"${optMap[src-resc]}" \
			"${optMap[dest-resc]}" \
			"${optMap[until]}"
	fi
}

resolve_opts() {
	local mapVar="$1"
	shift

	local opts
	if ! opts="$(format_opts "$@")"; then
		return 1
	fi

	eval set -- "$opts"

	while true; do
		case "$1" in
			-A|--age)
				eval "$mapVar""[age]='$2'"
				shift 2
				;;
			-C|--collection)
				eval "$mapVar""[collection]='$2'"
				shift 2
				;;
			-d|--debug)
				readonly DEBUG=debug
				shift 1
				;;
			-h|--help)
				eval "$mapVar"'[help]=help'
				shift 1
				;;
			-H|--host)
				PGHOST="$2"
				shift 2
				;;
			-M|--multiplier)
				eval "$mapVar""[multiplier]='$2'"
				shift 2
				;;
			-N|--max-transfer-threads)
				eval "$mapVar""[max-transfer-threads]='$2'"
				shift 2
				;;
			-P|--port)
				PGPORT="$2"
				shift 2
				;;
			-R|--dest-resc)
				eval "$mapVar""[dest-resc]='$2'"
				shift 2
				;;
			-S|--src-resc)
				eval "$mapVar""[src-resc]='$2'"
				shift 2
				;;
			-u|--until)
				eval "$mapVar""[until]='$2'"
				shift 2
				;;
			-U|--user)
				PGUSER="$2"
				shift 2
				;;
			-v|--version)
				eval "$mapVar"'[version]=version'
				shift 1
				;;
			--)
				shift
				break
				;;
			*)
				printf 'INTERNAL ERROR: failed to resolve command line options\n' >&2
				return 1
				;;
		esac
	done
}

format_opts() {
	local longOpts=(
		age:
		collection:
		debug
		help
		host:
		max-transfer-threads:
		multiplier:
		port:
		dest-resc:
		src-resc:
		until:
		user:
		version )

	local longOptsStr
	longOptsStr="$(printf '%s\n' "${longOpts[@]}" | paste --serial --delimiter=,)"
	getopt --name="$EXEC_NAME" --longoptions="$longOptsStr" --options=A:dC:hH:M:N:P:R:S:u:U:v -- "$@"
}

replicate() {
	local age="$1"
	local maxTransThreads="$2"
	local procMult="$3"
	local baseColl="$4"
	local srcResc="$5"
	local destResc="$6"
	local until="$7"

	local untilTS
	if ! untilTS="$(compute_until_timestamp "$until")"; then
		return 1
	fi

	validate_repl_conds "$procMult" "$srcResc" "$destResc" "$untilTS"
	prep_logging
	trap 'finish "$ObjCache" $?' EXIT
	ObjCache="$(mktemp)"

	printf 'retrieving data objects to replicate...\n' >&2
	if ! get_data_to_repl "$age" "$srcResc" "$destResc" "$baseColl" > "$ObjCache"; then
		printf 'failed to retrieve data objects\n' >&2
		return 1
	fi

	if [[ -n "$DEBUG" ]]; then
		printf 'DEBUG: object list in %s\n' "$ObjCache" >&2
	fi

	local totObjs
	totObjs="$(count < "$ObjCache")"
	printf '%d data objects to replicate\n' "$totObjs" >&2

	if (( totObjs == 0 )); then
		return 0
	fi

	local totRepl
	if (( maxTransThreads > 0 )); then
		totRepl="$(replicate_threaded \
			"$maxTransThreads" "$procMult" "$srcResc" "$destResc" "$untilTS" "$totObjs" "$ObjCache")"
	else
		totRepl="$(replicate_streaming \
			"$procMult" "$srcResc" "$destResc" "$untilTS" "$totObjs" "$ObjCache")"
	fi

	printf 'Replicated %s data objects\n' "$totRepl" >&2
}

finish() {
	local objCache="$1"
	local exitCode="$2"

	if [[ -z "$DEBUG" && -n "${objCache-}" ]]; then
		rm --force "$objCache"
	fi

	eval "exec 1>&$LOG $LOG>&-"

	if (( exitCode == TIMEDOUT )); then
		printf 'out of time\n' >&2
		exit 0
	fi

	exit "$exitCode"
}

compute_until_timestamp() {
	local until="$1"

	if [[ -n "$until" ]]; then
		if ! date --date="$until" '+%s'; then
			printf 'until "%s" is not a valid date format\n' "$until" >&2
			return 1
		fi
	fi
}

get_data_to_repl() {
	local age="$1"
	local srcResc="$2"
	local destResc="$3"
	local baseColl="$4"

	local replQuery
	replQuery="$(mk_repl_query "$age" "$baseColl" "$srcResc" "$destResc")"

	psql --no-align --tuples-only --record-separator-zero \
		--command="$replQuery" --field-separator=' ' \
		ICAT
}

mk_repl_query() {
	local age="$1"
	local baseColl="$2"
	local srcResc="$3"
	local destResc="$4"

	cat <<EOSQL
SELECT d.data_size, c.coll_name || '/' || d.data_name
FROM r_data_main AS d JOIN r_coll_main AS c ON c.coll_id = d.coll_id
WHERE d.data_id IN (SELECT data_id FROM r_data_main GROUP BY data_id HAVING COUNT(resc_id) = 1)
	AND d.create_ts < '$(mk_max_create_ts "$age")'
	AND ($(mk_coll_cond c.coll_name "$baseColl"))
	AND ($(mk_resc_cond d.resc_id "$srcResc" "$destResc"))
EOSQL
}

mk_coll_cond() {
	local collNameCol="$1"
	local baseColl="$2"

	if [[ -n "$baseColl" ]]; then
		printf $'%s SIMILAR TO \'%s(/%%)?\'' "$collNameCol" "$baseColl"
	else
		printf TRUE
	fi
}

mk_max_create_ts() {
	local age="$1"

	date --date="$(date --iso-8601 --date="$age days ago")" '+0%s'
}

mk_resc_cond() {
	local rescCol="$1"
	local srcResc="$2"
	local destResc="$3"

	if [[ -n "$srcResc" ]]; then
		mk_resc_restrict_cond "$rescCol" "$srcResc"
	elif [[ -n "$destResc" ]]; then
		mk_resc_restrict_cond "$rescCol" "$destResc"
	else
		printf TRUE
	fi
}

mk_resc_restrict_cond() {
	local rescCol="$1"
	local resc="$2"

cat <<SQL
$rescCol IN (
	WITH RECURSIVE child_mapping AS (
		SELECT resc_id AS id, (resc_net != 'EMPTY_RESC_HOST') AS storage, resc_name AS root
			FROM r_resc_main
			WHERE resc_name = '$resc'
		UNION SELECT r.resc_id, r.resc_net != 'EMPTY_RESC_HOST', m.root
			FROM r_resc_main AS r
				JOIN child_mapping AS m
					ON m.id = CASE WHEN r.resc_parent::TEXT = ''
						THEN NULL
						ELSE r.resc_parent::BIGINT END )
	SELECT id FROM child_mapping WHERE storage )
SQL
}

# Redirect stdout to FD 3 to use as a logging channel
prep_logging() {
	eval "exec $LOG>&1"
}

replicate_streaming() {
	local procMult="$1"
	local srcResc="$2"
	local destResc="$3"
	local untilTS="$4"
	local totObjs="$5"
	local objCache="$6"

	if ! CHECK_TIME "$untilTS"; then
		return "$TIMEDOUT"
	fi

	local maxProcs=$(( MAX_THREADS * procMult ))

	local maxIreplArgs
	maxIreplArgs="$(div_ceil "$MAX_DATA_OBJS_PER_IREPL" "$STREAMING_IREPL_ARG_FRAC")"

	local minSizeMiB maxSizeMiB
	minSizeMiB="$(threads_to_min_MiB -1)"
	maxSizeMiB="$(threads_to_max_MiB -1)"

	local cohortList
	cohortList="$(mktemp)"
	select_cohort "$minSizeMiB" "$maxSizeMiB" < "$objCache" > "$cohortList"

	local subTotal
	subTotal="$(count <"$cohortList")"

	if (( minSizeMiB <= 0 )) && [[ -z "$maxSizeMiB" ]]; then
		disp_cohort_info "$subTotal" "$maxProcs" "$maxIreplArgs" 0 '' includeMin
	elif (( minSizeMiB <= 0 )); then
		disp_cohort_info "$subTotal" "$maxProcs" "$maxIreplArgs" "$minSizeMiB" "$maxSizeMiB"
	else
		disp_cohort_info \
			"$subTotal" "$maxProcs" "$maxIreplArgs" "$minSizeMiB" "$maxSizeMiB" includeMin
	fi >&2

	if (( subTotal > 0 )); then
		parallel --line-buffer --no-notice --null \
					--halt 2 --max-args "$maxIreplArgs" --max-procs "$maxProcs" \
					REPL_BATCH "'$srcResc'" "'$destResc'" 0 "'$untilTS'" \
				< "$cohortList" \
				2>&"$LOG" \
			| tee >(cat >&"$LOG") \
			| track_prog 0 "$totObjs" "$subTotal"
	else
		echo 0
	fi

	rm --force "$cohortList"
}

partition() {
	local minSizeB="$1"
	local maxSizeB="$2"

	local objEntry
	while IFS= read -r -d '' objEntry; do
		local size="${objEntry%% *}"
		local path="${objEntry#* }"

		if ([[ -z "$maxSizeB" ]] && (( size >= minSizeB ))) \
			|| (( size >= minSizeB && size < maxSizeB ))
		then
			printf '%s\0' "$path"
		fi
	done
}

threads_to_max_MiB() {
	local threads="$1"

	if (( threads == -1 )); then
		echo ''
	elif (( threads == 0 )); then
		echo 0
	else
		echo $(( threads * TRANSFER_BUF_SZ ))
	fi
}

clear_msg() {
	local msg="$1"

	printf '\r%*s\r' ${#msg} ''
}

mk_prog_msg() {
	local count="$1"
	local total="$2"
	local subCount="$3"
	local subTotal="$4"

	printf 'cohort: %0*d/%d, all: %0*d/%d' \
		${#subTotal} "$subCount" "$subTotal" ${#total} "$count" "$total"
}




replicate_threaded() {
	local maxThreads="$1"
	local procMult="$2"
	local srcResc="$3"
	local destResc="$4"
	local untilTS="$5"
	local totObjs="$6"
	local objCache="$7"

	local cnt=0

	cnt="$(replicate_small_files \
		"$maxThreads" "$procMult" "$srcResc" "$destResc" "$untilTS" "$totObjs" "$objCache" "$cnt" )"

	cnt="$(replicate_large_files \
		"$maxThreads" "$procMult" "$srcResc" "$destResc" "$untilTS" "$totObjs" "$objCache" "$cnt" )"

	cnt="$(replicate_empty_files \
		"$maxThreads" "$procMult" "$srcResc" "$destResc" "$untilTS" "$totObjs" "$objCache" "$cnt" )"

	echo "$cnt"
}

replicate_empty_files() {
	local maxThreads="$1"
	local procMult="$2"
	local srcResc="$3"
	local destResc="$4"
	local untilTS="$5"
	local totObjs="$6"
	local objCache="$7"
	local replCnt="$8"

	local maxProcs maxIreplArgs
	maxProcs="$(div_ceil $(( MAX_THREADS * procMult )) 2)"
	maxIreplArgs="$(div_ceil "$MAX_DATA_OBJS_PER_IREPL" 2)"

	local minSizeMiB=0
	local maxSizeMiB=0

	local cohortList
	cohortList="$(mktemp)"
	select_cohort 0 0 < "$objCache" > "$cohortList"

	local subTotal
	subTotal="$(count < "$cohortList")"

	disp_cohort_info "$subTotal" "$maxProcs" "$maxIreplArgs" "$minSizeMiB" "$maxSizeMiB" >&2

	if (( subTotal > 0 )); then
		replCnt="$(parallel --line-buffer --no-notice --null \
					--halt 2 --max-args "$maxIreplArgs" --max-procs "$maxProcs" \
					REPL_BATCH "'$srcResc'" "'$destResc'" "'$maxThreads'" "'$untilTS'" \
				< "$cohortList" \
				2>&"$LOG" \
			| tee >(cat >&"$LOG") \
			| track_prog "$replCnt" "$totObjs" "$subTotal" )"
	fi

	rm --force "$cohortList"
	echo "$replCnt"
}

replicate_large_files() {
	local maxThreads="$1"
	local procMult="$2"
	local srcResc="$3"
	local destResc="$4"
	local untilTS="$5"
	local totObjs="$6"
	local objCache="$7"
	local replCnt="$8"

	local minSize
	minSize="$(threads_to_min_MiB "$maxThreads")"

	local lastCohortMinSz=$(( TRANSFER_BUF_SZ * (MAX_THREADS - 1) ))

	local maxIreplArgs
	maxIreplArgs="$(max_args_per_irepl "$maxThreads")"

	# still small enough that they wouldn't default to using MAX_THREADS
	if (( minSize < lastCohortMinSz )); then
		replCnt="$(select_cohort_by_size \
				"$replCnt" \
				"$totObjs" \
				"$srcResc" \
				"$destResc" \
				"$procMult" \
				"$maxThreads" \
				"$maxIreplArgs" \
				"$untilTS" \
				"$minSize" \
				"$lastCohortMinSz" \
			< "$objCache" )"
	fi

	# transfer files that would default to using MAX_THREADS
	maxIreplArgs=1

	replCnt="$(select_cohort_by_size \
			"$replCnt" \
			"$totObjs" \
			"$srcResc" \
			"$destResc" \
			"$procMult" \
			"$maxThreads" \
			"$maxIreplArgs" \
			"$untilTS" \
			"$lastCohortMinSz" \
		< "$objCache" )"

	echo "$replCnt"
}

replicate_small_files() {
	local maxThreads="$1"
	local procMult="$2"
	local srcResc="$3"
	local destResc="$4"
	local untilTS="$5"
	local totObjs="$6"
	local objCache="$7"
	local replCnt="$8"

	local threadsPerDataObj
	for (( threadsPerDataObj=1; threadsPerDataObj<maxThreads; threadsPerDataObj++ )); do
		if ! CHECK_TIME "$untilTS"; then
			return "$TIMEDOUT"
		fi

		local maxIreplArgs maxProcs
		maxProcs="$(max_procs "$threadsPerDataObj" "$procMult")"
		maxIreplArgs="$(max_args_per_irepl "$threadsPerDataObj")"

		local minSizeMiB maxSizeMiB
		minSizeMiB="$(threads_to_min_MiB "$threadsPerDataObj")"
		maxSizeMiB="$(threads_to_max_MiB "$threadsPerDataObj")"

		local cohortList
		cohortList="$(mktemp)"
		select_cohort "$minSizeMiB" "$maxSizeMiB" < "$objCache" > "$cohortList"

		local subTotal
		subTotal="$(count <"$cohortList")"

		if (( minSizeMiB <= 0 )) && [[ -z "$maxSizeMiB" ]]; then
			disp_cohort_info "$subTotal" "$maxProcs" "$maxIreplArgs" 0 '' includeMin
		elif (( minSizeMiB <= 0 )); then
			disp_cohort_info "$subTotal" "$maxProcs" "$maxIreplArgs" "$minSizeMiB" "$maxSizeMiB"
		else
			disp_cohort_info \
				"$subTotal" "$maxProcs" "$maxIreplArgs" "$minSizeMiB" "$maxSizeMiB" includeMin
		fi >&2

		if (( subTotal > 0 )); then
			replCnt="$(parallel --line-buffer --no-notice --null \
						--halt 2 --max-args "$maxIreplArgs" --max-procs "$maxProcs" \
						REPL_BATCH "'$srcResc'" "'$destResc'" "'$threadsPerDataObj'" "'$untilTS'" \
					< "$cohortList" \
					2>&"$LOG" \
				| tee >(cat >&"$LOG") \
				| track_prog "$replCnt" "$totObjs" "$subTotal" )"
		fi

		rm --force "$cohortList"
	done

	echo "$replCnt"
}




validate_repl_conds() {
	local procMult="$1"
	local srcResc="$2"
	local destResc="$3"
	local untilTS="$4"

	if [[ -n "$procMult" ]] && ! [[ "$procMult" =~ ^[1-9][0-9]* ]]; then
		printf 'The multiplier must be a positive number. The given value was %s.\n' "$procMult" >&2
		return 1
	fi

	if [[ -n "$destResc" && "$srcResc" == "$destResc" ]]; then
		printf 'The destination resource, %s, cannot be the same as the source resource, %s.\n' \
				"$destResc" "$srcResc" \
			>&2

		return 1
	fi

	if ! CHECK_TIME "$untilTS"; then
		printf 'Stop time is in the past\n' >&2
		return 1
	fi

	if ! iadmin lz &> /dev/null; then
		printf "aren't authenticated as a rodsadmin user\n" >&2
		return 1
	fi

	if ! psql ICAT <<<'' &> /dev/null; then
		printf "aren't able to connect to the ICAT DB without a password\n" >&2
		return 1
	fi
}



count() {
	local tot=0

	while read -r -d ''; do
		: $(( tot++ ))
	done

	echo "$tot"
}

disp_cohort_info() {
	local cohortSize="$1"
	local maxProcs="$2"
	local maxArgs="$3"
	local minObjSizeMiB="$4"
	local maxObjSizeMiB="$5"

	local includeMin=''
	if (( $# >= 6 )); then
		includeMin="$6"
	fi

	if [[ -z "$minSizeMiB" ]] || (( ${minSizeMiB-0} < 0 )); then
		minSizeMiB=0
	fi

	local objSizeInfo
	if [[ -z "$maxObjSizeMiB" ]]; then
		if [[ -n "$includeMin" ]]; then
			if (( minObjSizeMiB == 0 )); then
				objSizeInfo=''
			else
				printf -v objSizeInfo 'with size >= %s MiB' "$minObjSizeMiB"
			fi
		else
			printf -v objSizeInfo 'with size > %s MiB' "$minObjSizeMiB"
		fi
	elif (( minObjSizeMiB < maxObjSizeMiB )); then
		if [[ -n "$includeMin" ]]; then
			printf -v objSizeInfo 'with size in [%s, %s) MiB' "$minObjSizeMiB" "$maxObjSizeMiB"
		else
			printf -v objSizeInfo 'with size in (%s, %s) MiB' "$minObjSizeMiB" "$maxObjSizeMiB"
		fi
	else
		printf -v objSizeInfo 'with size %s MiB' "$minObjSizeMiB"
	fi

	printf 'Replicating %s files %s (%s procs, %s objects/proc)\n' \
		"$cohortSize" "$objSizeInfo" "$maxProcs" "$maxArgs"
}

div_ceil() {
	local numerator="$1"
	local denominator="$2"

	echo $(( 1 + (numerator - 1) / denominator ))
}

max_args_per_irepl() {
	local threadsPerDataObj="$1"

	div_ceil "$MAX_DATA_OBJS_PER_IREPL" $(( threadsPerDataObj * threadsPerDataObj ))
}

select_cohort() {
	local minSizeMiB="$1"
	local maxSizeMiB="$2"

	local maxSizeB minSizeB
	if [[ -z "$maxSizeMiB" ]]; then
		minSizeB=0
		maxSizeB=
	elif (( maxSizeMiB == 0 )); then
		minSizeB=0
		maxSizeB=1
	else
		if (( minSizeMiB == 0 )); then
			minSizeB=1
		else
			minSizeB=$(( minSizeMiB * 1024 ** 2 ))
		fi

		maxSizeB=$(( maxSizeMiB * 1024 ** 2 ))
	fi

	partition "$minSizeB" "$maxSizeB"
}

# This is intended for large files.
select_cohort_by_size() {
	local cnt="$1"
	local tot="$2"
	local srcResc="$3"
	local destResc="$4"
	local procMult="$5"
	local maxTransThreads="$6"
	local maxArgs="$7"
	local untilTS="$8"
	local minSizeMiB="$9"

	local maxSizeMiB=
	if (( $# >= 10 )); then
		maxSizeMiB="${10}"
	fi

	local maxProcs
	maxProcs="$(max_procs "$maxTransThreads" "$procMult")"

	if ! CHECK_TIME "$untilTS"; then
		return "$TIMEDOUT"
	fi

	local minSizeB=$(( minSizeMiB * 1024 ** 2 ))

	local maxSizeB=
	if [[ -n "$maxSizeMiB" ]]; then
		maxSizeB=$(( maxSizeMiB * 1024 ** 2 ))
	fi

	local cohortList
	cohortList="$(mktemp)"
	partition "$minSizeB" "$maxSizeB" > "$cohortList"

	local subTotal
	subTotal="$(count < "$cohortList")"

	disp_cohort_info "$subTotal" "$maxProcs" "$maxArgs" "$minSizeMiB" "$maxSizeMiB" includeMin >&2

	if (( subTotal > 0 )); then
		parallel --line-buffer --no-notice --null \
					--halt 2 --max-args "$maxArgs" --max-procs "$maxProcs" \
					REPL_BATCH "'$srcResc'" "'$destResc'" "'$maxTransThreads'" "'$untilTS'" \
				< "$cohortList" \
				2>&"$LOG" \
			| tee >(cat >&"$LOG") \
			| track_prog "$cnt" "$tot" "$subTotal"
	else
		printf '%s\n' "$cnt"
	fi

	rm --force "$cohortList"
}

max_procs() {
	local threadsPerDataObj="$1"
	local procMult="$2"

	div_ceil $(( MAX_THREADS * procMult )) "$threadsPerDataObj"
}

threads_to_min_MiB() {
	local threads="$1"

	if (( threads <= 0 )); then
		echo 0
	else
		echo $(( (threads - 1) * TRANSFER_BUF_SZ ))
	fi
}

track_prog() {
	local cnt="$1"
	local tot="$2"
	local subTot="$3"

	local subCnt=0

	local msg

	msg="$(mk_prog_msg "$cnt" "$tot" "$subCnt" "$subTot")"
	printf '%s' "$msg" >&2

	local event
	while read -r event; do
		if [[ "$event" != 'cliReconnManager: '* ]]; then
			: $(( subCnt++ ))
			: $(( cnt++ ))
			clear_msg "$msg" >&2
			msg="$(mk_prog_msg "$cnt" "$tot" "$subCnt" "$subTot")"
			printf '%s' "$msg" >&2
		fi
	done

	clear_msg "$msg" >&2
	msg="$(mk_prog_msg "$cnt" "$tot" "$subCnt" "$subTot")"
	printf '%s\n' "$msg" >&2
	printf '%s' "$cnt"
}

REPL_BATCH() {
	set -o errexit -o nounset -o pipefail

	local srcResc="$1"
	local destResc="$2"
	local maxTransThreads="$3"
	local untilTS="$4"
	shift 4

	CHECK_TIME "$untilTS"

	local replArgs=(-B -M -T -v -N "$maxTransThreads")

	if [[ -n "$destResc" ]]; then
		replArgs+=(-R "$destResc")
	fi

	# XXX - irepl -S doesn't work in 4.2.8, but it's fixed in 4.2.9
	#if [[ -n "$srcResc" ]]; then
   #	replArgs+=(-S "$srcResc")
	#fi

	local replStatus=0
	stdbuf --error=L irepl "${replArgs[@]}" "$@" || replStatus=$?

	if (( replStatus != 0 )); then
		printf 'repl: irepl exited with status %d\n' "$replStatus" >&2
	fi

	exit 0
}
export -f REPL_BATCH

CHECK_TIME() {
	set -o errexit -o nounset -o pipefail

	local untilTS="$1"

	if [[ -n "$untilTS" ]] && (( $(date '+%s') >= untilTS )); then
		return "$TIMEDOUT"
	fi
}
export -f CHECK_TIME

main "$@"