phymv

#!/usr/bin/env bash

show_help()
{
  cat <<EOF

$ExecName version $Version

Usage:
 $ExecName [options] <src_resc> <dest_resc>

Moves the files from one resource to another. It will not move any file
associated with a data object that has a replica on the destination resource.

Parameters:
 <src_resc>   the resource where the files are moved from
 <dest_resc>  the resouce where the files are moved to

Options:
 -c, --collection <collection>  only move the files associated with data objects
                                in the collection <collection>
 -d, --dbms <host>              connect to the ICAT's DBMS on the host <host>
                                instead of the PostgreSQL default
 --dbms-port <port>             connect to the ICAT's DBMS listening on TCP port
                                <port> instead of the PostgreSQL default
 -m, --multiplier <multiplier>  a multiplier on the number of processes to run
                                at once
 -U, --user <user>              authorize the DBMS connection as user <user>
                                instead of the default

 -h, --help     show help and exit
 -v, --version  show version and exit
EOF
}


ExecName=$(basename "$0")
readonly ExecName

readonly Version=1


show_version()
{
  printf '%s\n' "$Version"
}


exit_with_help()
{
  show_help >&2
  exit 1
}


finish()
{
  local objList="$1"

  rm --force "$objList"
  eval "exec 1>&$Log $Log>&-"
}


count_list()
{
  awk 'BEGIN {
         RS = "\0"
         tot = 0
       }

       { tot = tot + 1 }

       END { print tot }'
}


count_unmovable()
{
  local baseCond="$1"

  psql --no-align --quiet --tuples-only ICAT  <<EOSQL
BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ;


CREATE TEMPORARY TABLE storage_root_mapping(storage_id, storage_name, root_name) ON COMMIT DROP AS
WITH RECURSIVE child_mapping AS (
  SELECT
      resc_id AS id,
      resc_name AS name,
      (resc_net != 'EMPTY_RESC_HOST') AS storage,
      resc_name AS root
    FROM r_resc_main
    WHERE resc_parent = '' AND resc_name != 'bundleResc'
  UNION SELECT r.resc_id, r.resc_name, r.resc_net != 'EMPTY_RESC_HOST', m.root
    FROM r_resc_main AS r JOIN child_mapping AS m ON m.id::TEXT = r.resc_parent )
SELECT id, name, root FROM child_mapping WHERE storage;

CREATE INDEX idx_storage_root_mapping_storage ON storage_root_mapping(root_name);


SELECT COUNT(data_id)
  FROM r_data_main AS d JOIN r_coll_main AS c ON c.coll_id = d.coll_id
  WHERE d.data_id = ANY(ARRAY(
      SELECT data_id
        FROM r_data_main
        WHERE resc_id = ANY(ARRAY(
          SELECT storage_id FROM storage_root_mapping WHERE root_name = '$SrcResc' ))
      INTERSECT SELECT data_id
        FROM r_data_main
        WHERE resc_id = ANY(ARRAY(
          SELECT storage_id FROM storage_root_mapping WHERE root_name = '$DestResc' ))))
    AND ($baseCond);


ROLLBACK;
EOSQL
}


partition()
{
  local minSizeB="$1"

  if [ "$#" -ge 2 ]
  then
    local maxSizeB="$2"
  fi

  if [ -n "$maxSizeB" ]
  then
    awk --assign min="$minSizeB" --assign max="$maxSizeB" \
        'BEGIN {
           RS = "\0"
           FS = " "
           ORS = "\0"
         }

         {
           if ($1 >= min && $1 < max) { print substr($0, length($1) + 2) }
         }'
  else
    awk --assign min="$minSizeB" \
        'BEGIN {
           RS = "\0"
           FS = " "
           ORS = "\0"
         }

         {
           if ($1 >= min) { print substr($0, length($1) + 2) }
         }'
  fi
}


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

  local subCnt=0
  local msg=

  while read -r
  do
    ((++subCnt))
    ((++cnt))
    printf '\r%*s\r' "${#msg}" '' >&2
    printf -v msg \
           'cohort: %0*d/%d, all: %0*d/%d' \
           "${#subTot}" "$subCnt" "$subTot" "${#tot}" "$cnt" "$tot"
    printf '%s' "$msg" >&2
  done

  printf '\r%*s\rcohort: %0*d/%d, all: %0*d/%d\n' \
         "${#msg}" '' "${#subTot}" "$subCnt" "$subTot" "${#tot}" "$cnt" "$tot" \
      >&2

  printf '%s' "$cnt"
}


select_cohort()
{
  local cnt="$1"
  local tot="$2"
  local maxProcs="$3"
  local minThreads="$4"

  if [ "$#" -ge 5 ]
  then
    local maxThreads="$5"
  fi

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

  local cohortList
  cohortList=$(mktemp)

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

    partition "$minSizeB" "$maxSizeB"
  else
    partition "$minSizeB"
  fi > "$cohortList"

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

  if [ -n "$maxSizeMiB" ]
  then
    printf 'Physically moving %s files with size in [%s, %s) MiB\n' \
           "$subTotal" "$minSizeMiB" "$maxSizeMiB" \
      >&2
  else
    printf 'Physically moving %s files with size >= %s MiB\n' "$subTotal" "$minSizeMiB" >&2
  fi

  if [ "$subTotal" -gt 0 ]
  then
    local maxArgs=$(( 2 * maxProcs**2 ))
    maxProcs=$((maxProcs * ProcMult))

    xargs --null --max-args "$maxArgs" --max-procs "$maxProcs" \
          iphymv -M -v -R "$DestResc" -S "$SrcResc" \
        < "$cohortList" \
        2>&"$Log" \
        | tee >(cat >&"$Log") \
        | track_prog "$cnt" "$tot" "$subTotal"
  else
    printf '%s\n' "$cnt"
  fi

  rm --force "$cohortList"
}


set -e

Opts=$( \
  getopt \
    --name "$ExecName" \
    --longoptions collection:,dbms:,dbms-port:,help,multiplier:,user:,version \
    --options c:d:hm:U:v \
    -- \
    "$@")
readonly Opts

ret="$?"
if [ "$ret" -ne 0 ]
then
  printf '\n' >&2
  exit_with_help
fi

eval set -- "$Opts"

while true
do
  case "$1" in
    -c|--collection)
      readonly BaseColl="$2"
      shift 2
      ;;
    -d|--dbms)
      export PGHOST="$2"
      shift 2
      ;;
    --dbms-port)
      export PGPORT="$2"
      shift 2
      ;;
    -h|--help)
      show_help
      exit 0
      ;;
    -m|--multiplier)
      readonly ProcMult="$2"
      shift 2
      ;;
    -U|--user)
      export PGUSER="$2"
      shift 2
      ;;
    -v|--version)
      show_version
      exit 0
      ;;
    --)
      shift
      break
      ;;
    *)
      exit_with_help
      ;;
  esac
done

if [ -n "$ProcMult" ]
then
  if ! [[ "$ProcMult" =~ ^[1-9][0-9]* ]]
  then
    printf 'The -m option value must be a positive number. The given value was %s.\n' "$ProcMult" \
      >&2
    exit 1
  fi
else
  readonly ProcMult=1
fi

if [ "$#" -lt 2 ]
then
  exit_with_help
fi

readonly SrcResc="$1"
readonly DestResc="$2"

# Redirect stadout to FD 3 to use as a logging channel
readonly Log=3
eval "exec $Log>&1"

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

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

ObjectList=$(mktemp)
readonly ObjectList
trap 'finish "$ObjectList"' EXIT

if [ -n "$BaseColl" ]
then
  readonly BaseCond="c.coll_name = '$BaseColl' OR c.coll_name LIKE '$BaseColl/%'"
else
  readonly BaseCond=TRUE
fi

printf 'Checking to see if all data objects can be physically moved...\n' >&2

UnmovableCount=$(count_unmovable "$BaseCond")
readonly UnmovableCount

if [ "$UnmovableCount" -gt 0 ]
then
  cat <<EOF >&2
WARNING: NOT ALL DATA OBJECTS COULD BE MOVED BECAUSE REPLICAS ARE ALREADY ON THE
DESTINATION RESOURCE
EOF
fi

printf 'Retrieving data objects to physically move...\n' >&2

psql --no-align --quiet --tuples-only --record-separator-zero --field-separator ' ' ICAT \
<<EOSQL > "$ObjectList"
BEGIN;

CREATE TEMPORARY TABLE storage_root_mapping(storage_id, storage_name, root_name) ON COMMIT DROP AS
WITH RECURSIVE child_mapping AS (
  SELECT
      resc_id AS id,
      resc_name AS name,
      (resc_net != 'EMPTY_RESC_HOST') AS storage,
      resc_name AS root
    FROM r_resc_main
    WHERE resc_parent = '' AND resc_name != 'bundleResc'
  UNION SELECT r.resc_id, r.resc_name, r.resc_net != 'EMPTY_RESC_HOST', m.root
    FROM r_resc_main AS r JOIN child_mapping AS m ON m.id::TEXT = r.resc_parent )
SELECT id, name, root FROM child_mapping WHERE storage;

CREATE INDEX idx_storage_root_mapping_storage ON storage_root_mapping(root_name);


CREATE TEMPORARY TABLE src_data (id) AS
SELECT data_id
  FROM r_data_main
  WHERE resc_id = ANY(ARRAY(
    SELECT storage_id FROM storage_root_mapping WHERE root_name = '$SrcResc' ))
EXCEPT SELECT data_id
  FROM r_data_main
  WHERE resc_id = ANY(ARRAY(
    SELECT storage_id FROM storage_root_mapping WHERE root_name = '$DestResc' ));

CREATE INDEX src_data_idx ON src_data (id);


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
  JOIN src_data AS s ON s.id = d.data_id
WHERE d.resc_id = ANY(ARRAY(
    SELECT storage_id FROM storage_root_mapping WHERE root_name = '$SrcResc' ))
  AND ($BaseCond);

ROLLBACK;
EOSQL

Tot=$(count_list < "$ObjectList")
readonly Tot

printf '%d data objects to physically move\n' "$Tot" >&2

if [ "$Tot" -gt 0 ]
then
  cnt=0
  cnt=$(select_cohort "$cnt" "$Tot" 16   0  1 < "$ObjectList")  # 16 1-threaded transfers
  cnt=$(select_cohort "$cnt" "$Tot"  8   1  2 < "$ObjectList")  # 8 2-threaded
  cnt=$(select_cohort "$cnt" "$Tot"  6   2  3 < "$ObjectList")  # 6 3-threaded
  cnt=$(select_cohort "$cnt" "$Tot"  4   3  5 < "$ObjectList")  # 4 4--5-threaded
  cnt=$(select_cohort "$cnt" "$Tot"  3   5  7 < "$ObjectList")  # 3 6--7-threaded
  cnt=$(select_cohort "$cnt" "$Tot"  2   7 15 < "$ObjectList")  # 2 8--15-threaded
  cnt=$(select_cohort "$cnt" "$Tot"  1  15    < "$ObjectList")  # 1 16-threaded
fi