From 3f4e1425a2c6e2740c8ff15fc794c575ea892154 Mon Sep 17 00:00:00 2001 From: Wendell Jones Date: Wed, 3 Jun 2026 09:12:20 -0400 Subject: [PATCH] Refactor pg_backup.sh to track newest backup files and update deletion logic - Changed variable names from oldest to newest for clarity. - Updated logic to find the newest backup file instead of the oldest. - Adjusted logging to reflect the change in backup file tracking. - Modified file deletion criteria to use the newest backup timestamp. Add batch_rsync.ps1 for parallel rsync operations in PowerShell - Implemented a PowerShell script to launch multiple rsync processes concurrently. - Added features for auto-tuning worker count based on file size and destination type. - Included verbose output options and support for dry runs. - Ensured compatibility with Windows paths and rsync arguments. Enhance pyrsync.py with recursive file handling and auto-tuning - Added support for balanced recursive file chunking in rsync operations. - Implemented auto-tuning of worker count based on average file size. - Improved output handling and error reporting for rsync tasks. - Updated command-line argument parsing for enhanced usability. Create sync_data.ps1 for FreeFileSync job management - Developed a PowerShell script to manage synchronization jobs using FreeFileSync. - Implemented logging for synchronization processes and error handling. - Added functionality to copy HTML reports generated by FreeFileSync. - Included an option to force shutdown regardless of synchronization success. --- personal/configs/frigate/config.yml | 5 + personal/configs/go2rtc/go2rtc.yaml | 43 ++ personal/cron_scripts/pg_backup.sh | 20 +- personal/tools/batch_rsync.ps1 | 536 ++++++++++++++++++ personal/tools/pyrsync.py | 338 ++++++++++- .../{cron_scripts => tools}/sync_data.ps1 | 2 +- 6 files changed, 911 insertions(+), 33 deletions(-) create mode 100644 personal/configs/go2rtc/go2rtc.yaml create mode 100644 personal/tools/batch_rsync.ps1 rename personal/{cron_scripts => tools}/sync_data.ps1 (97%) diff --git a/personal/configs/frigate/config.yml b/personal/configs/frigate/config.yml index 50b709d..67dd6b0 100644 --- a/personal/configs/frigate/config.yml +++ b/personal/configs/frigate/config.yml @@ -26,6 +26,11 @@ record: motion: days: 3 go2rtc: + ffmpeg: + bin: ffmpeg + # Nvidia NVDEC hardware decode templates (used when #hardware is in stream URL) + h264: "-hwaccel cuda -hwaccel_output_format cuda -c:v h264_cuvid {input} -c:v copy -f rtsp {output}" + h265: "-hwaccel cuda -hwaccel_output_format cuda -c:v hevc_cuvid {input} -c:v copy -f rtsp {output}" streams: # Backbasement: # - ffmpeg:rtsp://tpadmin:Optimus0329@192.168.128.222:554/live/ch0#video=copy#audio=copy#audio=aac#hardware diff --git a/personal/configs/go2rtc/go2rtc.yaml b/personal/configs/go2rtc/go2rtc.yaml new file mode 100644 index 0000000..84bc773 --- /dev/null +++ b/personal/configs/go2rtc/go2rtc.yaml @@ -0,0 +1,43 @@ +api: + listen: ":1984" + +log: + level: "info" + format: "color" + output: "file:go2rtc.log" + +rtsp: + listen: ":8554" # RTSP Server TCP port, default - 8554 + #username: "admin" # optional, default - disabled + #password: "pass" # optional, default - disabled + #default_query: "video=h264" # optional, default codecs filters + +ffmpeg: + bin: ffmpeg + # Nvidia NVDEC hardware decode templates (used when #hardware is in stream URL) + h264: "-hwaccel cuda -hwaccel_output_format cuda -c:v h264_cuvid {input} -c:v copy -f rtsp {output}" + h265: "-hwaccel cuda -hwaccel_output_format cuda -c:v hevc_cuvid {input} -c:v copy -f rtsp {output}" + +streams: + porch_steps: rtsp://admin:Optimus0329@192.168.128.11:80/ch8_0.264#video=h264#hardware=cuda + driveway: rtsp://admin:Optimus0329@192.168.128.11:80/ch12_0.264#video=h264#hardware=cuda + fence_gate: rtsp://admin:Optimus0329@192.168.128.11:80/ch1_0.264#video=h264#hardware=cuda + shed_area: rtsp://admin:Optimus0329@192.168.128.11:80/ch13_0.264#video=h264#hardware=cuda + garden_area: rtsp://admin:Optimus0329@192.168.128.11:80/ch2_0.264#video=h264#hardware=cuda + backyard_1: rtsp://admin:Optimus0329@192.168.128.11:80/ch3_0.264#video=h264#hardware=cuda + backyard_2: rtsp://admin:Optimus0329@192.168.128.11:80/ch4_0.264#video=h264#hardware=cuda + backyard_3: rtsp://admin:Optimus0329@192.168.128.11:80/ch0_0.264#video=h264#hardware=cuda + backyard_4: rtsp://admin:Optimus0329@192.168.128.11:80/ch10_0.264#video=h264#hardware=cuda + backyard_5: rtsp://admin:Optimus0329@192.168.128.11:80/ch7_0.264#video=h264#hardware=cuda + inside_garage: rtsp://admin:Optimus0329@192.168.128.11:80/ch6_0.264#video=h264#hardware=cuda + pool_area: rtsp://admin:Optimus0329@192.168.128.11:80/ch9_0.264#video=h264#hardware=cuda + back_slider: rtsp://admin:Optimus0329@192.168.128.11:80/ch5_0.264#video=h264#hardware=cuda + side_house: rtsp://admin:Optimus0329@192.168.128.11:80/ch11_0.264#video=h264#hardware=cuda + patio_area: rtsp://admin:Optimus0329@192.168.128.11:80/ch14_0.264 + engineering: rtsp://admin:Optimus0329@192.168.128.65:1554/live/ch0 + front_basement: rtsp://admin:Optimus0329@192.168.128.223:554/live/ch0 + back_basement: rtsp://admin:Optimus0329@192.168.128.95:554/live/ch0 + back_basement2: rtsp://admin:Optimus0329@192.168.128.183:554/live/ch0 + +webrtc: + listen: ":8555" diff --git a/personal/cron_scripts/pg_backup.sh b/personal/cron_scripts/pg_backup.sh index ac4485e..40b2f36 100644 --- a/personal/cron_scripts/pg_backup.sh +++ b/personal/cron_scripts/pg_backup.sh @@ -21,9 +21,9 @@ WAL_ARCHIVE_DIR="/var/lib/postgresql_archive/18/main/pg_wal" WAL_MIN_AGE_DAYS=7 clean_walfiles() { - local oldest_backup="" - local oldest_backup_ts="" - local oldest_backup_dt="" + local newest_backup="" + local newest_backup_ts="" + local newest_backup_dt="" local age_cutoff_ts="" local age_cutoff_dt="" local backup_ts="" @@ -53,26 +53,26 @@ clean_walfiles() { for backup_file in "${backup_files[@]}"; do backup_ts=$(stat -c %Y "$backup_file") - if [[ -z "$oldest_backup_ts" || "$backup_ts" -lt "$oldest_backup_ts" ]]; then - oldest_backup_ts="$backup_ts" - oldest_backup="$backup_file" + if [[ -z "$newest_backup_ts" || "$backup_ts" -gt "$newest_backup_ts" ]]; then + newest_backup_ts="$backup_ts" + newest_backup="$backup_file" fi done age_cutoff_ts=$(date -d "$WAL_MIN_AGE_DAYS days ago" +%s) - oldest_backup_dt=$(date -d "@$oldest_backup_ts" '+%Y-%m-%d %H:%M:%S') + newest_backup_dt=$(date -d "@$newest_backup_ts" '+%Y-%m-%d %H:%M:%S') age_cutoff_dt=$(date -d "@$age_cutoff_ts" '+%Y-%m-%d %H:%M:%S') log "WAL archive dir : $WAL_ARCHIVE_DIR" - log "Oldest .backup file : $oldest_backup" - log "Oldest .backup mtime : $oldest_backup_dt" + log "Newest .backup file : $newest_backup" + log "Newest .backup mtime : $newest_backup_dt" log "WAL age cutoff : $age_cutoff_dt" while IFS= read -r -d '' file; do file_ts=$(stat -c %Y "$file") # Delete only when both constraints are true. - if [[ "$file_ts" -lt "$oldest_backup_ts" && "$file_ts" -lt "$age_cutoff_ts" ]]; then + if [[ "$file_ts" -lt "$newest_backup_ts" && "$file_ts" -lt "$age_cutoff_ts" ]]; then files_to_delete+=("$file") fi done < <(find "$WAL_ARCHIVE_DIR" -maxdepth 1 -type f ! -name '*.backup' -print0) diff --git a/personal/tools/batch_rsync.ps1 b/personal/tools/batch_rsync.ps1 new file mode 100644 index 0000000..565299a --- /dev/null +++ b/personal/tools/batch_rsync.ps1 @@ -0,0 +1,536 @@ +#Requires -Version 7.0 +# ============================================================================= +# batch_rsync.ps1 — Parallel rsync launcher for Windows PowerShell +# ============================================================================= +# +# KEY IMPLEMENTATION NOTES +# ------------------------ +# +# Parallelism +# Uses RunspacePool + [PowerShell]::Create() so multiple rsync processes +# run concurrently without blocking the main thread. All runspace script +# blocks are self-contained (no closures over outer variables). +# +# As-completed result collection +# WaitHandle.WaitAny() polls the async handles so results are printed as +# soon as each task finishes, matching the behaviour of Python's as_completed(). +# +# --files-from (recursive-files mode) +# rsync is fed a NUL-delimited temp file via --from0 --files-from= +# rather than stdin, because stdin is not reliably available inside a +# runspace. The temp file is always cleaned up in a finally block. +# +# Path handling +# All source paths are converted from Windows backslashes to forward slashes +# before being passed to rsync, which is required for MSYS2/Cygwin rsync +# to interpret them correctly (e.g. C:/data/... not C:\data\...). +# +# Argument passing to rsync +# Uses ProcessStartInfo.ArgumentList (System.Collections.Generic.List) +# rather than a single argument string. This avoids quoting/escaping issues +# with paths that contain spaces or special characters. +# Requires .NET 5+ — hence the #Requires -Version 7.0 constraint. +# +# Auto-tune worker selection +# Workers are derived from average file size and destination type (local vs +# remote). Smaller files favour higher concurrency; large files and remote +# destinations are throttled to avoid saturating the link or dest I/O. +# The -Workers value is always treated as an upper bound. +# +# Size-balanced chunk partitioning (recursive-files mode) +# Files are sorted largest-first and distributed across worker buckets with +# a greedy least-loaded algorithm to minimise the slowest-worker stall. +# +# Remote destination detection +# Heuristics check for rsync://, :: (daemon), Windows drive letters, UNC +# paths, and user@host:/path patterns to decide whether the destination is +# local or remote without requiring an actual connection. +# +# Verbose output +# Uses PowerShell's built-in -Verbose switch (CmdletBinding) so there is no +# custom flag. Pass -Verbose on the command line to enable detailed output. +# +# Requirements +# - PowerShell 7.0+ (for ProcessStartInfo.ArgumentList via .NET 5+) +# - rsync on PATH (MSYS2, Cygwin, or WSL interop on Windows) +# +# ============================================================================= + +<# +.SYNOPSIS + Parallel rsync launcher for Windows PowerShell. + +.DESCRIPTION + Runs multiple rsync invocations in parallel using either top-level entries + or balanced recursive file chunks. Requires rsync on PATH (e.g. via MSYS2 + or Cygwin on Windows). + + Paths are converted to forward-slash notation for rsync compatibility. + MSYS2 rsync accepts Windows absolute paths using forward slashes (C:/path/...). + +.EXAMPLE + # Basic sync (recursive-files mode, auto worker count): + .\batch_rsync.ps1 C:\data user@host:/dest/ + +.EXAMPLE + # Auto-tune workers based on file size and local vs remote destination: + .\batch_rsync.ps1 C:\data user@host:/dest/ -AutoTune + +.EXAMPLE + # Limit to a specific number of parallel workers: + .\batch_rsync.ps1 C:\data D:\backup -Workers 8 + +.EXAMPLE + # Use top-level mode (one rsync process per immediate source entry): + .\batch_rsync.ps1 C:\data D:\backup -Mode top-level + +.EXAMPLE + # Dry run (show what would be transferred without doing it): + .\batch_rsync.ps1 C:\data user@host:/dest/ -DryRun + +.EXAMPLE + # Verbose output (command templates, task previews, elapsed time): + .\batch_rsync.ps1 C:\data D:\backup -Verbose + +.EXAMPLE + # Show rsync output for all tasks: + .\batch_rsync.ps1 C:\data D:\backup -ShowRsyncOutput + +.EXAMPLE + # Show rsync output only when a task fails: + .\batch_rsync.ps1 C:\data user@host:/dest/ -ShowRsyncOutputOnFailure + +.EXAMPLE + # Full-featured (auto-tune, verbose, show on failure): + .\batch_rsync.ps1 C:\data user@host:/dest/ -AutoTune -Verbose -ShowRsyncOutputOnFailure + +.EXAMPLE + # Override rsync flags entirely: + .\batch_rsync.ps1 C:\data D:\backup -RsyncArgs '-az --checksum' +#> + +[CmdletBinding()] +param ( + [Parameter(Mandatory, Position = 0)] + [string]$Src, + + [Parameter(Mandatory, Position = 1)] + [string]$Dest, + + [ValidateRange(1, [int]::MaxValue)] + [int]$Workers = [Math]::Min(8, [Environment]::ProcessorCount), + + [ValidateSet('recursive-files', 'top-level')] + [string]$Mode = 'recursive-files', + + [string]$RsyncArgs = '-a --partial --info=stats2', + + [switch]$AutoTune, + + [switch]$ShowRsyncOutput, + + [switch]$ShowRsyncOutputOnFailure, + + [switch]$DryRun +) + +Set-StrictMode -Version Latest +$ErrorActionPreference = 'Stop' + +#region Helper functions + +function Format-Bytes { + param([double]$Bytes) + $units = @('B', 'KB', 'MB', 'GB', 'TB') + $value = $Bytes + foreach ($unit in $units) { + if ($value -lt 1024 -or $unit -eq 'TB') { + return '{0:F1}{1}' -f $value, $unit + } + $value /= 1024 + } + return '{0:F1}B' -f $Bytes +} + +function Test-RemoteDestination { + param([string]$Destination) + if ($Destination -match '^rsync://') { return $true } + if ($Destination -match '::') { return $true } + # Windows drive letters and UNC paths are local + if ($Destination -match '^[a-zA-Z]:[/\\]') { return $false } + if ($Destination -match '^\\\\') { return $false } + # user@host:/path or host:/path patterns are remote + return $Destination -match '^[^:/\\]+@?[^:/\\]*:[^/]' +} + +function Get-AutoTunedWorkers { + param( + [int] $TotalUnits, + [double]$AvgSizeBytes, + [bool] $IsRemote, + [int] $CpuCount, + [int] $MaxWorkers + ) + $cpu = [Math]::Max(1, $CpuCount) + $MB1 = 1 * 1024 * 1024 + $MB32 = 32 * 1024 * 1024 + $MB256 = 256 * 1024 * 1024 + + if ($AvgSizeBytes -lt $MB1) { + $base = [Math]::Min(32, $cpu * 4) + } elseif ($AvgSizeBytes -lt $MB32) { + $base = [Math]::Min(16, $cpu * 2) + } elseif ($AvgSizeBytes -lt $MB256) { + $base = [Math]::Min(8, $cpu) + } else { + $base = [Math]::Min(4, [Math]::Max(2, [int]($cpu / 2))) + } + + if ($IsRemote) { + $base = [Math]::Max(2, [Math]::Min($base, 8)) + } else { + $base = [Math]::Min($base + 2, 24) + } + + $tuned = [Math]::Max(1, [Math]::Min($base, $TotalUnits)) + return [Math]::Min($tuned, $MaxWorkers) +} + +function Get-RecursiveFileList { + param([string]$SrcDir) + $root = $SrcDir.TrimEnd('\', '/') + Get-ChildItem -LiteralPath $root -Recurse -File | + ForEach-Object { $_.FullName.Substring($root.Length).TrimStart('\', '/') } +} + +function Get-SizedFiles { + param([string]$SrcDir, [string[]]$RelFiles) + foreach ($rel in $RelFiles) { + $abs = Join-Path $SrcDir $rel + try { $size = (Get-Item -LiteralPath $abs -ErrorAction Stop).Length } + catch { $size = 0 } + [PSCustomObject]@{ Size = [long]$size; RelPath = $rel } + } +} + +function Get-BalancedChunks { + param($SizedFiles, [int]$Workers) + $buckets = @(1..$Workers | ForEach-Object { [System.Collections.Generic.List[string]]::new() }) + $bucketSizes = [long[]]::new($Workers) + + foreach ($entry in ($SizedFiles | Sort-Object -Property Size -Descending)) { + $minIdx = 0 + for ($i = 1; $i -lt $Workers; $i++) { + if ($bucketSizes[$i] -lt $bucketSizes[$minIdx]) { $minIdx = $i } + } + $buckets[$minIdx].Add($entry.RelPath) + $bucketSizes[$minIdx] += $entry.Size + } + + $buckets | Where-Object { $_.Count -gt 0 } +} + +#endregion + +#region Runspace script blocks +# These blocks run inside isolated runspaces — all needed logic must be self-contained. + +$rsyncChunkScriptBlock = { + param( + [string] $SrcDir, + [string[]]$RelFiles, + [string] $Dest, + [string[]]$RsyncArgsList, + [bool] $IsDryRun + ) + + $filesFromTmp = [System.IO.Path]::GetTempFileName() + try { + # Build a NUL-delimited file list using forward slashes for rsync + $lines = $RelFiles | ForEach-Object { $_ -replace '\\', '/' } + $allBytes = [System.Collections.Generic.List[byte]]::new() + foreach ($line in $lines) { + $allBytes.AddRange([System.Text.Encoding]::UTF8.GetBytes($line)) + $allBytes.Add([byte]0) + } + [System.IO.File]::WriteAllBytes($filesFromTmp, $allBytes.ToArray()) + + $srcRoot = ($SrcDir.TrimEnd('\', '/') -replace '\\', '/') + '/' + + $cmdArgs = [System.Collections.Generic.List[string]]::new($RsyncArgsList) + if ($IsDryRun) { $cmdArgs.Insert(0, '--dry-run') } + $cmdArgs.AddRange([string[]]@('--from0', "--files-from=$filesFromTmp", $srcRoot, $Dest)) + + $psi = [System.Diagnostics.ProcessStartInfo]::new('rsync') + $psi.RedirectStandardOutput = $true + $psi.RedirectStandardError = $true + $psi.UseShellExecute = $false + $psi.CreateNoWindow = $true + foreach ($arg in $cmdArgs) { $psi.ArgumentList.Add($arg) } + + try { + $proc = [System.Diagnostics.Process]::Start($psi) + $stdout = $proc.StandardOutput.ReadToEnd() + $stderr = $proc.StandardError.ReadToEnd() + $proc.WaitForExit() + $exitCode = $proc.ExitCode + } catch { + $stdout = '' + $stderr = "rsync not found on PATH: $($_.Exception.Message)" + $exitCode = 127 + } + + return [PSCustomObject]@{ + Label = "chunk($($RelFiles.Count) files)" + ExitCode = $exitCode + Stdout = $stdout + Stderr = $stderr + } + } finally { + Remove-Item -LiteralPath $filesFromTmp -Force -ErrorAction SilentlyContinue + } +} + +$rsyncItemScriptBlock = { + param( + [string] $Item, + [string] $Dest, + [string[]]$RsyncArgsList, + [bool] $IsDryRun + ) + + $srcPath = $Item -replace '\\', '/' + + $cmdArgs = [System.Collections.Generic.List[string]]::new($RsyncArgsList) + if ($IsDryRun) { $cmdArgs.Insert(0, '--dry-run') } + $cmdArgs.Add($srcPath) + $cmdArgs.Add($Dest) + + $psi = [System.Diagnostics.ProcessStartInfo]::new('rsync') + $psi.RedirectStandardOutput = $true + $psi.RedirectStandardError = $true + $psi.UseShellExecute = $false + $psi.CreateNoWindow = $true + foreach ($arg in $cmdArgs) { $psi.ArgumentList.Add($arg) } + + try { + $proc = [System.Diagnostics.Process]::Start($psi) + $stdout = $proc.StandardOutput.ReadToEnd() + $stderr = $proc.StandardError.ReadToEnd() + $proc.WaitForExit() + $exitCode = $proc.ExitCode + } catch { + $stdout = '' + $stderr = "rsync not found on PATH: $($_.Exception.Message)" + $exitCode = 127 + } + + return [PSCustomObject]@{ + Label = $Item + ExitCode = $exitCode + Stdout = $stdout + Stderr = $stderr + } +} + +#endregion + +#region Parallel job runner (as-completed using WaitHandle.WaitAny) + +function Invoke-ParallelJobs { + param( + [object[]] $Jobs, # PSCustomObjects with .PS and .Handle + [bool] $ShowOutput, + [bool] $ShowOnFailOnly + ) + + $results = [System.Collections.Generic.List[object]]::new() + $pending = [System.Collections.Generic.List[object]]::new() + foreach ($j in $Jobs) { $pending.Add($j) } + + while ($pending.Count -gt 0) { + $handles = @($pending | ForEach-Object { $_.Handle.AsyncWaitHandle }) + $idx = [System.Threading.WaitHandle]::WaitAny($handles) + $job = $pending[$idx] + $pending.RemoveAt($idx) + + $r = $job.PS.EndInvoke($job.Handle)[0] + $job.PS.Dispose() + $results.Add($r) + + $status = if ($r.ExitCode -eq 0) { 'OK' } else { "ERR($($r.ExitCode))" } + Write-Host "[$status] $($r.Label)" + + $printOutput = $ShowOutput -and (-not $ShowOnFailOnly -or $r.ExitCode -ne 0) + if ($printOutput) { + if ($r.Stdout.Trim()) { + Write-Host "--- stdout: $($r.Label) ---" + Write-Host $r.Stdout.TrimEnd() + } + if ($r.Stderr.Trim()) { + Write-Warning "--- stderr: $($r.Label) ---`n$($r.Stderr.TrimEnd())" + } + } + } + + return $results +} + +#endregion + +#region Main + +$timer = [System.Diagnostics.Stopwatch]::StartNew() + +$rsyncArgsList = $RsyncArgs -split '\s+' | Where-Object { $_ -ne '' } +$srcPath = $Src.TrimEnd('\', '/') +$isRemote = Test-RemoteDestination -Destination $Dest +$cpuCount = [Environment]::ProcessorCount +$showOutput = $ShowRsyncOutput -or $ShowRsyncOutputOnFailure +$showOnFail = $ShowRsyncOutputOnFailure -and -not $ShowRsyncOutput + +if ($ShowRsyncOutput -and $ShowRsyncOutputOnFailure) { + Write-Warning 'Both -ShowRsyncOutput and -ShowRsyncOutputOnFailure set; showing output for all tasks.' +} + +$allResults = $null + +if ($Mode -eq 'recursive-files' -and (Test-Path -LiteralPath $srcPath -PathType Container)) { + + $relFiles = @(Get-RecursiveFileList -SrcDir $srcPath) + if (-not $relFiles -or $relFiles.Count -eq 0) { + Write-Error "No files found for: $Src" + exit 2 + } + + $sizedFiles = @(Get-SizedFiles -SrcDir $srcPath -RelFiles $relFiles) + $totalBytes = ($sizedFiles | Measure-Object -Property Size -Sum).Sum + $avgSize = if ($sizedFiles.Count -gt 0) { $totalBytes / $sizedFiles.Count } else { 0 } + + if ($AutoTune) { + $effectiveWorkers = Get-AutoTunedWorkers -TotalUnits $sizedFiles.Count -AvgSizeBytes $avgSize ` + -IsRemote $isRemote -CpuCount $cpuCount -MaxWorkers $Workers + $destKind = if ($isRemote) { 'remote' } else { 'local' } + Write-Host "Auto-tuned workers=$effectiveWorkers ($destKind dest, avg file $(Format-Bytes $avgSize))" + } else { + $effectiveWorkers = [Math]::Min($Workers, $sizedFiles.Count) + } + + $chunks = @(Get-BalancedChunks -SizedFiles $sizedFiles -Workers $effectiveWorkers) + Write-Host "Starting $($chunks.Count) rsync chunk task(s) with $effectiveWorkers worker(s)..." + + if ($VerbosePreference -ne 'SilentlyContinue') { + $srcRoot = ($srcPath -replace '\\', '/') + '/' + $previewArgs = (@('rsync') + $rsyncArgsList + @('--from0', '--files-from=', $srcRoot, $Dest)) + if ($DryRun) { $previewArgs = @('rsync', '--dry-run') + $previewArgs[1..($previewArgs.Count-1)] } + Write-Verbose "Command template: $($previewArgs -join ' ')" + } + + $pool = [RunspaceFactory]::CreateRunspacePool(1, $effectiveWorkers) + $pool.Open() + + $jobs = foreach ($chunk in $chunks) { + $ps = [PowerShell]::Create() + $ps.RunspacePool = $pool + [void]$ps.AddScript($rsyncChunkScriptBlock) + [void]$ps.AddParameter('SrcDir', $srcPath) + [void]$ps.AddParameter('RelFiles', [string[]]$chunk) + [void]$ps.AddParameter('Dest', $Dest) + [void]$ps.AddParameter('RsyncArgsList', [string[]]$rsyncArgsList) + [void]$ps.AddParameter('IsDryRun', $DryRun.IsPresent) + [PSCustomObject]@{ PS = $ps; Handle = $ps.BeginInvoke() } + } + + $allResults = Invoke-ParallelJobs -Jobs $jobs -ShowOutput $showOutput -ShowOnFailOnly $showOnFail + $pool.Close() + $pool.Dispose() + +} else { + + if (Test-Path -LiteralPath $srcPath -PathType Container) { + $items = @(Get-ChildItem -LiteralPath $srcPath | ForEach-Object { $_.FullName }) + } elseif ($srcPath -match '[*?]') { + $items = @(Resolve-Path -Path $srcPath | ForEach-Object { $_.ProviderPath }) + } else { + $items = @($srcPath) + } + + if (-not $items -or $items.Count -eq 0) { + Write-Error "No files found for: $Src" + exit 2 + } + + $sizeSamples = foreach ($item in $items) { + try { + $fi = Get-Item -LiteralPath $item -ErrorAction Stop + if ($fi.PSIsContainer) { [long](64 * 1024 * 1024) } else { $fi.Length } + } catch { [long]0 } + } + $totalBytes = ($sizeSamples | Measure-Object -Sum).Sum + $avgSize = if ($items.Count -gt 0) { $totalBytes / $items.Count } else { 0 } + + if ($AutoTune) { + $effectiveWorkers = Get-AutoTunedWorkers -TotalUnits $items.Count -AvgSizeBytes $avgSize ` + -IsRemote $isRemote -CpuCount $cpuCount -MaxWorkers $Workers + $destKind = if ($isRemote) { 'remote' } else { 'local' } + Write-Host "Auto-tuned workers=$effectiveWorkers ($destKind dest, avg unit $(Format-Bytes $avgSize))" + } else { + $effectiveWorkers = [Math]::Min($Workers, $items.Count) + } + + Write-Host "Starting $($items.Count) rsync task(s) with $effectiveWorkers worker(s)..." + + if ($VerbosePreference -ne 'SilentlyContinue') { + Write-Verbose 'Task list preview:' + $previewItems = $items | Select-Object -First 10 + for ($i = 0; $i -lt $previewItems.Count; $i++) { + Write-Verbose (' {0}. {1}' -f ($i + 1), $previewItems[$i]) + } + if ($items.Count -gt 10) { Write-Verbose " ... and $($items.Count - 10) more" } + } + + $pool = [RunspaceFactory]::CreateRunspacePool(1, $effectiveWorkers) + $pool.Open() + + $jobs = foreach ($item in $items) { + if ($VerbosePreference -ne 'SilentlyContinue') { + $previewArgs = (@('rsync') + $rsyncArgsList + @(($item -replace '\\', '/'), $Dest)) + if ($DryRun) { $previewArgs = @('rsync', '--dry-run') + $previewArgs[1..($previewArgs.Count-1)] } + Write-Verbose "Scheduling: $($previewArgs -join ' ')" + } + $ps = [PowerShell]::Create() + $ps.RunspacePool = $pool + [void]$ps.AddScript($rsyncItemScriptBlock) + [void]$ps.AddParameter('Item', $item) + [void]$ps.AddParameter('Dest', $Dest) + [void]$ps.AddParameter('RsyncArgsList', [string[]]$rsyncArgsList) + [void]$ps.AddParameter('IsDryRun', $DryRun.IsPresent) + [PSCustomObject]@{ PS = $ps; Handle = $ps.BeginInvoke() } + } + + $allResults = Invoke-ParallelJobs -Jobs $jobs -ShowOutput $showOutput -ShowOnFailOnly $showOnFail + $pool.Close() + $pool.Dispose() +} + +#endregion + +#region Summary + +$failed = @($allResults | Where-Object { $_.ExitCode -ne 0 }) +$elapsed = [Math]::Round($timer.Elapsed.TotalSeconds, 2) +Write-Host "`nCompleted: $($allResults.Count) tasks, $($failed.Count) failed in ${elapsed}s." + +if ($failed.Count -gt 0) { + Write-Host 'Failures detail:' + foreach ($r in $failed) { + Write-Host "--- $($r.Label) (exit $($r.ExitCode)) ---" + if ($r.Stdout.Trim()) { Write-Host $r.Stdout.TrimEnd() } + if ($r.Stderr.Trim()) { Write-Host $r.Stderr.TrimEnd() } + } + exit 1 +} + +exit 0 + +#endregion diff --git a/personal/tools/pyrsync.py b/personal/tools/pyrsync.py index 570080c..f6a94d1 100644 --- a/personal/tools/pyrsync.py +++ b/personal/tools/pyrsync.py @@ -3,18 +3,56 @@ Parallel rsync launcher. Requires rsync on PATH (e.g. via MSYS/Cygwin on Windows). -Runs multiple rsync invocations in parallel (one per file/entry). +Runs multiple rsync invocations in parallel using either top-level entries +or balanced recursive file chunks. """ +# Usage examples: +# +# Basic sync (recursive-files mode, auto worker count): +# python pyrsync.py /src/dir user@host:/dest/ +# +# Auto-tune workers based on file size and local vs remote destination: +# python pyrsync.py /src/dir user@host:/dest/ --auto-tune +# +# Limit to a specific number of parallel workers: +# python pyrsync.py /src/dir /local/dest/ -w 8 +# +# Use top-level mode (one rsync process per immediate source entry): +# python pyrsync.py /src/dir /local/dest/ --mode top-level +# +# Dry run (show what would be transferred without doing it): +# python pyrsync.py /src/dir user@host:/dest/ --dry-run +# +# Verbose output (shows command templates, task list preview, and elapsed time): +# python pyrsync.py /src/dir /local/dest/ -v +# +# Show rsync output for all tasks: +# python pyrsync.py /src/dir /local/dest/ --show-rsync-output +# +# Show rsync output only when a task fails: +# python pyrsync.py /src/dir user@host:/dest/ --show-rsync-output-on-failure +# +# Full-featured example (auto-tune, verbose, show failures): +# python pyrsync.py /src/dir user@host:/dest/ --auto-tune -v --show-rsync-output-on-failure +# +# Override rsync flags entirely: +# python pyrsync.py /src/dir /local/dest/ -r "-az --checksum" + + import argparse +import os +import re +import shlex import subprocess import sys +import time from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path -from typing import List +from typing import List, Sequence, Tuple -#DEFAULT_RSYNC_ARGS = ["-av", "--partial", "--compress", "--info=progress2"] -DEFAULT_RSYNC_ARGS = ["-avP", "--partial"] +# DEFAULT_RSYNC_ARGS = ["-av", "--partial", "--compress", "--info=progress2"] +DEFAULT_RSYNC_ARGS = ["-a", "--partial", "--info=stats2"] def build_file_list(src: str) -> List[Path]: @@ -29,7 +67,105 @@ def build_file_list(src: str) -> List[Path]: return [p] -def run_rsync(item: Path, dest: str, rsync_args: List[str], dry_run: bool) -> (Path, int, str, str): +def build_recursive_file_list(src_dir: Path) -> List[Path]: + """Return all files under src_dir as relative paths for --files-from.""" + return [p.relative_to(src_dir) for p in src_dir.rglob("*") if p.is_file()] + + +def get_sized_files(src_dir: Path, rel_files: Sequence[Path]) -> List[Tuple[int, Path]]: + """Return (size_bytes, rel_path) tuples with best-effort size lookup.""" + sized_files: List[Tuple[int, Path]] = [] + for rel_path in rel_files: + abs_path = src_dir / rel_path + try: + size = abs_path.stat().st_size + except OSError: + size = 0 + sized_files.append((size, rel_path)) + return sized_files + + +def partition_by_size(sized_files: Sequence[Tuple[int, Path]], workers: int) -> List[List[Path]]: + """Greedy size-balanced partitioning to reduce worker skew.""" + buckets: List[List[Path]] = [[] for _ in range(workers)] + bucket_sizes = [0] * workers + + # Largest files first gives better load balancing across workers. + sorted_files = sorted(sized_files, key=lambda x: x[0], reverse=True) + for size, rel_path in sorted_files: + idx = min(range(workers), key=lambda i: bucket_sizes[i]) + buckets[idx].append(rel_path) + bucket_sizes[idx] += size + + return [b for b in buckets if b] + + +def detect_remote_destination(dest: str) -> bool: + """Heuristic detection of rsync remote destinations.""" + if dest.startswith("rsync://"): + return True + if "::" in dest: + return True + if re.match(r"^[a-zA-Z]:[\\/]", dest): + return False + if re.match(r"^\\\\", dest): + return False + return bool(re.match(r"^[^:/\\]+@?[^:/\\]*:[^/].*", dest)) + + +def format_bytes(num_bytes: float) -> str: + units = ["B", "KB", "MB", "GB", "TB"] + value = float(num_bytes) + for unit in units: + if value < 1024.0 or unit == units[-1]: + return f"{value:.1f}{unit}" + value /= 1024.0 + return f"{num_bytes:.1f}B" + + +def auto_tune_workers( + total_units: int, + avg_size_bytes: float, + is_remote: bool, + cpu_count: int, + configured_workers: int, +) -> int: + """Pick worker count from workload size profile and destination type.""" + cpu = max(1, cpu_count) + + if avg_size_bytes < 1 * 1024 * 1024: + base = min(32, cpu * 4) + elif avg_size_bytes < 32 * 1024 * 1024: + base = min(16, cpu * 2) + elif avg_size_bytes < 256 * 1024 * 1024: + base = min(8, cpu) + else: + base = min(4, max(2, cpu // 2)) + + if is_remote: + base = max(2, min(base, 8)) + else: + base = min(base + 2, 24) + + tuned = max(1, min(base, total_units)) + return min(tuned, configured_workers) + + +def maybe_print_task_output(label: str, out: str, err: str, show_output: bool, show_on_fail_only: bool, code: int) -> None: + if not show_output: + return + if show_on_fail_only and code == 0: + return + + if out.strip(): + print(f"--- stdout: {label} ---") + print(out.rstrip()) + if err.strip(): + print(f"--- stderr: {label} ---", file=sys.stderr) + print(err.rstrip(), file=sys.stderr) + + +def run_rsync(item: Path, dest: str, rsync_args: List[str], dry_run: bool) -> Tuple[Path, int, str, str]: cmd = ["rsync", *rsync_args] if dry_run: cmd.append("--dry-run") @@ -42,40 +178,198 @@ def run_rsync(item: Path, dest: str, rsync_args: List[str], dry_run: bool) -> (P return item, 127, "", "rsync not found on PATH" +def run_rsync_files_from( + src_dir: Path, + rel_files: Sequence[Path], + dest: str, + rsync_args: List[str], + dry_run: bool, +) -> Tuple[str, int, str, str]: + cmd = ["rsync", *rsync_args] + if dry_run: + cmd.append("--dry-run") + + cmd += ["--from0", "--files-from=-", f"{src_dir}{os.sep}", dest] + + payload = "\0".join(str(p) for p in rel_files) + "\0" + try: + proc = subprocess.run(cmd, input=payload, capture_output=True, text=True) + label = f"chunk({len(rel_files)} files)" + return label, proc.returncode, proc.stdout, proc.stderr + except FileNotFoundError: + return "chunk(0 files)", 127, "", "rsync not found on PATH" + + def main(): ap = argparse.ArgumentParser(description="Transfer files using rsync in parallel.") ap.add_argument("src", help="Source file, directory or glob pattern") ap.add_argument("dest", help="Rsync destination (local path or user@host:/path)") - ap.add_argument("-w", "--workers", type=int, default=4, help="Parallel rsync workers") + ap.add_argument( + "-w", + "--workers", + type=int, + default=min(8, (os.cpu_count() or 4)), + help="Parallel rsync workers (default: min(8, CPU count))", + ) + ap.add_argument( + "--mode", + choices=["top-level", "recursive-files"], + default="recursive-files", + help=( + "top-level: one rsync per immediate src entry; " + "recursive-files: split full file tree into balanced rsync chunks" + ), + ) ap.add_argument("-r", "--rsync-args", default=" ".join(DEFAULT_RSYNC_ARGS), - help="Additional rsync args (quoted). Default: -a --partial --compress --progress") + help="Additional rsync args (quoted). Default: -a --partial --info=stats2") + ap.add_argument( + "--auto-tune", + action="store_true", + help=( + "Auto-tune worker count from average file size and destination type. " + "Uses --workers as an upper bound." + ), + ) + ap.add_argument( + "-v", + "--verbose", + action="store_true", + help="Show detailed progress, including task start/completion and rsync command lines.", + ) + ap.add_argument( + "--show-rsync-output", + action="store_true", + help="Print rsync stdout/stderr for each task (can be noisy).", + ) + ap.add_argument( + "--show-rsync-output-on-failure", + action="store_true", + help="Print rsync stdout/stderr only for failed tasks.", + ) ap.add_argument("--dry-run", action="store_true", help="Don't transfer, just show what would happen") args = ap.parse_args() - items = build_file_list(args.src) - if not items: - print("No files found for:", args.src, file=sys.stderr) + if args.workers < 1: + print("--workers must be >= 1", file=sys.stderr) sys.exit(2) - rsync_args = args.rsync_args.split() + rsync_args = shlex.split(args.rsync_args) results = [] - print(f"Starting {len(items)} rsync tasks with {args.workers} workers...") + started_at = time.perf_counter() - with ThreadPoolExecutor(max_workers=args.workers) as ex: - futures = {ex.submit(run_rsync, item, args.dest, rsync_args, args.dry_run): item for item in items} - for fut in as_completed(futures): - item, code, out, err = fut.result() - results.append((item, code, out, err)) - status = "OK" if code == 0 else f"ERR({code})" - print(f"[{status}] {item}") + if args.show_rsync_output and args.show_rsync_output_on_failure: + print( + "Both --show-rsync-output and --show-rsync-output-on-failure set; " + "showing output for all tasks.", + file=sys.stderr, + ) + + show_output = args.show_rsync_output or args.show_rsync_output_on_failure + show_on_fail_only = args.show_rsync_output_on_failure and not args.show_rsync_output + + src_path = Path(args.src) + is_remote_dest = detect_remote_destination(args.dest) + cpu_count = os.cpu_count() or 4 + if args.mode == "recursive-files" and src_path.is_dir(): + rel_files = build_recursive_file_list(src_path) + if not rel_files: + print("No files found for:", args.src, file=sys.stderr) + sys.exit(2) + + sized_files = get_sized_files(src_path, rel_files) + avg_size = sum(size for size, _ in sized_files) / max(1, len(sized_files)) + if args.auto_tune: + workers = auto_tune_workers(len(rel_files), avg_size, is_remote_dest, cpu_count, args.workers) + dest_kind = "remote" if is_remote_dest else "local" + print( + f"Auto-tuned workers={workers} ({dest_kind} dest, avg file {format_bytes(avg_size)})" + ) + else: + workers = min(args.workers, len(rel_files)) + + chunks = partition_by_size(sized_files, workers) + print(f"Starting {len(chunks)} rsync chunk task(s) with {workers} worker(s)...") + + if args.verbose: + cmd_preview = ["rsync", *rsync_args] + if args.dry_run: + cmd_preview.append("--dry-run") + cmd_preview += ["--from0", "--files-from=-", f"{src_path}{os.sep}", args.dest] + print("Command template:", " ".join(shlex.quote(part) for part in cmd_preview)) + + with ThreadPoolExecutor(max_workers=workers) as ex: + futures = { + ex.submit(run_rsync_files_from, src_path, chunk, args.dest, rsync_args, args.dry_run): chunk + for chunk in chunks + } + for fut in as_completed(futures): + label, code, out, err = fut.result() + results.append((label, code, out, err)) + status = "OK" if code == 0 else f"ERR({code})" + print(f"[{status}] {label}") + maybe_print_task_output(label, out, err, show_output, show_on_fail_only, code) + else: + items = build_file_list(args.src) + if not items: + print("No files found for:", args.src, file=sys.stderr) + sys.exit(2) + + size_samples = [] + for item in items: + try: + if item.is_file(): + size_samples.append(item.stat().st_size) + else: + size_samples.append(64 * 1024 * 1024) + except OSError: + size_samples.append(0) + + avg_size = sum(size_samples) / max(1, len(size_samples)) + if args.auto_tune: + workers = auto_tune_workers(len(items), avg_size, is_remote_dest, cpu_count, args.workers) + dest_kind = "remote" if is_remote_dest else "local" + print( + f"Auto-tuned workers={workers} ({dest_kind} dest, avg unit {format_bytes(avg_size)})" + ) + else: + workers = min(args.workers, len(items)) + + print(f"Starting {len(items)} rsync task(s) with {workers} worker(s)...") + + if args.verbose: + print("Task list preview:") + preview_limit = 10 + for idx, item in enumerate(items[:preview_limit], start=1): + print(f" {idx}. {item}") + if len(items) > preview_limit: + print(f" ... and {len(items) - preview_limit} more") + + with ThreadPoolExecutor(max_workers=workers) as ex: + futures = {} + for item in items: + if args.verbose: + cmd_preview = ["rsync", *rsync_args] + if args.dry_run: + cmd_preview.append("--dry-run") + cmd_preview += [str(item), args.dest] + print("Scheduling:", " ".join(shlex.quote(part) for part in cmd_preview)) + futures[ex.submit(run_rsync, item, args.dest, rsync_args, args.dry_run)] = item + + for fut in as_completed(futures): + item, code, out, err = fut.result() + results.append((str(item), code, out, err)) + status = "OK" if code == 0 else f"ERR({code})" + print(f"[{status}] {item}") + maybe_print_task_output(str(item), out, err, show_output, show_on_fail_only, code) # summary failed = [r for r in results if r[1] != 0] - print(f"\nCompleted: {len(results)} tasks, {len(failed)} failed.") + elapsed = time.perf_counter() - started_at + print(f"\nCompleted: {len(results)} tasks, {len(failed)} failed in {elapsed:.2f}s.") if failed: print("Failures detail:") - for item, code, out, err in failed: - print(f"--- {item} (exit {code}) ---") + for label, code, out, err in failed: + print(f"--- {label} (exit {code}) ---") if out: print(out.strip()) if err: diff --git a/personal/cron_scripts/sync_data.ps1 b/personal/tools/sync_data.ps1 similarity index 97% rename from personal/cron_scripts/sync_data.ps1 rename to personal/tools/sync_data.ps1 index bc41a74..f8e1019 100644 --- a/personal/cron_scripts/sync_data.ps1 +++ b/personal/tools/sync_data.ps1 @@ -102,4 +102,4 @@ if ($HadFatalError) { } Write-Log 'Shutting down server now.' -Stop-Computer -Force +Invoke-CimMethod -ClassName Win32_OperatingSystem -MethodName Win32Shutdown -Arguments @{ Flags = 5 }