Add 10s per-file timeout to scan_once (prevents hung file blocking large scans)
This commit is contained in:
@@ -12,6 +12,8 @@ Performance: dhash-only rows (~6ms/file); phash only when check_near_dups.
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
import signal
|
||||||
|
import time as _time
|
||||||
|
|
||||||
from prefect import flow, task
|
from prefect import flow, task
|
||||||
from PIL import Image
|
from PIL import Image
|
||||||
@@ -49,15 +51,23 @@ def scan_once(base_dir: str, source: str, batch_size: int, check_near_dups: bool
|
|||||||
pending = []
|
pending = []
|
||||||
t0 = time.time()
|
t0 = time.time()
|
||||||
|
|
||||||
|
def _timeout_handler(signum, frame):
|
||||||
|
raise TimeoutError(f"file took >{FILE_TIMEOUT}s")
|
||||||
|
|
||||||
|
FILE_TIMEOUT = 10 # seconds per file — guards against hung/corrupt files
|
||||||
|
signal.signal(signal.SIGALRM, _timeout_handler)
|
||||||
|
|
||||||
for p_str in walk_files(base_dir):
|
for p_str in walk_files(base_dir):
|
||||||
p = Path(p_str)
|
p = Path(p_str)
|
||||||
try:
|
try:
|
||||||
|
signal.alarm(FILE_TIMEOUT)
|
||||||
known = conn.execute(
|
known = conn.execute(
|
||||||
"SELECT 1 FROM image_hashes WHERE path=? UNION SELECT 1 FROM known_paths WHERE path=?",
|
"SELECT 1 FROM image_hashes WHERE path=? UNION SELECT 1 FROM known_paths WHERE path=?",
|
||||||
(p_str, p_str),
|
(p_str, p_str),
|
||||||
).fetchone()
|
).fetchone()
|
||||||
if known:
|
if known:
|
||||||
totals["known"] += 1
|
totals["known"] += 1
|
||||||
|
signal.alarm(0)
|
||||||
continue
|
continue
|
||||||
|
|
||||||
sha = db.sha256_file(p_str)
|
sha = db.sha256_file(p_str)
|
||||||
@@ -68,6 +78,7 @@ def scan_once(base_dir: str, source: str, batch_size: int, check_near_dups: bool
|
|||||||
conn.execute(
|
conn.execute(
|
||||||
"INSERT OR IGNORE INTO known_paths (path, sha256) VALUES (?,?)", (p_str, sha))
|
"INSERT OR IGNORE INTO known_paths (path, sha256) VALUES (?,?)", (p_str, sha))
|
||||||
totals["exact_dups"] += 1
|
totals["exact_dups"] += 1
|
||||||
|
signal.alarm(0)
|
||||||
continue
|
continue
|
||||||
|
|
||||||
hx = db.hash_image(p)
|
hx = db.hash_image(p)
|
||||||
@@ -76,18 +87,24 @@ def scan_once(base_dir: str, source: str, batch_size: int, check_near_dups: bool
|
|||||||
pending.append((sha, hx["phash"], hx["dhash"], p.stat().st_size,
|
pending.append((sha, hx["phash"], hx["dhash"], p.stat().st_size,
|
||||||
w, h, str(p), source))
|
w, h, str(p), source))
|
||||||
totals["new"] += 1
|
totals["new"] += 1
|
||||||
|
signal.alarm(0)
|
||||||
|
|
||||||
if len(pending) >= batch_size:
|
if len(pending) >= batch_size:
|
||||||
conn.executemany(
|
conn.executemany(
|
||||||
"INSERT INTO image_hashes (sha256, phash, dhash, file_size, width, height, path, source) "
|
"INSERT INTO image_hashes (sha256, phash, dhash, file_size, width, height, path, source) "
|
||||||
"VALUES (?,?,?,?,?,?,?,?)", pending)
|
"VALUES (?,?,?,?,?,?,?,?)", pending)
|
||||||
conn.commit()
|
conn.commit()
|
||||||
print(f" checkpoint: {totals['new']} new ({time.time()-t0:.0f}s)", flush=True)
|
print(f" checkpoint: {totals['new']} new ({_time.time()-t0:.0f}s)", flush=True)
|
||||||
pending = []
|
pending = []
|
||||||
|
|
||||||
|
except TimeoutError as e:
|
||||||
|
totals["skipped"] += 1
|
||||||
|
print(f" TIMEOUT {p.name}: {e}", flush=True)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
totals["skipped"] += 1
|
totals["skipped"] += 1
|
||||||
print(f" SKIP {p.name}: {type(e).__name__}: {e}", flush=True)
|
print(f" SKIP {p.name}: {type(e).__name__}: {e}", flush=True)
|
||||||
|
finally:
|
||||||
|
signal.alarm(0)
|
||||||
|
|
||||||
if pending:
|
if pending:
|
||||||
conn.executemany(
|
conn.executemany(
|
||||||
|
|||||||
Reference in New Issue
Block a user