From 92667185d958efb723d6af1ea469b8d1e19552a3 Mon Sep 17 00:00:00 2001 From: "Simms, Andrew" Date: Sun, 8 Feb 2026 21:50:28 -0700 Subject: [PATCH 1/5] Linux: Use venv python from launcher --- modaq-upload.desktop.template | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/modaq-upload.desktop.template b/modaq-upload.desktop.template index a0da4ee..d496fdd 100644 --- a/modaq-upload.desktop.template +++ b/modaq-upload.desktop.template @@ -3,7 +3,7 @@ Version=0.2.1 Type=Application Name=MODAQ Upload Comment=MODAQ File Uploader -Exec=python3 {{PROJECT_DIR}}/launch.py +Exec={{PROJECT_DIR}}/venv/bin/python {{PROJECT_DIR}}/launch.py Icon={{PROJECT_DIR}}/app/static/images/modaq-logo.png Terminal=true Categories=Science;Utility; From 7454769eb3b2981c2d0ed333d88fc5306a08f3fa Mon Sep 17 00:00:00 2001 From: "Simms, Andrew" Date: Sun, 8 Feb 2026 21:51:33 -0700 Subject: [PATCH 2/5] Linux: Modify launcher to open browser if process is still running --- launch.py | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/launch.py b/launch.py index 8240155..a1b278f 100644 --- a/launch.py +++ b/launch.py @@ -39,9 +39,15 @@ def open_browser(url: str) -> None: import shutil if shutil.which("firefox"): - subprocess.Popen(["firefox", url], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + subprocess.Popen( + ["firefox", url], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True, + ) elif shutil.which("xdg-open"): - subprocess.Popen(["xdg-open", url], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + subprocess.Popen( + ["xdg-open", url], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True, + ) else: log(f"Could not find a browser. Open {url} manually.") @@ -98,6 +104,7 @@ def main() -> None: if port_in_use(PORT): log(f"Port {PORT} is already in use — opening browser to existing instance.") open_browser(BROWSER_URL) + time.sleep(2) sys.exit(0) # ── Register signal handlers ───────────────────────────── From 0b37cbb3ebab3037ea65a60835eeb18c185d9302 Mon Sep 17 00:00:00 2001 From: "Simms, Andrew" Date: Sun, 8 Feb 2026 21:52:21 -0700 Subject: [PATCH 3/5] Fix: Use 1 gunicorn worker with threads on linux This fixes a bug where the SSE jobs were working, but not attaching correctly to the frontend. The fix is to only use 1 gunicorn worker, with multiple threads --- deploy/modaq-upload.service | 2 +- launch.py | 2 ++ 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/deploy/modaq-upload.service b/deploy/modaq-upload.service index 7234597..cb1a8de 100644 --- a/deploy/modaq-upload.service +++ b/deploy/modaq-upload.service @@ -10,7 +10,7 @@ WorkingDirectory=/opt/modaq-upload Environment="PATH=/opt/modaq-upload/venv/bin" ExecStart=/opt/modaq-upload/venv/bin/gunicorn \ --bind 0.0.0.0:8080 \ - --workers 2 \ + --workers 1 \ --threads 4 \ --timeout 300 \ --access-logfile /var/log/modaq-upload/access.log \ diff --git a/launch.py b/launch.py index a1b278f..26b532d 100644 --- a/launch.py +++ b/launch.py @@ -120,6 +120,8 @@ def main() -> None: "--bind", f"127.0.0.1:{PORT}", "--workers", + "1", + "--threads", "4", "--timeout", "300", From 6ea8d9e82ba170f8a1cbf352a2d06732c2bb8cbe Mon Sep 17 00:00:00 2001 From: "Simms, Andrew" Date: Sun, 8 Feb 2026 21:57:21 -0700 Subject: [PATCH 4/5] Feat: Use ProcessPool for mcap threaded validation MCAP processing benefits from multiple processes running an once. The GIL makes this tricky with gunicorn. The solution is to use both ProcessPool and ThreadPool. ProcessPool can scale across cpus for MCAP validation, and ThreadPool can efficientely handle io for uploads. --- app/services/upload_manager.py | 123 ++++++++++++++++++++++++++++----- 1 file changed, 107 insertions(+), 16 deletions(-) diff --git a/app/services/upload_manager.py b/app/services/upload_manager.py index 0f7ffea..15aa369 100644 --- a/app/services/upload_manager.py +++ b/app/services/upload_manager.py @@ -1,11 +1,12 @@ """Upload manager for orchestrating file uploads to S3.""" import logging +import os import shutil import threading import uuid from collections.abc import Callable -from concurrent.futures import ThreadPoolExecutor, as_completed +from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, as_completed from dataclasses import dataclass, field from datetime import UTC, datetime from enum import Enum @@ -23,6 +24,18 @@ EPOCH_CUTOFF = datetime(1980, 1, 1, tzinfo=UTC) +def _extract_start_time_worker(local_path: str) -> datetime | str: + """Worker function for ProcessPoolExecutor — must be top-level for pickling. + + Returns: + datetime on success, or error message string on failure. + """ + try: + return mcap_service.extract_start_time(local_path) + except Exception as e: + return str(e) + + class UploadStatus(Enum): """Status of a file upload.""" @@ -348,6 +361,39 @@ def analyze_job( return job + def _check_duplicate( + self, + file_state: FileUploadState, + s3_client: Any, + s3_bucket: str, + use_cache: bool = True, + ) -> None: + """Check if a file already exists in S3 (I/O-bound, safe for threads).""" + s3_path = file_state.s3_path + if not s3_path: + return + + cache_result: bool | None = None + if use_cache: + cache = get_cache_service() + cache_result = cache.check_exists_cached(s3_bucket, s3_path) + + if cache_result is not None: + file_state.is_duplicate = cache_result + else: + file_state.is_duplicate = s3_service.check_file_exists( + s3_client, s3_bucket, s3_path + ) + if use_cache: + cache = get_cache_service() + cache.update_cache( + s3_bucket, + s3_path, + file_state.is_duplicate, + file_state.filename, + file_state.file_size, + ) + def _analyze_single_file( self, file_state: FileUploadState, @@ -497,34 +543,79 @@ def analyze_job_async( progress_callback(job, file_state) return job - # Analyze files in parallel - with ThreadPoolExecutor(max_workers=self.max_workers) as executor: - futures = {} - for file_state in job.files: - future = executor.submit( - self._analyze_single_file, + # Phase 1: MCAP parsing (CPU-bound) — use ProcessPoolExecutor for true + # parallelism across cores, bypassing the GIL. + cpu_workers = max(1, (os.cpu_count() or 4) - 1) + for file_state in job.files: + file_state.status = UploadStatus.ANALYZING + if progress_callback: + progress_callback(job, file_state) + + with ProcessPoolExecutor(max_workers=cpu_workers) as proc_executor: + parse_futures = { + proc_executor.submit( + _extract_start_time_worker, file_state.local_path + ): file_state + for file_state in job.files + } + for future in as_completed(parse_futures): + file_state = parse_futures[future] + result = future.result() + if isinstance(result, str): + # Error message returned from worker + file_state.status = UploadStatus.FAILED + file_state.error_message = result + log.error( + "analysis", + "file_analysis_failed", + f"Failed to analyze {file_state.filename}: {result}", + {"job_id": job_id, "filename": file_state.filename, "error": result}, + ) + else: + file_state.start_time = result + naive_start = mcap_service.to_naive_utc(result) + file_state.is_valid = naive_start >= EPOCH_CUTOFF.replace(tzinfo=None) + file_state.s3_path = mcap_service.generate_s3_path(result, file_state.filename) + if progress_callback: + progress_callback(job, file_state) + + # Phase 2: S3 duplicate checks (I/O-bound) — threads are fine here. + parsed_files = [f for f in job.files if f.status != UploadStatus.FAILED] + with ThreadPoolExecutor(max_workers=self.max_workers) as io_executor: + dup_futures: dict[Any, FileUploadState] = {} + for file_state in parsed_files: + fut = io_executor.submit( + self._check_duplicate, file_state, s3_client, s3_bucket, use_cache, - job_id, - progress_callback, - job, ) - futures[future] = file_state + dup_futures[fut] = file_state - # Process results as they complete - for future in as_completed(futures): - file_state = futures[future] + for fut in as_completed(dup_futures): + file_state = dup_futures[fut] try: - # Result is already updated in-place, but get it to handle exceptions future.result() + file_state.status = UploadStatus.READY + log.info( + "analysis", + "file_analysis_completed", + f"Analyzed {file_state.filename}", + { + "job_id": job_id, + "filename": file_state.filename, + "file_size": file_state.file_size, + "s3_path": file_state.s3_path, + "is_duplicate": file_state.is_duplicate, + "is_valid": file_state.is_valid, + }, + ) except Exception as e: with job.lock: file_state.status = UploadStatus.FAILED file_state.error_message = str(e) - # Call progress callback if progress_callback: progress_callback(job, file_state) From fd8eedefe69ee519de7e2df480e403b459c13470 Mon Sep 17 00:00:00 2001 From: "Simms, Andrew" Date: Sun, 8 Feb 2026 21:59:10 -0700 Subject: [PATCH 5/5] Release: v0.2.2 --- modaq-upload.desktop.template | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/modaq-upload.desktop.template b/modaq-upload.desktop.template index d496fdd..6294a0d 100644 --- a/modaq-upload.desktop.template +++ b/modaq-upload.desktop.template @@ -1,5 +1,5 @@ [Desktop Entry] -Version=0.2.1 +Version=0.2.2 Type=Application Name=MODAQ Upload Comment=MODAQ File Uploader