Raise the minimum RAM required per CPU in the docs #1191 (#1192)

Signed-off-by: tdruez <tdruez@nexb.com>

Author: tdruez

CHANGELOG.rst

@@ -31,6 +31,15 @@ v34.5.0 (unreleased)
 - Add "Product name" and "Product version" as new project settings.
   https://github.com/nexB/scancode.io/issues/1197
 
+- Add "Product name" and "Product version" as new project settings.
+  https://github.com/nexB/scancode.io/issues/1197
+
+- Raise the minimum RAM required per CPU code in the docs.
+  A good rule of thumb is to allow **2 GB of memory per CPU**.
+  For example, if Docker is configured for 8 CPUs, a minimum of 16 GB of memory is
+  required.
+  https://github.com/nexB/scancode.io/issues/1191
+
 v34.4.0 (2024-04-22)
 --------------------
 

docs/installation.rst

@@ -74,8 +74,8 @@ An overview of the web application usage is available at :ref:`user_interface`.
 
     **Make sure to allow enough memory to support each CPU processes**.
 
-    A good rule of thumb is to allow **1 GB of memory per CPU**.
-    For example, if Docker is configured for 8 CPUs, a minimum of 8 GB of memory is
+    A good rule of thumb is to allow **2 GB of memory per CPU**.
+    For example, if Docker is configured for 8 CPUs, a minimum of 16 GB of memory is
     required.
 
 .. tip::

scanpipe/pipelines/__init__.py

@@ -168,6 +168,19 @@ def log(self, message):
         logger.info(message)
         self.run.append_to_log(message)
 
+    @staticmethod
+    def output_from_exception(exception):
+        """Return a formatted error message including the traceback."""
+        output = f"{exception}\n\n"
+
+        if exception.__cause__ and str(exception.__cause__) != str(exception):
+            output += f"Cause: {exception.__cause__}\n\n"
+
+        traceback_formatted = "".join(traceback.format_tb(exception.__traceback__))
+        output += f"Traceback:\n{traceback_formatted}"
+
+        return output
+
     def execute(self):
         """Execute each steps in the order defined on this pipeline class."""
         self.log(f"Pipeline [{self.pipeline_name}] starting")
@@ -189,10 +202,9 @@ def execute(self):
 
             try:
                 step(self)
-            except Exception as e:
+            except Exception as exception:
                 self.log("Pipeline failed")
-                tb = "".join(traceback.format_tb(e.__traceback__))
-                return 1, f"{e}\n\nTraceback:\n{tb}"
+                return 1, self.output_from_exception(exception)
 
             step_run_time = timer() - step_start_time
             self.log(f"Step [{step_name}] completed in {humanize_time(step_run_time)}")

scanpipe/pipes/scancode.py

@@ -20,13 +20,14 @@
 # ScanCode.io is a free software code scanning tool from nexB Inc. and others.
 # Visit https://github.com/nexB/scancode.io for support and download.
 
-import concurrent.futures
 import json
 import logging
 import multiprocessing
 import os
 import shlex
+import warnings
 from collections import defaultdict
+from concurrent import futures
 from functools import partial
 from pathlib import Path
 
@@ -57,6 +58,10 @@
 scanpipe_app = apps.get_app_config("scanpipe")
 
 
+class InsufficientResourcesError(Exception):
+    pass
+
+
 def get_max_workers(keep_available):
     """
     Return the `SCANCODEIO_PROCESSES` if defined in the setting,
@@ -67,16 +72,28 @@ def get_max_workers(keep_available):
     but for example "spawn", such as on macOS, multiprocessing and threading are
     disabled by default returning -1 `max_workers`.
     """
-    processes = settings.SCANCODEIO_PROCESSES
-    if processes is not None:
-        return processes
+    processes_from_settings = settings.SCANCODEIO_PROCESSES
+    if processes_from_settings in [-1, 0, 1]:
+        return processes_from_settings
 
     if multiprocessing.get_start_method() != "fork":
         return -1
 
     max_workers = os.cpu_count() - keep_available
     if max_workers < 1:
         return 1
+
+    if processes_from_settings is not None:
+        if processes_from_settings <= max_workers:
+            return processes_from_settings
+        else:
+            msg = (
+                f"The value {processes_from_settings} specified in SCANCODEIO_PROCESSES"
+                f" exceeds the number of available CPUs on this machine."
+                f" {max_workers} CPUs will be used instead for multiprocessing."
+            )
+            warnings.warn(msg, ResourceWarning)
+
     return max_workers
 
 
@@ -305,20 +322,28 @@ def scan_resources(
 
     logger.info(f"Starting ProcessPoolExecutor with {max_workers} max_workers")
 
-    with concurrent.futures.ProcessPoolExecutor(max_workers) as executor:
+    with futures.ProcessPoolExecutor(max_workers) as executor:
         future_to_resource = {
             executor.submit(scan_func, resource.location): resource
             for resource in resource_iterator
         }
 
         # Iterate over the Futures as they complete (finished or cancelled)
-        future_as_completed = concurrent.futures.as_completed(future_to_resource)
+        future_as_completed = futures.as_completed(future_to_resource)
 
         for future in progress.iter(future_as_completed):
             resource = future_to_resource[future]
             progress.log_progress()
             logger.debug(f"{scan_func.__name__} pk={resource.pk}")
-            scan_results, scan_errors = future.result()
+            try:
+                scan_results, scan_errors = future.result()
+            except futures.process.BrokenProcessPool as broken_pool_error:
+                message = (
+                    "You may not have enough resources to complete this operation. "
+                    "Please ensure that there is at least 2 GB of available memory per "
+                    "CPU core for successful execution."
+                )
+                raise broken_pool_error from InsufficientResourcesError(message)
             save_func(resource, scan_results, scan_errors)
 
 

scanpipe/tests/pipes/test_scancode.py

@@ -21,6 +21,7 @@
 # Visit https://github.com/nexB/scancode.io for support and download.
 
 import json
+import multiprocessing
 import os
 import sys
 import tempfile
@@ -434,10 +435,11 @@ def test_scanpipe_pipes_scancode_run_scan_args(self, mock_run_scan):
             run_scan_kwargs = mock_run_scan.call_args.kwargs
             self.assertEqual(10, run_scan_kwargs.get("timeout"))
 
-        with override_settings(SCANCODEIO_PROCESSES=10):
+        expected_processes = -1 if multiprocessing.get_start_method() != "fork" else 2
+        with override_settings(SCANCODEIO_PROCESSES=2):
             scancode.run_scan(location=None, output_file=output_file, run_scan_args={})
             run_scan_kwargs = mock_run_scan.call_args.kwargs
-            self.assertEqual(10, run_scan_kwargs.get("processes"))
+            self.assertEqual(expected_processes, run_scan_kwargs.get("processes"))
 
     def test_scanpipe_pipes_scancode_make_results_summary(self, regen=FIXTURES_REGEN):
         # Ensure the policies index is empty to avoid any side effect on results