Raincloud/ProjectStructure_HOME
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

zipseq 7z instance tolerance for RAM overflow

Browse Source
This commit is contained in:
Raincloud
2025-11-10 20:56:57 -07:00
parent f795ea3848
commit ff6d3f584f
3 changed files with 3864 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

322
zip_sequences.py
View File

@@ -12,6 +12,7 @@ from __future__ import annotations
import argparse
import json
import os
import platform
import shutil
import subprocess
import sys
@@ -21,6 +22,22 @@ from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from typing import Iterator, Sequence
# Try to import psutil for cross-platform RAM detection
try:
import psutil
HAS_PSUTIL = True
except ImportError:
HAS_PSUTIL = False
# For Windows fallback
if platform.system() == "Windows":
try:
import ctypes
HAS_CTYPES = True
except ImportError:
HAS_CTYPES = False
else:
HAS_CTYPES = False
RENDER_ROOT = Path("Renders")
ARCHIVE_ROOT = RENDER_ROOT / "_zipped"
@@ -37,6 +54,7 @@ DEFAULT_CONFIG = {
"zipper": "7z",
"compression": 9,
"dailyFormat": "daily_YYMMDD",
"Max7zInst": 0, # Maximum concurrent 7z instances (0 = auto-calculate)
}
@@ -89,6 +107,19 @@ if not isinstance(COMPRESSION_LEVEL, int):
COMPRESSION_LEVEL = 9
COMPRESSION_LEVEL = max(0, min(9, COMPRESSION_LEVEL))
MAX_7Z_INSTANCES = CONFIG.get("Max7zInst", 0)
if MAX_7Z_INSTANCES is not None:
if isinstance(MAX_7Z_INSTANCES, str):
try:
MAX_7Z_INSTANCES = int(MAX_7Z_INSTANCES)
except ValueError:
MAX_7Z_INSTANCES = 0
if not isinstance(MAX_7Z_INSTANCES, int) or MAX_7Z_INSTANCES < 1:
MAX_7Z_INSTANCES = 0
# Treat 0 as None (auto-calculate)
if MAX_7Z_INSTANCES == 0:
MAX_7Z_INSTANCES = None
SEVEN_Z_EXE: str | None = None
if ZIPPER_TYPE == "7z":
SEVEN_Z_EXE = shutil.which("7z") or shutil.which("7za")
@@ -107,12 +138,219 @@ def parse_args() -> argparse.Namespace:
return parser.parse_args()
def max_workers(requested: int | None) -> int:
def get_available_ram() -> int | None:
"""Get available RAM in bytes, reserving 20% for system.
Returns:
Available RAM in bytes, or None if detection fails.
"""
try:
if HAS_PSUTIL:
# Use psutil for cross-platform RAM detection
mem = psutil.virtual_memory()
# Reserve 20% for system, use 80% for compression jobs
available = int(mem.total * 0.8)
return available
elif HAS_CTYPES and platform.system() == "Windows":
# Windows fallback using ctypes
class MEMORYSTATUSEX(ctypes.Structure):
_fields_ = [
("dwLength", ctypes.c_ulong),
("dwMemoryLoad", ctypes.c_ulong),
("ullTotalPhys", ctypes.c_ulonglong),
("ullAvailPhys", ctypes.c_ulonglong),
("ullTotalPageFile", ctypes.c_ulonglong),
("ullAvailPageFile", ctypes.c_ulonglong),
("ullTotalVirtual", ctypes.c_ulonglong),
("ullAvailVirtual", ctypes.c_ulonglong),
("ullAvailExtendedVirtual", ctypes.c_ulonglong),
]
kernel32 = ctypes.windll.kernel32
kernel32.GlobalMemoryStatusEx.argtypes = [ctypes.POINTER(MEMORYSTATUSEX)]
kernel32.GlobalMemoryStatusEx.restype = ctypes.c_bool
mem_status = MEMORYSTATUSEX()
mem_status.dwLength = ctypes.sizeof(MEMORYSTATUSEX)
if kernel32.GlobalMemoryStatusEx(ctypes.byref(mem_status)):
# Reserve 20% for system, use 80% for compression jobs
available = int(mem_status.ullTotalPhys * 0.8)
return available
except Exception:
pass
return None
def estimate_ram_per_job(seq_dir: Path, seq_state: dict) -> int:
"""Estimate RAM usage per compression job based on folder size.
Args:
seq_dir: Path to the sequence directory
seq_state: State dictionary containing file information
Returns:
Estimated RAM usage in bytes
"""
# Calculate total folder size from seq_state
total_bytes = sum(entry.get("size", 0) for entry in seq_state.get("files", []))
if ZIPPER_TYPE == "7z":
# Base RAM: 500MB per job
base_ram = 500 * 1024 * 1024 # 500 MB
# Compression factor: 7z can use significant RAM, especially for large files
# Use 0.15x factor (conservative estimate accounting for 7z's 80% usage)
compression_factor = 0.15
# For very large folders (>10GB), cap at 8GB per job
max_ram_per_job = 8 * 1024 * 1024 * 1024 # 8 GB
large_folder_threshold = 10 * 1024 * 1024 * 1024 # 10 GB
if total_bytes > large_folder_threshold:
estimated_ram = max_ram_per_job
else:
estimated_ram = max(base_ram, int(total_bytes * compression_factor))
estimated_ram = min(estimated_ram, max_ram_per_job)
return estimated_ram
else:
# zip compression is more memory-efficient
# Conservative estimate: 1GB per job
return 1024 * 1024 * 1024 # 1 GB
def max_workers(
requested: int | None,
work_items: list[tuple[Path, Path, Path, dict]] | None = None,
*,
verbose: bool = False
) -> tuple[int, int | None]:
"""Calculate maximum worker count based on CPU and RAM constraints.
Args:
requested: User-requested worker count (from --jobs)
work_items: List of work items (seq_dir, zip_path, state_path, seq_state)
verbose: Whether to log RAM-based calculations
Returns:
Tuple of (worker_count, per_job_memory_limit_bytes)
per_job_memory_limit_bytes is None if not using 7z or RAM detection failed
"""
cpu = os.cpu_count() or 1
limit = max(1, min(8, cpu))
cpu_limit = max(1, min(8, cpu))
if requested and requested > 0:
return min(requested, max(1, cpu))
return limit
cpu_limit = min(requested, max(1, cpu))
# If no work items provided, return CPU-based limit
if work_items is None or len(work_items) == 0:
return (cpu_limit, None)
# Try to calculate RAM-based limit
available_ram = get_available_ram()
if available_ram is None:
# RAM detection failed, fall back to CPU limit
if verbose:
log("zip", "RAM detection failed, using CPU-based worker limit", verbose_only=True, verbose=verbose)
return (cpu_limit, None)
# For 7z: use fixed dictionary size and calculate workers
if ZIPPER_TYPE == "7z":
# Fixed dictionary size: 1GB (1024MB)
FIXED_DICT_SIZE_MB = 1024
fixed_dict_size_bytes = FIXED_DICT_SIZE_MB * 1024 * 1024
# Check if Max7zInst is configured
if MAX_7Z_INSTANCES is not None:
# Use configured maximum instances, but still respect user's --jobs
final_limit = MAX_7Z_INSTANCES
if requested and requested > 0:
final_limit = min(final_limit, requested)
if verbose:
log(
"zip",
f"Using Max7zInst={MAX_7Z_INSTANCES} from config → "
f"requested: {requested}, final: {final_limit}",
verbose_only=True,
verbose=verbose
)
return (final_limit, fixed_dict_size_bytes)
# Auto-calculate based on RAM if Max7zInst not configured
if available_ram is not None:
# 7z uses ~2-3x dictionary size in RAM, but with overhead use 8x for safety
# This accounts for 7z's internal buffers, OS overhead, and other processes
FIXED_RAM_PER_JOB = FIXED_DICT_SIZE_MB * 8 * 1024 * 1024 # 8GB per job
# available_ram is already 80% of total (20% reserved for system)
# Use only 40% of that for compression jobs (very conservative to prevent swapping)
compression_ram = int(available_ram * 0.4)
# Calculate worker limit based on fixed per-job RAM
ram_limit = max(1, compression_ram // FIXED_RAM_PER_JOB)
# Use RAM limit directly (no CPU limit)
final_limit = ram_limit
if requested and requested > 0:
final_limit = min(final_limit, requested)
if verbose:
ram_gb = available_ram / (1024 ** 3)
compression_ram_gb = compression_ram / (1024 ** 3)
ram_per_job_gb = FIXED_RAM_PER_JOB / (1024 ** 3)
log(
"zip",
f"RAM: {ram_gb:.1f}GB available (80% of total), {compression_ram_gb:.1f}GB for compression (40%), {ram_per_job_gb:.1f}GB per job (dict: {FIXED_DICT_SIZE_MB}MB) → "
f"RAM limit: {ram_limit}, requested: {requested}, final: {final_limit}",
verbose_only=True,
verbose=verbose
)
return (final_limit, fixed_dict_size_bytes)
# RAM detection failed, use a safe default (no CPU limit)
if verbose:
log("zip", "RAM detection failed and Max7zInst not set, using default worker limit of 4", verbose_only=True, verbose=verbose)
default_limit = 4
if requested and requested > 0:
default_limit = requested
return (default_limit, fixed_dict_size_bytes)
# For zip compression, use existing estimation-based approach
# Estimate RAM per job for each work item
ram_estimates = []
for seq_dir, zip_path, state_path, seq_state in work_items:
try:
estimated_ram = estimate_ram_per_job(seq_dir, seq_state)
ram_estimates.append(estimated_ram)
except Exception:
# If estimation fails, use fallback estimate
ram_estimates.append(1024 * 1024 * 1024) # 1GB fallback for zip
if not ram_estimates:
return (cpu_limit, None)
max_ram_per_job = max(ram_estimates)
ram_limit = max(1, available_ram // max_ram_per_job)
ram_limit = min(ram_limit, 6) # Conservative limit for zip
final_limit = min(cpu_limit, ram_limit)
if verbose:
ram_gb = available_ram / (1024 ** 3)
max_ram_gb = max_ram_per_job / (1024 ** 3)
log(
"zip",
f"RAM: {ram_gb:.1f}GB available, ~{max_ram_gb:.1f}GB per job → "
f"RAM limit: {ram_limit}, CPU limit: {cpu_limit}, final: {final_limit}",
verbose_only=True,
verbose=verbose
)
return (final_limit, None)
def log(mode: str, message: str, *, verbose_only: bool = False, verbose: bool = False) -> None:
@@ -200,7 +438,7 @@ def state_path_for(zip_path: Path) -> Path:
return zip_path.with_suffix(zip_path.suffix + STATE_SUFFIX)
def zip_sequence(seq_dir: Path, zip_path: Path) -> None:
def zip_sequence(seq_dir: Path, zip_path: Path, per_job_memory_limit: int | None = None) -> None:
if ZIPPER_TYPE == "7z":
if SEVEN_Z_EXE is None:
raise RuntimeError(
@@ -258,9 +496,18 @@ def zip_sequence(seq_dir: Path, zip_path: Path) -> None:
"-bb0", # Suppress progress output
f"-mx={COMPRESSION_LEVEL}",
"-t7z", # Use 7z format, not zip
]
# Add fixed dictionary size if specified (7z memory usage is controlled by dictionary size)
if per_job_memory_limit is not None:
# per_job_memory_limit is actually the fixed dictionary size in bytes
dict_size_mb = per_job_memory_limit // (1024 * 1024)
cmd.append(f"-md={dict_size_mb}m")
cmd.extend([
str(temp_zip_abs),
f"@{list_file_abs}",
]
])
result = subprocess.run(
cmd,
cwd=seq_dir,
@@ -376,9 +623,9 @@ def expand_sequence(zip_path: Path, seq_state: dict) -> None:
os.utime(file_path, ns=(entry["mtime_ns"], entry["mtime_ns"]))
def process_zip(seq_dir: Path, zip_path: Path, state_path: Path, seq_state: dict, *, verbose: bool) -> Sequence[Path]:
def process_zip(seq_dir: Path, zip_path: Path, state_path: Path, seq_state: dict, per_job_memory_limit: int | None, *, verbose: bool) -> Sequence[Path]:
log("zip", f"{seq_dir} -> {zip_path}", verbose_only=True, verbose=verbose)
zip_sequence(seq_dir, zip_path)
zip_sequence(seq_dir, zip_path, per_job_memory_limit)
state_path.write_text(json.dumps(seq_state, indent=2))
return (zip_path, state_path)
@@ -388,39 +635,57 @@ def process_expand(zip_path: Path, state: dict, *, verbose: bool) -> None:
expand_sequence(zip_path, state)
def run_zip(worker_count: int, *, verbose: bool) -> int:
def run_zip(requested_workers: int | None, *, verbose: bool) -> int:
work_items: list[tuple[Path, Path, Path, dict]] = []
if RENDER_ROOT.exists():
for seq_dir in find_sequence_dirs(RENDER_ROOT):
seq_state = compute_state(seq_dir)
if not seq_state["files"]:
continue
# Get the target archive path (will be .7z if ZIPPER_TYPE is "7z")
zip_path = archive_path_for(seq_dir)
state_path = state_path_for(zip_path)
# Quick check: if archive exists, load stored state first (fast)
stored_state = load_state(state_path)
# Check if we need to upgrade from .zip to .7z
old_zip_path = None
if ZIPPER_TYPE == "7z":
# Check if an old .zip file exists
old_zip_path = zip_path.with_suffix(".zip")
if old_zip_path.exists():
# Check if the old .zip's metadata matches current state
old_state_path = state_path_for(old_zip_path)
old_stored_state = load_state(old_state_path)
if not state_changed(seq_state, old_stored_state):
# Old .zip is up to date, skip conversion
continue
# Old .zip is out of date, will be replaced with .7z
# If old .zip exists and archive doesn't, we'll check state later
if not zip_path.exists() and old_stored_state is not None:
stored_state = old_stored_state
# Check if the target archive (e.g., .7z) already exists and is up to date
stored_state = load_state(state_path)
if not state_changed(seq_state, stored_state):
# If archive exists and we have stored state, do quick check before computing full state
if zip_path.exists() and stored_state is not None:
# Quick check: if directory mtime is older than archive, likely unchanged
try:
dir_mtime = seq_dir.stat().st_mtime_ns
archive_mtime = zip_path.stat().st_mtime_ns
# If directory wasn't modified since archive was created, skip state computation
if dir_mtime <= archive_mtime:
# Still need to check for old .zip cleanup
if old_zip_path and old_zip_path.exists():
old_zip_path.unlink(missing_ok=True)
old_state_path = state_path_for(old_zip_path)
if old_state_path.exists():
old_state_path.unlink(missing_ok=True)
continue
except OSError:
# If stat fails, fall through to full state computation
pass
# Compute current state only if we need to
seq_state = compute_state(seq_dir)
if not seq_state["files"]:
continue
# Check if state changed
if stored_state is not None and not state_changed(seq_state, stored_state):
# Target archive is up to date, but we might still need to clean up old .zip
if old_zip_path and old_zip_path.exists():
# Old .zip exists but we have a newer .7z, remove the old one
old_zip_path.unlink(missing_ok=True)
old_state_path = state_path_for(old_zip_path)
if old_state_path.exists():
@@ -436,6 +701,9 @@ def run_zip(worker_count: int, *, verbose: bool) -> int:
log("zip", "Archives already up to date; no sequences needed zipping.")
return 0
# Calculate RAM-aware worker count based on work items
worker_count, per_job_memory_limit = max_workers(requested_workers, work_items, verbose=verbose)
updated_paths: list[Path] = []
total = len(work_items)
@@ -443,7 +711,7 @@ def run_zip(worker_count: int, *, verbose: bool) -> int:
with ThreadPoolExecutor(max_workers=worker_count) as executor:
future_map = {
executor.submit(process_zip, seq_dir, zip_path, state_path, seq_state, verbose=verbose): seq_dir
executor.submit(process_zip, seq_dir, zip_path, state_path, seq_state, per_job_memory_limit, verbose=verbose): seq_dir
for seq_dir, zip_path, state_path, seq_state in work_items
}
@@ -556,13 +824,15 @@ def cleanup_orphan_archives(*, verbose: bool) -> int:
def main() -> int:
args = parse_args()
workers = max_workers(args.jobs)
if args.mode == "expand":
# For expand mode, use simple CPU-based worker calculation
workers, _ = max_workers(args.jobs, work_items=None, verbose=args.verbose)
run_expand(workers, verbose=args.verbose)
return 0
updated = run_zip(workers, verbose=args.verbose)
# For zip mode, work items will be calculated in run_zip
updated = run_zip(args.jobs, verbose=args.verbose)
return 0 if updated >= 0 else 1