analyser: introduce an option for disabling parallel mode

Fuzz Introspector is failed when building reports for Tarantool
and other projects, see [1]:

During handling of the above exception, another exception occurred:

  File "/usr/local/lib/python3.10/multiprocessing/managers.py", line 817, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File "/fuzz-introspector/src/fuzz_introspector/datatypes/fuzzer_profile.py", line 359, in accummulate_profile
    return_dict[uniq_id] = self
AttributeError: 'ForkAwareLocal' object has no attribute 'connection'

The issue (theoretically) is related to processing data in
parallel mode and cannot be reproduced locally. The patch
introduce an option for controlling parallel mode -
FUZZ_INTROSPECTOR_PARALLEL. When envronment variable is set to
`false` Fuzz Introspector will process data in sequential mode.

1. google/oss-fuzz#13226

Related to #2267
Related to google/oss-fuzz#13226

Author: ligurio

src/fuzz_introspector/analysis.py

@@ -88,23 +88,31 @@ def load_data_files(self,
 
         logger.info("[+] Accummulating profiles")
         logger.info("Accummulating using multiprocessing")
-        manager = multiprocessing.Manager()
-        semaphore = multiprocessing.Semaphore(10)
 
-        return_dict = manager.dict()
+        result_dict: Dict[Any, Any] = dict()
+        if parallelise:
+            manager = multiprocessing.Manager()
+            semaphore = multiprocessing.Semaphore(10)
 
-        jobs = []
-        idx = 0
-        for profile in self.profiles:
-            p = multiprocessing.Process(
-                target=fuzzer_profile.FuzzerProfile.accummulate_profile,
-                args=(profile, self.base_folder, return_dict, f"uniq-{idx}",
-                      semaphore))
-            jobs.append(p)
-            idx += 1
-            p.start()
-        for proc in jobs:
-            proc.join()
+            return_dict = manager.dict()
+
+            jobs = []
+            idx = 0
+            for profile in self.profiles:
+                p = multiprocessing.Process(
+                    target=fuzzer_profile.FuzzerProfile.accummulate_profile,
+                    args=(profile, self.base_folder, return_dict, f"uniq-{idx}",
+                          semaphore))
+                jobs.append(p)
+                idx += 1
+                p.start()
+            for proc in jobs:
+                proc.join()
+            result_dict = return_dict.copy()
+        else:
+            for profile in self.profiles:
+                fuzzer_profile.FuzzerProfile.accummulate_profile(
+                    profile, self.base_folder, result_dict, f"uniq-{idx}")
 
         new_profiles = []
         for idx in return_dict:

src/fuzz_introspector/commands.py

@@ -113,6 +113,7 @@ def analyse_end_to_end(arg_language,
         correlation_file = ''
 
     try:
+        is_parallel = os.getenv('FUZZ_INTROSPECTOR_PARALLEL', "true").lower() == "true"
         exit_code, return_values2 = run_analysis_on_dir(
             target_folder=out_dir,
             coverage_url=coverage_url,
@@ -124,6 +125,7 @@ def analyse_end_to_end(arg_language,
             out_dir=out_dir,
             dump_files=dump_files,
             harness_lists=harness_lists)
+            parallelise=is_parallel
         for k, v in return_values2.items():
             return_values[k] = v
     except DataLoaderError: