diff --git a/performance_profiling/matrix_profile_update.bash b/performance_profiling/matrix_profile_update.bash new file mode 100755 index 00000000..4443803a --- /dev/null +++ b/performance_profiling/matrix_profile_update.bash @@ -0,0 +1,462 @@ +#!/bin/bash +set -e + +# Analagous to CI/CD matrix testing of Python 3.11, 3.12, 3.13, +# here we will do a matrix performance profiling by comparing performance numbers +# for every combination of create/update/update on 3 different directories. + +############################################################################### +# Manually edit parameters here: + +work_dir=/lcrc/group/e3sm/ac.forsyth2/zstash_performance/ +unique_id=profile_update_20251219 + +dir_to_copy_from=/lcrc/group/e3sm/ac.forsyth2/E3SMv2/v2.LR.historical_0201/ +subdir0=build/ +subdir1=run/ +subdir2=init/ + +fresh_globus=true +# ENDPOINT UUIDS: +# LCRC_IMPROV_DTN_ENDPOINT=15288284-7006-4041-ba1a-6b52501e49f1 +# NERSC_PERLMUTTER_ENDPOINT=6bdc7956-fc0f-4ad2-989c-7aa5ee643a79 +# NERSC_HPSS_ENDPOINT=9cd89cfd-6d04-11e5-ba46-22000b92c6ec +# PIC_COMPY_DTN_ENDPOINT=68fbd2fa-83d7-11e9-8e63-029d279f7e24 +# GLOBUS_TUTORIAL_COLLECTION_1_ENDPOINT=6c54cade-bde5-45c1-bdea-f4bd71dba2cc +dst_endpoint_uuid=9cd89cfd-6d04-11e5-ba46-22000b92c6ec +dst_archive_dir=/home/f/forsyth/zstash_performance/ + +############################################################################### +# Utility functions + +# Colors for output +RED='\033[0;31m' +GREEN='\033[0;32m' +YELLOW='\033[1;33m' +BLUE='\033[0;34m' +CYAN='\033[0;36m' +NC='\033[0m' # No Color + +# Functions to print colored messages +print_step() { + echo -e "${CYAN}[STEP]${NC} $1" +} + +print_info() { + echo -e "${BLUE}[INFO]${NC} $1" +} + +print_success() { + echo -e "${GREEN}[SUCCESS]${NC} $1" +} + +print_error() { + echo -e "${RED}[ERROR]${NC} $1" +} + +print_warning() { + echo -e "${YELLOW}[WARNING]${NC} $1" +} + +confirm() +{ + read -p "$1 (y/n): " -n 1 -r + echo + [[ $REPLY =~ ^[Yy]$ ]] +} + +valiate_configuration() +{ + local dir_to_copy_from="${1}" + local subdir0="${2}" + local subdir1="${3}" + local subdir2="${4}" + + print_step "Validating configuration..." + + if [ ! -d "$dir_to_copy_from" ]; then + print_error "Source directory does not exist: $dir_to_copy_from" + exit 1 + fi + + if [ "$subdir0" != "none" ] && [ -n "$subdir0" ]; then + if [ ! -d "${dir_to_copy_from}${subdir0}" ]; then + print_error "subdir0 does not exist: ${dir_to_copy_from}${subdir0}" + exit 1 + fi + fi + + if [ "$subdir1" != "none" ] && [ -n "$subdir1" ]; then + if [ ! -d "${dir_to_copy_from}${subdir1}" ]; then + print_error "subdir1 does not exist: ${dir_to_copy_from}${subdir1}" + exit 1 + fi + fi + + if [ "$subdir2" != "none" ] && [ -n "$subdir2" ]; then + if [ ! -d "${dir_to_copy_from}${subdir2}" ]; then + print_error "subdir2 does not exist: ${dir_to_copy_from}${subdir2}" + exit 1 + fi + fi + + print_success "Configuration validated" +} + +refresh_globus() +{ + print_step "Setting up fresh Globus authentication..." + + # 1. Activate endpoints + echo "Go to https://app.globus.org/file-manager?two_pane=true > For 'Collection', choose the endpoints you're using, and authenticate if needed:" + echo "LCRC Improv DTN, NERSC Perlmutter, NERSC HPSS, pic#compy-dtn" + if ! confirm "Have you authenticated into the correct endpoints?"; then + exit 1 + fi + + # 2. Reset authentication token files + INI_PATH=${HOME}/.zstash.ini + TOKEN_FILE=${HOME}/.zstash_globus_tokens.json + + if [ -f "${INI_PATH}" ]; then + rm -f "${INI_PATH}" + print_info "Removed ${INI_PATH}" + fi + + if [ -f "${TOKEN_FILE}" ]; then + rm -f "${TOKEN_FILE}" + print_info "Removed ${TOKEN_FILE}" + fi + + # 3. Reset Globus consents + echo "https://auth.globus.org/v2/web/consents > Globus Endpoint Performance Monitoring > rescind all" + if ! confirm "Have you revoked Globus consents?"; then + exit 1 + fi + + print_success "Globus authentication reset complete" +} + +############################################################################### +# Core functions + +analyze_directories() { + # Array of directories to analyze + local dirs=("$@") + local cumulative_size=0 + + for dir in "${dirs[@]}"; do + if [[ ! -d "$dir" ]]; then + echo "Warning: '$dir' is not a directory, skipping..." + continue + fi + + echo "========================================" + echo "Analyzing: $dir" + echo "========================================" + + # Get total size in bytes for calculation + local total_size_bytes=$(du -sb "$dir" 2>/dev/null | cut -f1) + + # Add to cumulative total + cumulative_size=$((cumulative_size + total_size_bytes)) + + # Convert to human-readable format + local total_size_human=$(numfmt --to=iec-i --suffix=B "$total_size_bytes" 2>/dev/null || echo "${total_size_bytes}B") + echo "Total size: $total_size_human" + + # 2. Total number of files (excluding directories) + local file_count=$(find "$dir" -type f 2>/dev/null | wc -l) + echo "Number of files: $file_count" + + # 3. Average file size + if [[ $file_count -gt 0 ]]; then + local avg_bytes=$((total_size_bytes / file_count)) + + # Convert to human-readable format + if [[ $avg_bytes -lt 1024 ]]; then + echo "Average file size: ${avg_bytes}B" + elif [[ $avg_bytes -lt 1048576 ]]; then + echo "Average file size: $((avg_bytes / 1024))KB" + elif [[ $avg_bytes -lt 1073741824 ]]; then + echo "Average file size: $((avg_bytes / 1048576))MB" + else + echo "Average file size: $((avg_bytes / 1073741824))GB" + fi + else + echo "Average file size: N/A (no files found)" + fi + + echo "" + done + + # Summary statistics + echo "========================================" + echo "SUMMARY" + echo "========================================" + + local num_dirs=${#dirs[@]} + local cumulative_human=$(numfmt --to=iec-i --suffix=B "$cumulative_size" 2>/dev/null || echo "${cumulative_size}B") + echo "Cumulative size of all directories: $cumulative_human" + + if [[ $num_dirs -gt 1 ]]; then + local permutation_count=3 # We're going to run 3 permutations of these directories + local permutation_size=$((cumulative_size * permutation_count)) + local permutation_human=$(numfmt --to=iec-i --suffix=B "$permutation_size" 2>/dev/null || echo "${permutation_size}B") + + echo "Number of directories: $num_dirs" + echo "Number of permutations we will test: $permutation_count" + echo "Space needed: $permutation_human" + fi +} + +run_create() +{ + local dir_to_copy_from="${1}" + local subdir="${2}" + local archive_dir="${3}" + local dst_endpoint_uuid="${4}" + local dst_archive_subdir="${5}" + local cache_dir="${6}" + local create_log="${7}" + print_step "Starting CREATE operation..." + + print_info "Copying data from ${dir_to_copy_from}${subdir}" + cp -r "${dir_to_copy_from}${subdir}" "${archive_dir}${subdir}" + + print_info "Running zstash create..." + print_info "Command: zstash create --hpss=globus://${dst_endpoint_uuid}/${dst_archive_subdir} --cache=${cache_dir} -v ${archive_dir}" + + if { time zstash create --hpss="globus://${dst_endpoint_uuid}/${dst_archive_subdir}" --cache="${cache_dir}" -v "${archive_dir}" ; } 2>&1 | tee "${create_log}"; then + print_success "zstash create completed successfully" + else + print_error "zstash create failed with exit code $?" + exit 1 + fi + + echo "" + print_warning "Verification suggestion" + echo "Go to your destination machine and run:" + echo " ls ${dst_archive_subdir}" + echo "Expected output: 000000.tar index.db" + echo "" +} + +run_update() +{ + local dir_to_copy_from="${1}" + local subdir="${2}" + local archive_dir="${3}" + local dst_endpoint_uuid="${4}" + local dst_archive_subdir="${5}" + local cache_dir="${6}" + local update_log="${7}" + + print_step "Starting UPDATE operation..." + + print_info "Copying additional data from ${dir_to_copy_from}${subdir}" + cp -r "${dir_to_copy_from}${subdir}" "${archive_dir}${subdir}" + + print_info "Running zstash update..." + print_info "Command: zstash update --hpss=globus://${dst_endpoint_uuid}/${dst_archive_subdir} --cache=${cache_dir} -v" + + if { time zstash update --hpss="globus://${dst_endpoint_uuid}/${dst_archive_subdir}" --cache="${cache_dir}" -v ; } 2>&1 | tee "${update_log}"; then + print_success "zstash update completed successfully" + else + print_error "zstash update failed with exit code $?" + exit 1 + fi + + echo "" + print_warning "Verification suggestion:" + echo "Go to your destination machine and run:" + echo " ls ${dst_archive_subdir}" + echo "Expected output (after update1): 000000.tar 000001.tar index.db" + echo "Expected output (after update2): 000000.tar 000001.tar 000002.tar index.db" + echo "" +} + +############################################################################### +# Main script: + +valiate_configuration $dir_to_copy_from $subdir0 $subdir1 $subdir2 + +if [ "${fresh_globus}" == "true" ]; then + refresh_globus +fi + +analyze_directories "${dir_to_copy_from}${subdir0}" "${dir_to_copy_from}${subdir1}" "${dir_to_copy_from}${subdir2}" +# Array of subdirectories +subdirs=("$subdir0" "$subdir1" "$subdir2") +# Define the 3 specific test configurations +# Each array contains indices into the subdirs array +declare -a test_configs=( + "0 1 2" # 123: subdir0, subdir1, subdir2 + "1 2 0" # 231: subdir1, subdir2, subdir0 + "2 0 1" # 312: subdir2, subdir0, subdir1 +) +declare -a test_labels=("123" "231" "312") +# Store all log files for combined analysis +declare -a all_update1_logs +declare -a all_update2_logs +declare -a test_names + +# Loop through the 3 test configurations +for test_idx in 0 1 2; do + # Parse the configuration + config=(${test_configs[$test_idx]}) + i=${config[0]} + j=${config[1]} + k=${config[2]} + + # Get the subdirectories for this test + create_subdir="${subdirs[$i]}" + update1_subdir="${subdirs[$j]}" + update2_subdir="${subdirs[$k]}" + + # Create a label for this test + test_label="${test_labels[$test_idx]}" + test_names+=("Test_${test_label}") + + print_step "==========================================" + print_step "Running Test ${test_label}" + print_step " Create: $create_subdir" + print_step " Update1: $update1_subdir" + print_step " Update2: $update2_subdir" + print_step "==========================================" + + # Create unique work directories for this test + dst_archive_subdir="${dst_archive_dir}${unique_id}/test${test_label}/" + work_subdir="${work_dir}${unique_id}/test${test_label}/" + mkdir -p "${work_subdir}" + archive_dir="${work_subdir}archive_dir/" + cache_dir="${work_subdir}cache/" + log_dir="${work_subdir}logs/" + mkdir -p "${archive_dir}" + mkdir -p "${cache_dir}" + mkdir -p "${log_dir}" + + # Define log file paths + create_log="${log_dir}create.log" + update1_log="${log_dir}update1.log" + update2_log="${log_dir}update2.log" + + # Store logs for combined analysis + all_update1_logs+=("$update1_log") + all_update2_logs+=("$update2_log") + + print_success "Work directories created at ${work_subdir}" + + cd "${work_subdir}" + run_create "$dir_to_copy_from" "$create_subdir" "$archive_dir" "$dst_endpoint_uuid" "$dst_archive_subdir" "$cache_dir" "$create_log" + + # For update, we need to be in archive_dir: + cd "${archive_dir}" + run_update "$dir_to_copy_from" "$update1_subdir" "$archive_dir" "$dst_endpoint_uuid" "$dst_archive_subdir" "$cache_dir" "$update1_log" + run_update "$dir_to_copy_from" "$update2_subdir" "$archive_dir" "$dst_endpoint_uuid" "$dst_archive_subdir" "$cache_dir" "$update2_log" + + cd "${work_subdir}" + + print_success "Test ${test_label} completed" + echo "" +done + +# Combined analysis of all tests +print_step "==========================================" +print_step "COMBINED PERFORMANCE ANALYSIS" +print_step "==========================================" + +combined_analysis_report="${work_dir}${unique_id}/combined_analysis.txt" + +{ + echo "==========================================" + echo "Combined Performance Analysis" + echo "Generated: $(date)" + echo "==========================================" + echo "" + echo "Test configurations:" + echo " Test 123: create=$subdir0, update1=$subdir1, update2=$subdir2" + echo " Test 231: create=$subdir1, update1=$subdir2, update2=$subdir0" + echo " Test 312: create=$subdir2, update1=$subdir1, update2=$subdir0" + echo "" + echo "==========================================" + echo "" + + # Create markdown table header + echo "| Test/Update | File Gathering | Database Comparison | Add Files |" + echo "| --- | --- | --- | --- |" + + # Extract metrics for each test + for idx in 0 1 2; do + test_name="${test_names[$idx]}" + u1_log="${all_update1_logs[$idx]}" + u2_log="${all_update2_logs[$idx]}" + + # Extract Update 1 metrics + u1_file_gathering=$(grep "File gathering:" "$u1_log" 2>/dev/null | grep -oP '\d+\.\d+s \(\d+\.\d+%\)' | head -1) + u1_db_comparison=$(grep "Database comparison:" "$u1_log" 2>/dev/null | grep -oP '\d+\.\d+s \(\d+\.\d+%\)' | head -1) + u1_add_files=$(grep "Add files:" "$u1_log" 2>/dev/null | grep -oP '\d+\.\d+s \(\d+\.\d+%\)' | head -1) + + # Extract Update 2 metrics + u2_file_gathering=$(grep "File gathering:" "$u2_log" 2>/dev/null | grep -oP '\d+\.\d+s \(\d+\.\d+%\)' | head -1) + u2_db_comparison=$(grep "Database comparison:" "$u2_log" 2>/dev/null | grep -oP '\d+\.\d+s \(\d+\.\d+%\)' | head -1) + u2_add_files=$(grep "Add files:" "$u2_log" 2>/dev/null | grep -oP '\d+\.\d+s \(\d+\.\d+%\)' | head -1) + + # Default to N/A if not found + u1_file_gathering=${u1_file_gathering:-"N/A"} + u1_db_comparison=${u1_db_comparison:-"N/A"} + u1_add_files=${u1_add_files:-"N/A"} + u2_file_gathering=${u2_file_gathering:-"N/A"} + u2_db_comparison=${u2_db_comparison:-"N/A"} + u2_add_files=${u2_add_files:-"N/A"} + + # Print markdown table rows for both updates + echo "| ${test_name}_Update1 | $u1_file_gathering | $u1_db_comparison | $u1_add_files |" + echo "| ${test_name}_Update2 | $u2_file_gathering | $u2_db_comparison | $u2_add_files |" + done + + echo "" + echo "==========================================" + echo "Detailed Metrics by Test" + echo "==========================================" + echo "" + + for idx in 0 1 2; do + test_name="${test_names[$idx]}" + u1_log="${all_update1_logs[$idx]}" + u2_log="${all_update2_logs[$idx]}" + + echo "----------------------------------------" + echo "$test_name" + echo "----------------------------------------" + echo "" + echo "UPDATE 1:" + echo "--------" + grep "File gathering:" "$u1_log" 2>/dev/null || echo " File gathering: N/A" + grep "Database comparison:" "$u1_log" 2>/dev/null || echo " Database comparison: N/A" + grep "Add files:" "$u1_log" 2>/dev/null || echo " Add files: N/A" + grep "TOTAL TIME:" "$u1_log" 2>/dev/null | tail -1 || echo " TOTAL TIME: N/A" + echo "" + echo "UPDATE 2:" + echo "--------" + grep "File gathering:" "$u2_log" 2>/dev/null || echo " File gathering: N/A" + grep "Database comparison:" "$u2_log" 2>/dev/null || echo " Database comparison: N/A" + grep "Add files:" "$u2_log" 2>/dev/null || echo " Add files: N/A" + grep "TOTAL TIME:" "$u2_log" 2>/dev/null | tail -1 || echo " TOTAL TIME: N/A" + echo "" + done + + echo "==========================================" + echo "Individual log files:" + echo "==========================================" + for idx in 0 1 2; do + test_name="${test_names[$idx]}" + echo "${test_name}:" + echo " Update1: ${all_update1_logs[$idx]}" + echo " Update2: ${all_update2_logs[$idx]}" + done + +} | tee "$combined_analysis_report" + +print_success "Combined analysis saved to: $combined_analysis_report" +print_info "All log files listed above for detailed review" diff --git a/performance_profiling/profile_performance_for_update.bash b/performance_profiling/profile_performance_for_update.bash new file mode 100755 index 00000000..424d98b9 --- /dev/null +++ b/performance_profiling/profile_performance_for_update.bash @@ -0,0 +1,312 @@ +#!/bin/bash +set -e + +############################################################################### +# Manually edit parameters here: + +work_dir=/lcrc/group/e3sm/ac.forsyth2/zstash_performance/ +unique_id=profile_update_20251218 + +dir_to_copy_from=/lcrc/group/e3sm/ac.forsyth2/E3SMv2/v2.LR.historical_0201/ +subdir_for_create=build/ # Use "none" to skip create +subdir_for_update=init/ # Use "none" to skip update + +fresh_globus=true +# ENDPOINT UUIDS: +# LCRC_IMPROV_DTN_ENDPOINT=15288284-7006-4041-ba1a-6b52501e49f1 +# NERSC_PERLMUTTER_ENDPOINT=6bdc7956-fc0f-4ad2-989c-7aa5ee643a79 +# NERSC_HPSS_ENDPOINT=9cd89cfd-6d04-11e5-ba46-22000b92c6ec +# PIC_COMPY_DTN_ENDPOINT=68fbd2fa-83d7-11e9-8e63-029d279f7e24 +# GLOBUS_TUTORIAL_COLLECTION_1_ENDPOINT=6c54cade-bde5-45c1-bdea-f4bd71dba2cc +dst_endpoint_uuid=9cd89cfd-6d04-11e5-ba46-22000b92c6ec +dst_archive_dir=/home/f/forsyth/zstash_performance/ + +############################################################################### +# Colored messages + +# Colors for output +RED='\033[0;31m' +GREEN='\033[0;32m' +YELLOW='\033[1;33m' +BLUE='\033[0;34m' +CYAN='\033[0;36m' +NC='\033[0m' # No Color + +# Functions to print colored messages +print_step() { + echo -e "${CYAN}[STEP]${NC} $1" +} + +print_info() { + echo -e "${BLUE}[INFO]${NC} $1" +} + +print_success() { + echo -e "${GREEN}[SUCCESS]${NC} $1" +} + +print_error() { + echo -e "${RED}[ERROR]${NC} $1" +} + +print_warning() { + echo -e "${YELLOW}[WARNING]${NC} $1" +} + +############################################################################### +# Utility functions + +confirm() { + read -p "$1 (y/n): " -n 1 -r + echo + [[ $REPLY =~ ^[Yy]$ ]] +} + +############################################################################### +# Validation + +print_step "Validating configuration..." + +if [ ! -d "$dir_to_copy_from" ]; then + print_error "Source directory does not exist: $dir_to_copy_from" + exit 1 +fi + +if [ "$subdir_for_create" != "none" ] && [ -n "$subdir_for_create" ]; then + if [ ! -d "${dir_to_copy_from}${subdir_for_create}" ]; then + print_error "Create subdirectory does not exist: ${dir_to_copy_from}${subdir_for_create}" + exit 1 + fi +fi + +if [ "$subdir_for_update" != "none" ] && [ -n "$subdir_for_update" ]; then + if [ ! -d "${dir_to_copy_from}${subdir_for_update}" ]; then + print_error "Update subdirectory does not exist: ${dir_to_copy_from}${subdir_for_update}" + exit 1 + fi +fi + +print_success "Configuration validated" + +############################################################################### +# Main script: + +if [ "${fresh_globus}" == "true" ]; then + print_step "Setting up fresh Globus authentication..." + + # 1. Activate endpoints + echo "Go to https://app.globus.org/file-manager?two_pane=true > For 'Collection', choose the endpoints you're using, and authenticate if needed:" + echo "LCRC Improv DTN, NERSC Perlmutter, NERSC HPSS, pic#compy-dtn" + if ! confirm "Have you authenticated into the correct endpoints?"; then + exit 1 + fi + + # 2. Reset authentication token files + INI_PATH=${HOME}/.zstash.ini + TOKEN_FILE=${HOME}/.zstash_globus_tokens.json + + if [ -f "${INI_PATH}" ]; then + rm -f "${INI_PATH}" + print_info "Removed ${INI_PATH}" + fi + + if [ -f "${TOKEN_FILE}" ]; then + rm -f "${TOKEN_FILE}" + print_info "Removed ${TOKEN_FILE}" + fi + + # 3. Reset Globus consents + echo "https://auth.globus.org/v2/web/consents > Globus Endpoint Performance Monitoring > rescind all" + if ! confirm "Have you revoked Globus consents?"; then + exit 1 + fi + + print_success "Globus authentication reset complete" +fi + +print_step "Creating work directories..." + +work_subdir=${work_dir}${unique_id}/ +mkdir -p "${work_subdir}" + +archive_dir=${work_subdir}archive_dir/ +cache_dir=${work_subdir}cache/ +log_dir=${work_subdir}logs/ +mkdir -p "${archive_dir}" +mkdir -p "${cache_dir}" +mkdir -p "${log_dir}" + +# Define log file paths early so they're always available +create_log=${log_dir}create.log +update_log=${log_dir}update.log + +cd "${work_subdir}" +dst_archive_subdir=${dst_archive_dir}${unique_id}/ + +print_success "Work directories created at ${work_subdir}" + +############################################################################### +# CREATE operation + +if [ -n "$subdir_for_create" ] && [ "$subdir_for_create" != "none" ]; then + print_step "Starting CREATE operation..." + + print_info "Copying data from ${dir_to_copy_from}${subdir_for_create}" + cp -r "${dir_to_copy_from}${subdir_for_create}" "${archive_dir}${subdir_for_create}" + + echo "" + print_info "Archive directory size:" + time du -sh "${archive_dir}" + echo "" + + print_info "Running zstash create..." + print_info "Command: zstash create --hpss=globus://${dst_endpoint_uuid}/${dst_archive_subdir} --cache=${cache_dir} -v ${archive_dir}" + + if { time zstash create --hpss="globus://${dst_endpoint_uuid}/${dst_archive_subdir}" --cache="${cache_dir}" -v "${archive_dir}" ; } 2>&1 | tee "${create_log}"; then + print_success "zstash create completed successfully" + else + print_error "zstash create failed with exit code $?" + exit 1 + fi + + echo "" + print_warning "Verification required:" + echo "Go to your destination machine and run:" + echo " ls ${dst_archive_subdir}" + echo "Expected output: 000000.tar index.db" + echo "" + + if ! confirm "Is the ls result correct?"; then + print_error "Verification failed" + exit 1 + fi + + print_success "CREATE operation verified" +else + print_info "Skipping CREATE operation (subdir_for_create is 'none' or empty)" +fi + +############################################################################### +# UPDATE operation + +if [ -n "$subdir_for_update" ] && [ "$subdir_for_update" != "none" ]; then + print_step "Starting UPDATE operation..." + + print_info "Copying additional data from ${dir_to_copy_from}${subdir_for_update}" + cp -r "${dir_to_copy_from}${subdir_for_update}" "${archive_dir}${subdir_for_update}" + + echo "" + print_info "Size of newly added data:" + time du -sh "${dir_to_copy_from}${subdir_for_update}" + echo "" + + # For update, we need to be in archive_dir + cd "${archive_dir}" + + print_info "Running zstash update..." + print_info "Command: zstash update --hpss=globus://${dst_endpoint_uuid}/${dst_archive_subdir} --cache=${cache_dir} -v" + + if { time zstash update --hpss="globus://${dst_endpoint_uuid}/${dst_archive_subdir}" --cache="${cache_dir}" -v ; } 2>&1 | tee "${update_log}"; then + print_success "zstash update completed successfully" + else + print_error "zstash update failed with exit code $?" + exit 1 + fi + + echo "" + print_warning "Verification required:" + echo "Go to your destination machine and run:" + echo " ls ${dst_archive_subdir}" + echo "Expected output: 000000.tar 000001.tar index.db" + echo "(May show more tars if you have run update multiple times)" + echo "" + if ! confirm "Is the ls result correct?"; then + print_error "Verification failed" + exit 1 + fi + + print_success "UPDATE operation verified" +else + print_info "Skipping UPDATE operation (subdir_for_update is 'none' or empty)" +fi + +############################################################################### +# Performance Analysis + +cd "${work_subdir}" + +if [ -f "${update_log}" ] && [ -s "${update_log}" ]; then + print_step "Analyzing performance metrics from update log..." + + echo "" + print_info "==========================================" + print_info "Performance Analysis (OPTIMIZED VERSION)" + print_info "==========================================" + + # File gathering metrics (OPTIMIZED - now with stats) + if grep -q "PERFORMANCE (get_files_to_archive_with_stats)" "$update_log" 2>/dev/null; then + echo "" + print_info "File Gathering Breakdown (with stats collection):" + grep "PERFORMANCE (scandir):" "$update_log" | tail -5 + grep "PERFORMANCE (sort):" "$update_log" + grep "PERFORMANCE (get_files_to_archive_with_stats): TOTAL TIME:" "$update_log" + fi + + # Database comparison metrics (OPTIMIZED - no stats) + if grep -q "PERFORMANCE: Database comparison completed" "$update_log" 2>/dev/null; then + echo "" + print_info "Database Comparison Breakdown (OPTIMIZED):" + grep "PERFORMANCE: Total comparison time:" "$update_log" + grep "PERFORMANCE: Files checked:" "$update_log" + grep "PERFORMANCE: New files to archive:" "$update_log" + grep "PERFORMANCE: Average rate:" "$update_log" + grep "database load:" "$update_log" + grep "comparison (in-memory):" "$update_log" + fi + + # Optimization impact + if grep -q "PERFORMANCE: Optimization impact:" "$update_log" 2>/dev/null; then + echo "" + print_info "Optimization Impact:" + grep "stat operations eliminated:" "$update_log" + grep "All stats performed during initial filesystem walk" "$update_log" + fi + + # Overall summary + if grep -q "PERFORMANCE: Update complete - Summary:" "$update_log" 2>/dev/null; then + echo "" + print_info "Overall Time Breakdown:" + grep "PERFORMANCE: Update complete - Summary:" -A 6 "$update_log" | tail -6 + fi + + echo "" + print_info "==========================================" + print_info "Analysis Commands" + print_info "==========================================" + print_success "View full logs:" + echo " cat $log_dir/create.log # Create profiling" + echo " cat $log_dir/update.log # Update profiling" + echo "" + print_success "Extract specific metrics:" + echo " grep 'PERFORMANCE' $update_log | less" + echo " grep 'PERFORMANCE (scandir)' $update_log" + echo " grep 'database load' $update_log" + echo " grep 'Optimization impact' $update_log" + echo "" + print_success "Compare times:" + echo " grep 'TOTAL TIME' $update_log" + echo "" + print_success "View optimization benefits:" + echo " grep 'stat operations eliminated' $update_log" + echo "" +else + print_warning "No update log found or log is empty - skipping performance analysis" +fi + +echo "" +print_info "==========================================" +print_success "Profiling Complete!" +print_info "==========================================" +print_info "Working directory: ${work_subdir}" +print_info "Logs available at: ${log_dir}" +echo "" diff --git a/tests/unit/test_update.py b/tests/unit/test_update.py new file mode 100644 index 00000000..d80c2e88 --- /dev/null +++ b/tests/unit/test_update.py @@ -0,0 +1,189 @@ +import time +from datetime import datetime +from typing import Dict, Tuple + +from zstash.update import UpdatePerformanceLogger + + +class TestUpdatePerformanceLogger: + """Tests for UpdatePerformanceLogger class""" + + def test_initialization(self): + """Test that logger initializes with zero values""" + perf = UpdatePerformanceLogger() + assert perf.overall_start == 0 + assert perf.db_elapsed == 0 + assert perf.gather_elapsed == 0 + assert perf.check_elapsed == 0 + + def test_start_overall(self): + """Test overall timing start""" + perf = UpdatePerformanceLogger() + perf.start_overall() + assert perf.overall_start > 0 + assert perf.overall_start <= time.time() + + def test_database_open_timing(self): + """Test database open timing""" + perf = UpdatePerformanceLogger() + perf.start_database_open() + time.sleep(0.01) # Small delay + perf.end_database_open() + assert perf.db_elapsed >= 0.01 + assert perf.db_elapsed < 0.1 # Should be quick + + def test_file_gathering_timing(self): + """Test file gathering timing""" + perf = UpdatePerformanceLogger() + perf.start_file_gathering() + time.sleep(0.01) + perf.end_file_gathering(100) + assert perf.gather_elapsed >= 0.01 + + def test_database_check_timing(self): + """Test database check timing""" + perf = UpdatePerformanceLogger() + perf.start_database_check() + perf.start_database_load() + time.sleep(0.01) + perf.end_database_load(50) + + perf.start_comparison() + time.sleep(0.01) + perf.end_database_check(100, 10) + + assert perf.db_load_elapsed >= 0.01 + assert perf.comparison_elapsed >= 0.01 + assert perf.check_elapsed >= 0.02 + + def test_comparison_progress_logging(self): + """Test that progress logging doesn't raise errors""" + perf = UpdatePerformanceLogger() + perf.start_database_check() + perf.start_comparison() + + # Should log at interval + perf.log_comparison_progress(1000, 5000, interval=1000) + perf.log_comparison_progress(2000, 5000, interval=1000) + + # Should not log between intervals + perf.log_comparison_progress(1500, 5000, interval=1000) + + def test_overall_summary_logging(self): + """Test overall summary logging completes without error""" + perf = UpdatePerformanceLogger() + perf.start_overall() + + # Simulate operation timings + perf.start_database_open() + perf.end_database_open() + + perf.start_file_gathering() + perf.end_file_gathering(100) + + perf.start_database_check() + perf.start_database_load() + perf.end_database_load(50) + perf.start_comparison() + perf.end_database_check(100, 10) + + perf.start_tar_preparation() + perf.end_tar_preparation() + + perf.start_add_files() + perf.end_add_files() + + # Should complete without error + perf.log_overall_summary() + + def test_early_exit_logging(self): + """Test early exit logging""" + perf = UpdatePerformanceLogger() + perf.start_overall() + time.sleep(0.01) + perf.log_early_exit("no updates needed") + + +class TestUpdateDatabaseOptimization: + """Tests for the optimized update_database function""" + + def test_in_memory_comparison(self): + """Test that database comparison uses in-memory lookup""" + # Mock data structures + archived_files: Dict[str, Tuple[int, datetime]] = { + "file1.txt": (100, datetime(2024, 1, 1, 12, 0, 0)), + "file2.txt": (200, datetime(2024, 1, 2, 12, 0, 0)), + } + + file_stats: Dict[str, Tuple[int, datetime]] = { + "file1.txt": (100, datetime(2024, 1, 1, 12, 0, 0)), # Unchanged + "file2.txt": (250, datetime(2024, 1, 2, 12, 0, 0)), # Changed size + "file3.txt": (150, datetime(2024, 1, 3, 12, 0, 0)), # New file + } + + newfiles = [] + TIME_TOL = 3600 # 1 hour tolerance + + for file_path, (size_new, mdtime_new) in file_stats.items(): + if file_path not in archived_files: + # New file + newfiles.append(file_path) + else: + # Check if changed + archived_size, archived_mtime = archived_files[file_path] + if not ( + (size_new == archived_size) + and (abs((mdtime_new - archived_mtime).total_seconds()) <= TIME_TOL) + ): + newfiles.append(file_path) + + # Should detect file2.txt (changed) and file3.txt (new) + assert "file2.txt" in newfiles + assert "file3.txt" in newfiles + assert "file1.txt" not in newfiles + assert len(newfiles) == 2 + + def test_mtime_tolerance(self): + """Test that mtime tolerance is respected""" + base_time = datetime(2024, 1, 1, 12, 0, 0) + TIME_TOL = 3600 # 1 hour + + archived_files: Dict[str, Tuple[int, datetime]] = { + "file1.txt": (100, base_time), + } + + # File with mtime within tolerance + file_stats_within: Dict[str, Tuple[int, datetime]] = { + "file1.txt": (100, datetime(2024, 1, 1, 12, 30, 0)), # 30 min difference + } + + # File with mtime outside tolerance + file_stats_outside: Dict[str, Tuple[int, datetime]] = { + "file1.txt": (100, datetime(2024, 1, 1, 14, 0, 1)), # >1 hour difference + } + + # Test within tolerance + newfiles = [] + for file_path, (size_new, mdtime_new) in file_stats_within.items(): + if file_path in archived_files: + archived_size, archived_mtime = archived_files[file_path] + if not ( + (size_new == archived_size) + and (abs((mdtime_new - archived_mtime).total_seconds()) <= TIME_TOL) + ): + newfiles.append(file_path) + + assert len(newfiles) == 0 # Within tolerance, no update needed + + # Test outside tolerance + newfiles = [] + for file_path, (size_new, mdtime_new) in file_stats_outside.items(): + if file_path in archived_files: + archived_size, archived_mtime = archived_files[file_path] + if not ( + (size_new == archived_size) + and (abs((mdtime_new - archived_mtime).total_seconds()) <= TIME_TOL) + ): + newfiles.append(file_path) + + assert len(newfiles) == 1 # Outside tolerance, needs update diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py new file mode 100644 index 00000000..9386cc02 --- /dev/null +++ b/tests/unit/test_utils.py @@ -0,0 +1,251 @@ +import os +import tempfile +import time +from datetime import datetime +from pathlib import Path +from unittest.mock import patch + +import pytest + +from zstash.utils import ( + DirectoryScanner, + FileGatheringPerformanceLogger, + get_files_to_archive_with_stats, +) + + +class TestFileGatheringPerformanceLogger: + """Tests for FileGatheringPerformanceLogger class""" + + @patch("zstash.utils.logger") + def test_log_scandir_progress(self, mock_logger): + """Test scandir progress logging""" + perf = FileGatheringPerformanceLogger() + + perf.log_scandir_progress(1000, 5000, 10.5) + assert mock_logger.debug.called + + @patch("zstash.utils.logger") + def test_log_scandir_complete(self, mock_logger): + """Test scandir completion logging""" + perf = FileGatheringPerformanceLogger() + + perf.log_scandir_complete(100, 500, 10, 5.5) + assert mock_logger.debug.call_count >= 5 + + @patch("zstash.utils.logger") + def test_log_filter(self, mock_logger): + """Test filter logging""" + perf = FileGatheringPerformanceLogger() + + perf.log_filter("include", "*.py", 0.5, 100, 50) + assert mock_logger.debug.call_count >= 2 + + +class TestDirectoryScanner: + """Tests for DirectoryScanner class""" + + @pytest.fixture + def temp_dir(self): + """Create a temporary directory structure for testing""" + with tempfile.TemporaryDirectory() as tmpdir: + # Create test structure + base = Path(tmpdir) + + # Regular files + (base / "file1.txt").write_text("content1") + (base / "file2.txt").write_text("content2") + + # Subdirectory with files + subdir = base / "subdir" + subdir.mkdir() + (subdir / "file3.txt").write_text("content3") + + # Empty directory + empty = base / "empty" + empty.mkdir() + + # Cache directory (should be excluded) + cache = base / "cache" + cache.mkdir() + (cache / "cached.txt").write_text("cached") + + # Symlink + symlink = base / "link.txt" + symlink.symlink_to(base / "file1.txt") + + yield tmpdir + + def test_scan_basic_structure(self, temp_dir): + """Test scanning basic directory structure""" + perf_logger = FileGatheringPerformanceLogger() + + cache_path = os.path.join(temp_dir, "cache") + scanner = DirectoryScanner(cache_path, perf_logger, time.time()) + + # Change to temp dir and scan + original_dir = os.getcwd() + try: + os.chdir(temp_dir) + scanner.scan_directory(".") + + # Should find files but not cache contents + assert scanner.file_count >= 3 # file1, file2, file3, link + assert scanner.dir_count >= 2 # root, subdir, empty + assert scanner.empty_dir_count >= 1 # empty dir + + # Cache file should not be in results + cache_file = os.path.normpath(os.path.join(cache_path, "cached.txt")) + assert cache_file not in scanner.file_stats + + # Regular files should be in results + file1 = os.path.normpath(os.path.join(".", "file1.txt")) + assert file1 in scanner.file_stats or "./file1.txt" in scanner.file_stats + + finally: + os.chdir(original_dir) + + +class TestGetFilesToArchiveWithStats: + """Tests for get_files_to_archive_with_stats function""" + + @pytest.fixture + def temp_archive_dir(self): + """Create a temporary directory with files to archive""" + with tempfile.TemporaryDirectory() as tmpdir: + base = Path(tmpdir) + + # Create various files + (base / "data.txt").write_text("data") + (base / "script.py").write_text("print('hello')") + (base / "config.json").write_text('{"key": "value"}') + + # Create subdirectory + subdir = base / "logs" + subdir.mkdir() + (subdir / "log1.txt").write_text("log entry 1") + (subdir / "log2.txt").write_text("log entry 2") + + # Create cache directory + cache = base / "zstash_cache" + cache.mkdir() + (cache / "index.db").write_text("database") + + yield tmpdir + + def test_basic_file_gathering(self, temp_archive_dir): + """Test basic file gathering with stats""" + original_dir = os.getcwd() + try: + os.chdir(temp_archive_dir) + + result = get_files_to_archive_with_stats("zstash_cache", None, None) + + # Should return dict mapping paths to (size, mtime) tuples + assert isinstance(result, dict) + assert len(result) > 0 + + # Check structure of results + for path, stats in result.items(): + assert isinstance(path, str) + assert isinstance(stats, tuple) + assert len(stats) == 2 + size, mtime = stats + assert isinstance(size, int) + assert isinstance(mtime, datetime) + + # Cache directory files should be excluded + for path in result.keys(): + assert "zstash_cache" not in path + + finally: + os.chdir(original_dir) + + def test_include_pattern(self, temp_archive_dir): + """Test include pattern filtering""" + original_dir = os.getcwd() + try: + os.chdir(temp_archive_dir) + + # Only include .txt files + result = get_files_to_archive_with_stats("zstash_cache", "*.txt", None) + + # All results should be .txt files + for path in result.keys(): + if path and not path.endswith("/"): # Not empty dir + assert path.endswith(".txt") + + finally: + os.chdir(original_dir) + + def test_exclude_pattern(self, temp_archive_dir): + """Test exclude pattern filtering""" + original_dir = os.getcwd() + try: + os.chdir(temp_archive_dir) + + # Exclude .py files + result = get_files_to_archive_with_stats("zstash_cache", None, "*.py") + + # No .py files should be in results + for path in result.keys(): + assert not path.endswith(".py") + + finally: + os.chdir(original_dir) + + def test_include_and_exclude(self, temp_archive_dir): + """Test both include and exclude patterns""" + original_dir = os.getcwd() + try: + os.chdir(temp_archive_dir) + + # Include all .txt, but exclude logs + result = get_files_to_archive_with_stats("zstash_cache", "*.txt", "logs/*") + + # Should have .txt files but not from logs directory + for path in result.keys(): + if path and not path.endswith("/"): + assert path.endswith(".txt") + assert "logs" not in path + + finally: + os.chdir(original_dir) + + def test_returns_ordered_dict(self, temp_archive_dir): + """Test that results maintain sorted order""" + original_dir = os.getcwd() + try: + os.chdir(temp_archive_dir) + + result = get_files_to_archive_with_stats("zstash_cache", None, None) + + # Keys should be in sorted order + keys = list(result.keys()) + sorted_keys = sorted(keys) + assert keys == sorted_keys + + finally: + os.chdir(original_dir) + + def test_empty_directory_handling(self, temp_archive_dir): + """Test handling of empty directories""" + original_dir = os.getcwd() + try: + os.chdir(temp_archive_dir) + + # Create an empty directory + empty_dir = Path(temp_archive_dir) / "empty_folder" + empty_dir.mkdir() + + result = get_files_to_archive_with_stats("zstash_cache", None, None) + + # Empty directory should be in results with size 0 + empty_path = os.path.normpath("./empty_folder") + if empty_path in result: + size, mtime = result[empty_path] + assert size == 0 + assert isinstance(mtime, datetime) + + finally: + os.chdir(original_dir) diff --git a/zstash/update.py b/zstash/update.py index b0f2af40..56ddeb61 100644 --- a/zstash/update.py +++ b/zstash/update.py @@ -4,26 +4,204 @@ import logging import os.path import sqlite3 -import stat import sys +import time from datetime import datetime -from typing import List, Optional, Tuple +from typing import Dict, List, Optional, Tuple from .globus import globus_activate, globus_finalize from .hpss import hpss_get, hpss_put from .hpss_utils import add_files -from .settings import ( - DEFAULT_CACHE, - TIME_TOL, - FilesRow, - TupleFilesRow, - config, - get_db_filename, - logger, -) -from .utils import get_files_to_archive, update_config +from .settings import DEFAULT_CACHE, TIME_TOL, config, get_db_filename, logger +from .utils import get_files_to_archive_with_stats, update_config + + +# Classes ##################################################################### +class UpdatePerformanceLogger: + """ + Performance logger for tracking and reporting timing metrics + during the update_database operation. + """ + + def __init__(self): + self.overall_start: float = 0 + self.db_start: float = 0 + self.db_elapsed: float = 0 + self.gather_start: float = 0 + self.gather_elapsed: float = 0 + self.check_start: float = 0 + self.check_elapsed: float = 0 + self.db_load_start: float = 0 + self.db_load_elapsed: float = 0 + self.comparison_start: float = 0 + self.comparison_elapsed: float = 0 + self.tar_prep_start: float = 0 + self.tar_prep_elapsed: float = 0 + self.add_files_start: float = 0 + self.add_files_elapsed: float = 0 + + def start_overall(self): + """Start timing the overall operation.""" + self.overall_start = time.time() + logger.debug("=" * 80) + logger.debug("PERFORMANCE PROFILING: Starting update_database") + logger.debug("=" * 80) + + def start_database_open(self): + """Start timing database opening.""" + self.db_start = time.time() + logger.debug("Opening index database") + + def end_database_open(self): + """End timing database opening and config update.""" + self.db_elapsed = time.time() - self.db_start + logger.debug( + f"PERFORMANCE: Database open and config update: {self.db_elapsed:.2f} seconds" + ) + + def start_file_gathering(self): + """Start timing file gathering operation.""" + self.gather_start = time.time() + logger.debug("PERFORMANCE: Starting file gathering with stats (OPTIMIZED)...") + + def end_file_gathering(self, file_count: int): + """End timing file gathering operation.""" + self.gather_elapsed = time.time() - self.gather_start + logger.debug( + f"PERFORMANCE: File gathering completed: {self.gather_elapsed:.2f} seconds" + ) + logger.debug(f"PERFORMANCE: Total files found: {file_count}") + + def start_database_check(self): + """Start timing database comparison.""" + self.check_start = time.time() + logger.debug( + "PERFORMANCE: Starting database comparison (OPTIMIZED - NO STATS)..." + ) + + def start_database_load(self): + """Start timing database loading into memory.""" + self.db_load_start = time.time() + logger.debug("PERFORMANCE: Loading database into memory...") + def end_database_load(self, archived_count: int): + """End timing database loading.""" + self.db_load_elapsed = time.time() - self.db_load_start + logger.debug( + f"PERFORMANCE: Database loaded: {self.db_load_elapsed:.2f} seconds" + ) + logger.debug(f"PERFORMANCE: Archived files in database: {archived_count}") + + def start_comparison(self): + """Start timing the comparison operation.""" + self.comparison_start = time.time() + + def log_comparison_progress( + self, files_checked: int, total_files: int, interval: int = 1000 + ): + """Log comparison progress at regular intervals.""" + if files_checked % interval == 0: + elapsed_so_far = time.time() - self.comparison_start + rate = files_checked / elapsed_so_far if elapsed_so_far > 0 else 0 + logger.debug( + f"PERFORMANCE: Compared {files_checked}/{total_files} files " + f"({rate:.1f} files/sec, {elapsed_so_far:.1f}s elapsed)" + ) + + def end_database_check(self, files_checked: int, new_files_count: int): + """End timing database comparison and log detailed metrics.""" + self.comparison_elapsed = time.time() - self.comparison_start + self.check_elapsed = time.time() - self.check_start + + logger.debug("=" * 80) + logger.debug("PERFORMANCE: Database comparison completed (OPTIMIZED)") + logger.debug( + f"PERFORMANCE: Total comparison time: {self.check_elapsed:.2f} seconds" + ) + logger.debug(f"PERFORMANCE: Files checked: {files_checked}") + logger.debug(f"PERFORMANCE: New files to archive: {new_files_count}") + logger.debug( + f"PERFORMANCE: Average rate: {files_checked / self.check_elapsed:.1f} files/sec" + ) + logger.debug("-" * 80) + logger.debug("PERFORMANCE: Time breakdown:") + logger.debug( + f" - database load: {self.db_load_elapsed:.2f}s " + f"({self.db_load_elapsed / self.check_elapsed * 100:.1f}%)" + ) + logger.debug( + f" - comparison (in-memory): {self.comparison_elapsed:.2f}s " + f"({self.comparison_elapsed / self.check_elapsed * 100:.1f}%)" + ) + logger.debug("-" * 80) + logger.debug("PERFORMANCE: Optimization impact:") + logger.debug(f" - stat operations eliminated: {files_checked} (100%)") + logger.debug(" - All stats performed during initial filesystem walk") + logger.debug("=" * 80) + + def start_tar_preparation(self): + """Start timing tar archive preparation.""" + self.tar_prep_start = time.time() + logger.debug("PERFORMANCE: Finding last used tar archive...") + + def end_tar_preparation(self): + """End timing tar archive preparation.""" + self.tar_prep_elapsed = time.time() - self.tar_prep_start + logger.debug( + f"PERFORMANCE: Tar archive preparation: {self.tar_prep_elapsed:.2f} seconds" + ) + + def start_add_files(self): + """Start timing add_files operation.""" + self.add_files_start = time.time() + logger.debug("PERFORMANCE: Starting add_files operation...") + + def end_add_files(self): + """End timing add_files operation.""" + self.add_files_elapsed = time.time() - self.add_files_start + logger.debug( + f"PERFORMANCE: add_files operation completed: {self.add_files_elapsed:.2f} seconds" + ) + + def log_overall_summary(self): + """Log the complete performance summary.""" + overall_elapsed = time.time() - self.overall_start + + logger.debug("=" * 80) + logger.debug("PERFORMANCE: Update complete - Summary:") + logger.debug( + f" - Database open/config: {self.db_elapsed:.2f}s " + f"({self.db_elapsed / overall_elapsed * 100:.1f}%)" + ) + logger.debug( + f" - File gathering: {self.gather_elapsed:.2f}s " + f"({self.gather_elapsed / overall_elapsed * 100:.1f}%)" + ) + logger.debug( + f" - Database comparison: {self.check_elapsed:.2f}s " + f"({self.check_elapsed / overall_elapsed * 100:.1f}%)" + ) + logger.debug( + f" - Tar preparation: {self.tar_prep_elapsed:.2f}s " + f"({self.tar_prep_elapsed / overall_elapsed * 100:.1f}%)" + ) + logger.debug( + f" - Add files: {self.add_files_elapsed:.2f}s " + f"({self.add_files_elapsed / overall_elapsed * 100:.1f}%)" + ) + logger.debug(f" - TOTAL TIME: {overall_elapsed:.2f} seconds") + logger.debug("=" * 80) + + def log_early_exit(self, reason: str = ""): + """Log performance for early exits (no updates, dry-run).""" + overall_elapsed = time.time() - self.overall_start + suffix = f" ({reason})" if reason else "" + logger.debug( + f"PERFORMANCE: Total execution time{suffix}: {overall_elapsed:.2f} seconds" + ) + +# Functions ##################################################################### def update(): args: argparse.Namespace @@ -149,8 +327,13 @@ def setup_update() -> Tuple[argparse.Namespace, str]: def update_database( # noqa: C901 args: argparse.Namespace, cache: str ) -> Optional[List[str]]: + + # Initialize performance logger + perf = UpdatePerformanceLogger() + perf.start_overall() + # Open database - logger.debug("Opening index database") + perf.start_database_open() if not os.path.exists(get_db_filename(cache)): # The database file doesn't exist in the cache. # We need to retrieve it from HPSS @@ -175,6 +358,7 @@ def update_database( # noqa: C901 cur: sqlite3.Cursor = con.cursor() update_config(cur) + perf.end_database_open() if config.maxsize is not None: maxsize = config.maxsize @@ -201,40 +385,71 @@ def update_database( # noqa: C901 logger.debug("Max size : {}".format(maxsize)) logger.debug("Keep local tar files : {}".format(keep)) - files: List[str] = get_files_to_archive(cache, args.include, args.exclude) + # Gather files to archive + perf.start_file_gathering() + file_stats: Dict[str, Tuple[int, datetime]] = get_files_to_archive_with_stats( + cache, args.include, args.exclude + ) + files: List[str] = list(file_stats.keys()) + perf.end_file_gathering(len(files)) + + # Database checking - OPTIMIZED VERSION + perf.start_database_check() + + # Load all archived files into memory once + perf.start_database_load() + + # Dictionary mapping file path -> (size, mtime) for O(1) lookup + archived_files: Dict[str, Tuple[int, datetime]] = {} - # Eliminate files that are already archived and up to date + cur.execute("SELECT name, size, mtime FROM files") + db_rows = cur.fetchall() + + for row in db_rows: + file_path: str = row[0] + size: int = row[1] + mtime: datetime = row[2] + + # If file appears multiple times, keep the one with latest mtime + if file_path in archived_files: + existing_mtime = archived_files[file_path][1] + if mtime > existing_mtime: + archived_files[file_path] = (size, mtime) + else: + archived_files[file_path] = (size, mtime) + + perf.end_database_load(len(archived_files)) + + # Compare using pre-collected stats - NO os.lstat() calls! + perf.start_comparison() newfiles: List[str] = [] + files_checked = 0 + for file_path in files: - statinfo: os.stat_result = os.lstat(file_path) - mdtime_new: datetime = datetime.utcfromtimestamp(statinfo.st_mtime) - mode: int = statinfo.st_mode - # For symbolic links or directories, size should be 0 - size_new: int - if stat.S_ISLNK(mode) or stat.S_ISDIR(mode): - size_new = 0 + # Get the stat info we already collected during filesystem walk + size_new, mdtime_new = file_stats[file_path] + + # Check if file exists in database + if file_path not in archived_files: + # File not in database - it's new + newfiles.append(file_path) else: - size_new = statinfo.st_size - - # Select the file matching the path. - cur.execute("select * from files where name = ?", (file_path,)) - new: bool = True - while True: - # Get the corresponding row in the 'files' table - match_: Optional[TupleFilesRow] = cur.fetchone() - if match_ is None: - break - else: - match: FilesRow = FilesRow(match_) + # File exists in database - check if it changed + archived_size, archived_mtime = archived_files[file_path] - if (size_new == match.size) and ( - abs((mdtime_new - match.mtime).total_seconds()) <= TIME_TOL + if not ( + (size_new == archived_size) + and (abs((mdtime_new - archived_mtime).total_seconds()) <= TIME_TOL) ): - # File exists with same size and modification time within tolerance - new = False - break - if new: - newfiles.append(file_path) + # File has changed + newfiles.append(file_path) + + files_checked += 1 + + # Progress logging every 1000 files + perf.log_comparison_progress(files_checked, len(files)) + + perf.end_database_check(files_checked, len(newfiles)) # Anything to do? if len(newfiles) == 0: @@ -242,6 +457,8 @@ def update_database( # noqa: C901 # Close database con.commit() con.close() + + perf.log_early_exit() return None # --dry-run option @@ -252,9 +469,12 @@ def update_database( # noqa: C901 # Close database con.commit() con.close() + + perf.log_early_exit("dry-run") return None # Find last used tar archive + perf.start_tar_preparation() itar: int = -1 cur.execute("select distinct tar from files") tfiles: List[Tuple[str]] = cur.fetchall() @@ -262,6 +482,11 @@ def update_database( # noqa: C901 tfile_string: str = tfile[0] itar = max(itar, int(tfile_string[0:6], 16)) + perf.end_tar_preparation() + + # Add files + perf.start_add_files() + failures: List[str] if args.follow_symlinks: try: @@ -295,8 +520,12 @@ def update_database( # noqa: C901 overwrite_duplicate_tars=args.overwrite_duplicate_tars, ) + perf.end_add_files() + # Close database con.commit() con.close() + perf.log_overall_summary() + return failures diff --git a/zstash/utils.py b/zstash/utils.py index ea793603..332c92a4 100644 --- a/zstash/utils.py +++ b/zstash/utils.py @@ -3,14 +3,203 @@ import os import shlex import sqlite3 +import stat as stat_module import subprocess +import time +from collections import OrderedDict from datetime import datetime, timezone from fnmatch import fnmatch -from typing import Any, List, Tuple +from typing import Any, Dict, List, Tuple from .settings import TupleTarsRow, config, logger +# Classes ##################################################################### +class FileGatheringPerformanceLogger: + """Helper class to handle performance logging.""" + + def log_separator(self): + """Log a separator line.""" + logger.debug("-" * 80) + + def log_start(self, operation: str): + """Log the start of an operation.""" + logger.debug(f"PERFORMANCE ({operation}): Starting file discovery with stats") + + def log_scandir_progress(self, dir_count: int, file_count: int, elapsed: float): + """Log progress during directory scanning.""" + rate = dir_count / elapsed if elapsed > 0 else 0 + logger.debug( + f"PERFORMANCE (scandir): Scanned {dir_count} directories, " + f"{file_count} files ({rate:.1f} dirs/sec, {elapsed:.1f}s elapsed)" + ) + + def log_scandir_complete( + self, dir_count: int, file_count: int, empty_dir_count: int, elapsed: float + ): + """Log completion of directory scanning.""" + logger.debug("PERFORMANCE (scandir): Completed filesystem walk with stats") + logger.debug(f" - Directories scanned: {dir_count}") + logger.debug(f" - Files found: {file_count}") + logger.debug(f" - Empty directories: {empty_dir_count}") + logger.debug(f" - Time: {elapsed:.2f} seconds") + logger.debug( + f" - Rate: {dir_count / elapsed:.1f} dirs/sec, {file_count / elapsed:.1f} files/sec" + ) + + def log_filter( + self, + filter_type: str, + pattern: str, + elapsed: float, + files_after: int, + files_filtered: int, + ): + """Log filtering operation results.""" + logger.debug( + f"PERFORMANCE ({filter_type} filter): Applied {filter_type} pattern '{pattern}': {elapsed:.2f} seconds" + ) + logger.debug( + f" - Files after {filter_type}: {files_after} (filtered out {files_filtered})" + ) + + def log_sort(self, num_entries: int, elapsed: float): + """Log sorting operation results.""" + logger.debug( + f"PERFORMANCE (sort): Sorted {num_entries} entries: {elapsed:.2f} seconds" + ) + + def log_total(self, operation: str, total_elapsed: float, final_count: int): + """Log total operation time.""" + logger.debug( + f"PERFORMANCE ({operation}): TOTAL TIME: {total_elapsed:.2f} seconds" + ) + logger.debug(f"PERFORMANCE ({operation}): Final file count: {final_count}") + + def log_breakdown( + self, + operation: str, + total_elapsed: float, + walk_elapsed: float, + include_elapsed: float, + exclude_elapsed: float, + sort_elapsed: float, + include_enabled: bool, + exclude_enabled: bool, + ): + """Log time breakdown percentages.""" + if total_elapsed > 0: + logger.debug(f"PERFORMANCE ({operation}): Time breakdown:") + logger.debug( + f" - Filesystem walk with stats: {walk_elapsed:.2f}s " + f"({walk_elapsed / total_elapsed * 100:.1f}%)" + ) + if include_enabled: + logger.debug( + f" - Include filtering: {include_elapsed:.2f}s " + f"({include_elapsed / total_elapsed * 100:.1f}%)" + ) + if exclude_enabled: + logger.debug( + f" - Exclude filtering: {exclude_elapsed:.2f}s " + f"({exclude_elapsed / total_elapsed * 100:.1f}%)" + ) + logger.debug( + f" - Sorting: {sort_elapsed:.2f}s " + f"({sort_elapsed / total_elapsed * 100:.1f}%)" + ) + + +class DirectoryScanner: + """Helper class to scan directories and collect file stats.""" + + def __init__( + self, + cache_path: str, + perf_logger: FileGatheringPerformanceLogger, + walk_start_time: float, + ): + self.cache_path = cache_path + self.perf_logger = perf_logger + self.walk_start_time = walk_start_time + self.file_stats: Dict[str, Tuple[int, datetime]] = {} + self.dir_count = 0 + self.file_count = 0 + self.empty_dir_count = 0 + + def scan_directory(self, path: str): + """Recursively scan directory using os.scandir() for efficiency.""" + try: + entries = list(os.scandir(path)) + except PermissionError: + logger.warning(f"Permission denied: {path}") + return + + self.dir_count += 1 + has_contents = False + + for entry in entries: + # Skip the cache directory entirely + if entry.path == self.cache_path or entry.path.startswith( + self.cache_path + os.sep + ): + continue + + try: + # Get stat info - scandir provides this efficiently + # Use entry.stat(follow_symlinks=False) to match os.lstat() behavior + stat_info = entry.stat(follow_symlinks=False) + mode = stat_info.st_mode + + if entry.is_dir(follow_symlinks=False): + # Recursively scan subdirectory + self.scan_directory(entry.path) + has_contents = True + else: + # It's a file or symlink + has_contents = True + self.file_count += 1 + + # For symbolic links or directories, size should be 0 + if stat_module.S_ISLNK(mode): + size = 0 + else: + size = stat_info.st_size + + mtime = datetime.utcfromtimestamp(stat_info.st_mtime) + + # Normalize the path + normalized_path = os.path.normpath(entry.path) + self.file_stats[normalized_path] = (size, mtime) + + except (OSError, PermissionError) as e: + logger.warning(f"Error accessing {entry.path}: {e}") + continue + + # Handle empty directories + if not has_contents and path != ".": + self.empty_dir_count += 1 + normalized_path = os.path.normpath(path) + # Get actual mtime for empty directory + try: + stat_info = os.lstat(path) + mtime = datetime.utcfromtimestamp(stat_info.st_mtime) + self.file_stats[normalized_path] = (0, mtime) + except (OSError, PermissionError): + # Fallback if we can't stat the directory + self.file_stats[normalized_path] = (0, datetime.utcnow()) + + # Progress logging every 1000 directories + if self.dir_count % 1000 == 0: + elapsed = time.time() - self.walk_start_time + self.perf_logger.log_scandir_progress( + self.dir_count, self.file_count, elapsed + ) + + +# Functions ##################################################################### + + def ts_utc(): return datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S_%f") @@ -71,42 +260,117 @@ def run_command(command: str, error_str: str): raise RuntimeError(error_str) -def get_files_to_archive(cache: str, include: str, exclude: str) -> List[str]: - # List of files - logger.info("Gathering list of files to archive") - # Tuples of the form (path, filename) - file_tuples: List[Tuple[str, str]] = [] - # Walk the current directory - for root, dirnames, filenames in os.walk("."): - if not dirnames and not filenames: - # There are no subdirectories nor are there files. - # This directory is empty. - file_tuples.append((root, "")) - for filename in filenames: - # Loop over files - # filenames is a list, so if it is empty, no looping will occur. - file_tuples.append((root, filename)) - - # Sort first on directories (x[0]) - # Further sort on filenames (x[1]) - file_tuples = sorted(file_tuples, key=lambda x: (x[0], x[1])) - - # Relative file paths, excluding the cache - files: List[str] = [ - os.path.normpath(os.path.join(x[0], x[1])) - for x in file_tuples - if x[0] != os.path.join(".", cache) - ] - - # First, add files based on include pattern +def get_files_to_archive_with_stats( + cache: str, include: str, exclude: str +) -> Dict[str, Tuple[int, datetime]]: + """ + OPTIMIZED VERSION: Gather list of files to archive along with their stats. + + Uses os.scandir() to get file stats during the directory walk, + eliminating the need to stat files again later during database comparison. + + Returns: + Dictionary mapping file_path -> (size, mtime) + """ + # PERFORMANCE: Start timing file gathering + gather_total_start = time.time() + perf_logger = FileGatheringPerformanceLogger() + + perf_logger.log_separator() + perf_logger.log_start("get_files_to_archive_with_stats") + + # List of files with their stats + logger.debug("Gathering list of files to archive (with stats)") + + # PERFORMANCE: Time the os.scandir operation + walk_start = time.time() + cache_path = os.path.join(".", cache) + + scanner = DirectoryScanner(cache_path, perf_logger, walk_start) + scanner.scan_directory(".") + + walk_elapsed = time.time() - walk_start + perf_logger.log_scandir_complete( + scanner.dir_count, scanner.file_count, scanner.empty_dir_count, walk_elapsed + ) + + file_stats = scanner.file_stats + initial_file_count = len(file_stats) + + # Apply include/exclude filters + # PERFORMANCE: Time include filtering + include_elapsed = 0.0 if include is not None: - files = include_files(include, files) + include_start = time.time() + file_list = list(file_stats.keys()) + filtered_list = include_files(include, file_list) + # Keep only files that passed the filter + file_stats = {path: file_stats[path] for path in filtered_list} + include_elapsed = time.time() - include_start + perf_logger.log_filter( + "include", + include, + include_elapsed, + len(file_stats), + initial_file_count - len(file_stats), + ) + initial_file_count = len(file_stats) - # Then, eliminate files based on exclude pattern + # PERFORMANCE: Time exclude filtering + exclude_elapsed = 0.0 if exclude is not None: - files = exclude_files(exclude, files) + exclude_start = time.time() + file_list = list(file_stats.keys()) + filtered_list = exclude_files(exclude, file_list) + # Keep only files that passed the filter + file_stats = {path: file_stats[path] for path in filtered_list} + exclude_elapsed = time.time() - exclude_start + perf_logger.log_filter( + "exclude", + exclude, + exclude_elapsed, + len(file_stats), + initial_file_count - len(file_stats), + ) + + # PERFORMANCE: Time the sorting operation to maintain deterministic order + sort_start = time.time() + # Sort paths to match original behavior (directory first, then filename) + # Use an OrderedDict to preserve the sorted order + sorted_paths = sorted(file_stats.keys()) + file_stats = OrderedDict((path, file_stats[path]) for path in sorted_paths) + sort_elapsed = time.time() - sort_start + perf_logger.log_sort(len(file_stats), sort_elapsed) + + gather_total_elapsed = time.time() - gather_total_start + perf_logger.log_separator() + perf_logger.log_total( + "get_files_to_archive_with_stats", gather_total_elapsed, len(file_stats) + ) + + # Breakdown percentages + perf_logger.log_breakdown( + "get_files_to_archive_with_stats", + gather_total_elapsed, + walk_elapsed, + include_elapsed, + exclude_elapsed, + sort_elapsed, + include is not None, + exclude is not None, + ) + perf_logger.log_separator() + + return file_stats + - return files +def get_files_to_archive(cache: str, include: str, exclude: str) -> List[str]: + """ + LEGACY VERSION: For backward compatibility. + Uses the optimized version but returns only the file list. + """ + file_stats = get_files_to_archive_with_stats(cache, include, exclude) + return list(file_stats.keys()) def update_config(cur: sqlite3.Cursor):