From dd273d53c65d46877b46132e7643e14bb3a509f0 Mon Sep 17 00:00:00 2001 From: Nubenetes Bot Date: Sun, 14 Jun 2026 12:12:39 +0200 Subject: [PATCH] perf(automation): massive parallelization for Health, Metadata and AI engines --- src/gemini_utils.py | 2 +- src/v2_metadata.py | 19 +++++++++++++++---- src/v2_optimizer.py | 10 ++++++---- 3 files changed, 22 insertions(+), 9 deletions(-) diff --git a/src/gemini_utils.py b/src/gemini_utils.py index 85deadcf..230b1545 100644 --- a/src/gemini_utils.py +++ b/src/gemini_utils.py @@ -16,7 +16,7 @@ CURRENT_KEY_INDEX = 0 DISCOVERED_MODELS = [] GLOBAL_COOLDOWN_UNTIL = 0 THROTTLED_MODELS = {} # {model_name: timestamp} -GLOBAL_AI_SEMAPHORE = asyncio.Semaphore(5) # Max 5 concurrent calls globally +GLOBAL_AI_SEMAPHORE = asyncio.Semaphore(15) # Increased to 15 for high-tier processing class GeminiSessionTracker: def __init__(self): diff --git a/src/v2_metadata.py b/src/v2_metadata.py index 1b83beb6..0d09c08b 100644 --- a/src/v2_metadata.py +++ b/src/v2_metadata.py @@ -15,6 +15,13 @@ async def run_metadata_enrichment(): processed_gh_metadata = set() gh_fetch_count = 0 force_full = os.getenv("ENRICH_METADATA", "false").lower() == "true" + + gh_tasks = [] + gh_sem = asyncio.Semaphore(20) # Increased concurrency for final sprint + + async def _fetch_gh_with_sem(url: str): + async with gh_sem: + return url, await get_github_activity(url) for l in all_v1_links: norm_url = normalize_url(l["url"]) @@ -23,14 +30,18 @@ async def run_metadata_enrichment(): cached = engine.inventory.get(norm_url, {}) # Enrich if missing or if forced if (force_full or not cached.get("gh_stars")) and norm_url not in processed_gh_metadata: - log_event(f" [METADATA] Fetching GH Activity for {norm_url}") processed_gh_metadata.add(norm_url) - gh_data = await get_github_activity(norm_url) + gh_tasks.append(_fetch_gh_with_sem(norm_url)) + + if gh_tasks: + log_event(f" [METADATA] Pulse: Fetching {len(gh_tasks)} GitHub profiles in parallel...") + gh_results = await asyncio.gather(*gh_tasks) + for norm_url, gh_data in gh_results: if gh_data: if norm_url not in engine.inventory: engine.inventory[norm_url] = {} engine.inventory[norm_url].update(gh_data) - - gh_fetch_count += 1 + gh_fetch_count += 1 + if gh_fetch_count % 100 == 0: engine._save_inventory() log_event(f" [💾] Periodic Save: {gh_fetch_count} fetches...") diff --git a/src/v2_optimizer.py b/src/v2_optimizer.py index c6519c25..9d3f56a9 100644 --- a/src/v2_optimizer.py +++ b/src/v2_optimizer.py @@ -217,14 +217,16 @@ class V2VisionEngine: online_links = list(fast_online) total_needs = len(needs_check) async with httpx.AsyncClient(timeout=15.0, follow_redirects=True, verify=False) as client: - for i in range(0, total_needs, 50): - batch = needs_check[i:i+50] + # Mandate 16/22: Resilient asynchronous health checks with increased concurrency + CHUNK_SIZE = 100 + for i in range(0, total_needs, CHUNK_SIZE): + batch = needs_check[i:i+CHUNK_SIZE] tasks = [self._check_single_link_resilient(client, l) for l in batch] results = await asyncio.gather(*tasks) online_links.extend([r for r in results if r is not None]) if i % 100 == 0: log_event(f" [>] Progress: [{i}/{total_needs}] links validated over network...") - await asyncio.sleep(0.1) + await asyncio.sleep(0.05) # Minimal delay return online_links async def _check_single_link_resilient(self, client, link: Dict): @@ -274,7 +276,7 @@ class V2VisionEngine: processed_gh_metadata = set() gh_fetch_count = 0 gh_tasks = [] - gh_sem = asyncio.Semaphore(15) # Up to 15 concurrent fetches for GitHub API stability + gh_sem = asyncio.Semaphore(30) # Increased for final sprint strategy async def _fetch_gh_with_sem(url: str): async with gh_sem: