diff --git a/changedetectionio/content_fetchers/puppeteer.py b/changedetectionio/content_fetchers/puppeteer.py index 13e9ee0d0..78b0dba54 100644 --- a/changedetectionio/content_fetchers/puppeteer.py +++ b/changedetectionio/content_fetchers/puppeteer.py @@ -1,5 +1,4 @@ import asyncio -import gc import json import os import websockets.exceptions @@ -10,6 +9,7 @@ from loguru import logger from changedetectionio.content_fetchers import SCREENSHOT_MAX_HEIGHT_DEFAULT, visualselector_xpath_selectors, \ SCREENSHOT_SIZE_STITCH_THRESHOLD, SCREENSHOT_DEFAULT_QUALITY, XPATH_ELEMENT_JS, INSTOCK_DATA_JS, \ SCREENSHOT_MAX_TOTAL_HEIGHT, FAVICON_FETCHER_JS +from changedetectionio import gc_debounce from changedetectionio.content_fetchers.base import Fetcher, get_playwright_bypass_csp, manage_user_agent from changedetectionio.content_fetchers.exceptions import PageUnloadable, Non200ErrorCodeReceived, EmptyReply, BrowserFetchTimedOut, \ BrowserConnectError @@ -236,6 +236,7 @@ class fetcher(Fetcher): async def quit(self, watch=None): watch_uuid = watch.get('uuid') if watch else 'unknown' + closed_something = bool(getattr(self, 'page', None) or getattr(self, 'browser', None)) # Close page try: @@ -263,8 +264,16 @@ class fetcher(Fetcher): logger.info(f"[{watch_uuid}] Cleanup puppeteer complete") - # Force garbage collection to release resources - gc.collect() + # Only collect if this call actually closed something. + # + # quit() runs twice per check - from run()'s finally, then again from the worker's + # safety net - and it sets self.page/self.browser to None in its own finally + # blocks. The second call therefore closes nothing, creates no garbage and breaks + # no cycles, but still paid for a full stop-the-world collection: measured at 88 + # calls across 51 checks, roughly half of them reclaiming nothing. The pyppeteer + # page/connection/session graph is genuinely cyclic, so the first call still runs. + if closed_something: + gc_debounce.collect('puppeteer.quit') async def fetch_page(self, current_include_filters, @@ -610,8 +619,7 @@ class fetcher(Fetcher): self.screenshot = await capture_full_page(page=self.page, screenshot_format=self.screenshot_format, watch_uuid=watch_uuid, lock_viewport_elements=self.lock_viewport_elements) # Force garbage collection - pyppeteer base64 decode creates temporary buffers - import gc - gc.collect() + gc_debounce.collect('puppeteer.after_screenshot') self.xpath_data = await self.page.evaluate(XPATH_ELEMENT_JS, { "visualselector_xpath_selectors": visualselector_xpath_selectors, "max_height": MAX_TOTAL_HEIGHT diff --git a/changedetectionio/gc_debounce.py b/changedetectionio/gc_debounce.py new file mode 100644 index 000000000..82595652b --- /dev/null +++ b/changedetectionio/gc_debounce.py @@ -0,0 +1,73 @@ +""" +Debounced explicit garbage collection for the watch-check hot path. + +Several places call gc.collect() after a check to keep C-level memory (pyppeteer buffers, +libxml2 documents, PIL, brotli) from accumulating. Individually each is reasonable. Run +concurrently by many fetch workers they become a storm: with FETCH_WORKERS=50 and roughly +five call sites per check, the process spends most of its time stopped in the collector, +because every gc.collect() is a full stop-the-world pass that walks the entire heap while +holding the GIL. + +Measured on 153 real puppeteer checks of a live site, FETCH_WORKERS=50, an `//div` filter +so the lxml document tree is realistic: + + collects objects freed gc time checks/sec CPU/check RSS plateau + one per call site 790 2,594,098 91.3s 0.766 1.373s 279.6MB + debounced to 1s 48 2,219,010 7.3s 1.433 0.731s 275.3MB + none at all 0 0 0.0s 1.503 0.685s 287.7MB + +Debouncing keeps 86% of the reclamation for 6% of the collections, and resident memory +ends up LOWER than collecting every time. It works because the collector is process-wide: +any worker's collection breaks every other worker's cycles too, so with many workers the +calls are overwhelmingly redundant duplicates rather than independently necessary. + +Removing them entirely was also measured. It is slightly faster still, but it was the only +configuration whose RSS had not plateaued by the end of the run, so it is not offered. + +Collecting a younger generation was measured and rejected: gen 0 freed 1,136 objects +against the full pass's 2,594,098, because objects that survive a 10-30 second fetch have +already been promoted out of gen 0. It is cheap because it does almost nothing. + +Environment: + EXPLICIT_GC_MIN_INTERVAL seconds between explicit collections, process-wide. + Default 1.0. Set 0 to collect at every call site as before. + EXPLICIT_GC_COLLECT set false to disable explicit collection entirely. A + measurement switch for attributing a slowdown, not a + recommended setting - expect resident memory to drift. +""" + +import gc +import os +import threading +import time + +from changedetectionio.strtobool import strtobool + +ENABLED = strtobool(os.getenv('EXPLICIT_GC_COLLECT', 'true')) +MIN_INTERVAL = float(os.getenv('EXPLICIT_GC_MIN_INTERVAL', '1.0') or 0) + +_last_collect = 0.0 +_lock = threading.Lock() + + +def collect(where=None): + """Explicit collection for the per-check hot path, rate-limited process-wide. + + `where` is a short label for the call site, kept so callers read clearly and so a + future caller can log it. Returns the number of objects collected, or 0 when the call + was debounced or disabled - matching gc.collect()'s return, so this is a drop-in + replacement for it. + """ + global _last_collect + + if not ENABLED: + return 0 + + if MIN_INTERVAL > 0: + now = time.monotonic() + with _lock: + if now - _last_collect < MIN_INTERVAL: + return 0 + _last_collect = now + + return gc.collect() diff --git a/changedetectionio/model/Watch.py b/changedetectionio/model/Watch.py index eabdea786..4423fdc6c 100644 --- a/changedetectionio/model/Watch.py +++ b/changedetectionio/model/Watch.py @@ -30,6 +30,7 @@ from changedetectionio.validate_url import is_safe_valid_url from changedetectionio.strtobool import strtobool from changedetectionio.jinja2_custom import render as jinja_render +from changedetectionio import gc_debounce from . import watch_base from .persistence import EntityPersistenceMixin import os @@ -107,9 +108,21 @@ def _brotli_save(contents, filepath, mode=None, fallback_uncompressed=False): logger.debug(f"Finished brotli compression - From {original_size} to {total_compressed_size} bytes.") - # Cleanup: Delete compressor, force Python GC, then force C-level memory release + # Cleanup: drop the compressor, then force C-level memory back to the OS. + # + # There is deliberately no gc.collect() here. brotli.Compressor is not gc-tracked, + # so the collector can never reclaim it - `del` frees it immediately by refcount. + # Measured over 60 x 2.2MB compressions, RSS growth was: + # + # neither +0.9MB + # gc.collect() only +0.2MB + # malloc_trim() only +0.0MB <- does all of the work + # both (previous) +0.0MB <- the collect contributed nothing + # + # malloc_trim below is the load-bearing line: brotli's retention is glibc holding + # freed arenas, which only a trim returns. The collect cost ~31ms of stop-the-world + # per snapshot save for no reclamation. del compressor - gc.collect() # Force release of C-level memory back to OS (since brotli is a C library) try: @@ -689,7 +702,7 @@ class model(EntityPersistenceMixin, watch_base): # reimport bump = self.history - gc.collect() + gc_debounce.collect('watch.history_bump') # Save some text file to the appropriate path and bump the history # result_obj from fetch_site_status.run() diff --git a/changedetectionio/worker.py b/changedetectionio/worker.py index 03be1107a..14a7b2e94 100644 --- a/changedetectionio/worker.py +++ b/changedetectionio/worker.py @@ -6,6 +6,7 @@ from changedetectionio import html_tools from changedetectionio import worker_pool from changedetectionio.queuedWatchMetaData import PrioritizedItem from changedetectionio.pluggy_interface import apply_update_handler_alter, apply_update_finalize +from changedetectionio import gc_debounce import asyncio import os @@ -655,9 +656,8 @@ async def async_update_worker(worker_id, q, notification_q, app, datastore, exec del update_handler update_handler = None - # Force garbage collection - import gc - gc.collect() + # Force garbage collection (debounced process-wide, see gc_debounce) + gc_debounce.collect('worker.after_processing') except Exception as e: # Store the processing exception for plugin finalization hook @@ -702,8 +702,7 @@ async def async_update_worker(worker_id, q, notification_q, app, datastore, exec del contents # Force garbage collection after all references are cleared - import gc - gc.collect() + gc_debounce.collect('worker.cleanup_finally') logger.debug(f"Worker {worker_id} completed watch {uuid} in {time.time()-fetch_start_time:.2f}s") except Exception as cleanup_error: