[feat] Make sync store more sync

After the introduction of async store, massStoreRun was implemented in
a way that it is polling the task ID in every 5 seconds. This makes the
text execution too long.

In this commit a pipe has been added to the abstract task so it can
directly notify the massStoreRun function when the task is ready. No
other API function should use this pipe.

Author: bruntib

codechecker_common/compatibility/multiprocessing.py

@@ -14,14 +14,17 @@
 # pylint: disable=unused-import
 if sys.platform in ["darwin", "win32"]:
     from multiprocess import \
-        Pool, Process, \
+        Pipe, Pool, Process, \
         Queue, \
         Value, \
         cpu_count
+    from multiprocess.managers import SyncManager
 else:
     from concurrent.futures import ProcessPoolExecutor as Pool
     from multiprocessing import \
+        Pipe, \
         Process, \
         Queue, \
         Value, \
         cpu_count
+    from multiprocessing.managers import SyncManager

web/client/codechecker_client/cli/store.py

@@ -58,7 +58,7 @@ def assemble_blame_info(_, __) -> int:
 from codechecker_common.compatibility.multiprocessing import Pool
 from codechecker_common.source_code_comment_handler import \
     SourceCodeCommentHandler
-from codechecker_common.util import format_size, load_json
+from codechecker_common.util import format_size, load_json, strtobool
 
 from codechecker_web.shared import webserver_context, host_check
 from codechecker_web.shared.env import get_default_workspace
@@ -964,51 +964,111 @@ def main(args):
         description = args.description if 'description' in args else None
 
         LOG.info("Storing results to the server ...")
-        task_token: str = client.massStoreRunAsynchronous(
-            b64zip,
-            SubmittedRunOptions(
-                runName=args.name,
-                tag=args.tag if "tag" in args else None,
-                version=str(context.version),
-                force="force" in args,
-                trimPathPrefixes=trim_path_prefixes,
-                description=description)
-            )
-        LOG.info("Reports submitted to the server for processing.")
-
-        if client.allowsStoringAnalysisStatistics():
-            store_analysis_statistics(client, args.input, args.name)
-
-        if "detach" in args:
-            LOG.warning("Exiting the 'store' subcommand as '--detach' was "
-                        "specified: not waiting for the result of the store "
-                        "operation.\n"
-                        "The server might not have finished processing "
-                        "everything at this point, so do NOT rely on querying "
-                        "the results just yet!\n"
-                        "To await the completion of the processing later, "
-                        "you can execute:\n\n"
-                        "\tCodeChecker cmd serverside-tasks --token %s "
-                        "--await",
-                        task_token)
-            # Print the token to stdout as well, so scripts can use "--detach"
-            # meaningfully.
-            print(task_token)
-            return
-
-        task_client = libclient.setup_task_client(protocol, host, port)
-        task_status: str = await_task_termination(LOG, task_token,
-                                                  task_api_client=task_client)
-
-        if task_status == "COMPLETED":
-            LOG.info("Storing the reports finished successfully.")
+
+        if strtobool(os.environ.get('CC_FORCE_SYNC_STORE', 'no')):
+            try:
+                with _timeout_watchdog(timedelta(hours=1),
+                                       signal.SIGUSR1):
+                    client.massStoreRun(args.name,
+                                        args.tag if 'tag' in args else None,
+                                        str(context.version),
+                                        b64zip,
+                                        'force' in args,
+                                        trim_path_prefixes,
+                                        description)
+            except WatchdogError as we:
+                LOG.warning("%s", str(we))
+
+                # Showing parts of the exception stack is important here.
+                # We **WANT** to see that the timeout happened during a wait on
+                # Thrift reading from the TCP connection (something deep in the
+                # Python library code at "sock.recv_into").
+                import traceback
+                _, _, tb = sys.exc_info()
+                frames = traceback.extract_tb(tb)
+                first, last = frames[0], frames[-2]
+                formatted_frames = traceback.format_list([first, last])
+                fmt_first, fmt_last = formatted_frames[0], formatted_frames[1]
+                LOG.info("Timeout was triggered during:\n%s", fmt_first)
+                LOG.info("Timeout interrupted this low-level operation:\n%s",
+                         fmt_last)
+
+                LOG.error(
+                    "Timeout!"
+                    "\n\tThe server's reply did not arrive after "
+                    "%d seconds (%s) elapsed since the server-side "
+                    "processing began."
+                    "\n\n\tThis does *NOT* mean that there was an issue "
+                    "with the run you were storing!"
+                    "\n\tThe server might still be processing the results..."
+                    "\n\tHowever, it is more likely that the "
+                    "server had already finished, but the client did not "
+                    "receive a response."
+                    "\n\tUsually, this is caused by the underlying TCP "
+                    "connection failing to signal a low-level disconnect."
+                    "\n\tClients potentially hanging indefinitely in these "
+                    "scenarios is an unfortunate and known issue."
+                    "\n\t\tSee http://github.com/Ericsson/codechecker/"
+                    "issues/3672 for details!"
+                    "\n\n\tThis error here is a temporary measure to ensure "
+                    "an infinite hang is replaced with a well-explained "
+                    "timeout."
+                    "\n\tA more proper solution will be implemented in a "
+                    "subsequent version of CodeChecker.",
+                    we.timeout.total_seconds(), str(we.timeout))
+                sys.exit(1)
+
+            if client.allowsStoringAnalysisStatistics():
+                store_analysis_statistics(client, args.input, args.name)
         else:
-            LOG.error("Storing the reports failed! "
-                      "The job terminated in status '%s'. "
-                      "The comments associated with the failure are:\n\n%s",
-                      task_status,
-                      task_client.getTaskInfo(task_token).comments)
-            sys.exit(1)
+            task_token: str = client.massStoreRunAsynchronous(
+                b64zip,
+                SubmittedRunOptions(
+                    runName=args.name,
+                    tag=args.tag if "tag" in args else None,
+                    version=str(context.version),
+                    force="force" in args,
+                    trimPathPrefixes=trim_path_prefixes,
+                    description=description)
+                )
+            LOG.info("Reports submitted to the server for processing.")
+
+            if client.allowsStoringAnalysisStatistics():
+                store_analysis_statistics(client, args.input, args.name)
+
+            if "detach" in args:
+                LOG.warning(
+                    "Exiting the 'store' subcommand as '--detach' was "
+                    "specified: not waiting for the result of the store "
+                    "operation.\n"
+                    "The server might not have finished processing "
+                    "everything at this point, so do NOT rely on querying "
+                    "the results just yet!\n"
+                    "To await the completion of the processing later, "
+                    "you can execute:\n\n"
+                    "\tCodeChecker cmd serverside-tasks --token %s "
+                    "--await",
+                    task_token)
+                # Print the token to stdout as well, so scripts can use
+                # "--detach" meaningfully.
+                print(task_token)
+                return
+
+            task_client = libclient.setup_task_client(protocol, host, port)
+            task_status: str = await_task_termination(
+                LOG, task_token, task_api_client=task_client)
+
+            if task_status == "COMPLETED":
+                LOG.info("Storing the reports finished successfully.")
+            else:
+                LOG.error(
+                    "Storing the reports failed! "
+                    "The job terminated in status '%s'. "
+                    "The comments associated with the failure are:\n\n%s",
+                    task_status,
+                    task_client.getTaskInfo(task_token).comments)
+                sys.exit(1)
+
     except Exception as ex:
         import traceback
         traceback.print_exc()

web/server/codechecker_server/api/mass_store_run.py

@@ -1209,8 +1209,8 @@ def __realise_fake_checkers(self, session):
             checker: Tuple[str, str] = checker_name_for_report(report)
             grouped_by_checker[checker].append(cast(int, db_id))
 
-        for chk, report_ids in grouped_by_checker.items():
-            analyzer_name, checker_name = chk
+        for checker, report_ids in grouped_by_checker.items():
+            analyzer_name, checker_name = checker
             chk_obj = cast(Checker, self.__get_checker(session,
                                                        analyzer_name,
                                                        checker_name))
@@ -1222,18 +1222,18 @@ def __realise_fake_checkers(self, session):
     def __add_report_context(self, session, file_path_to_id):
         for db_report, report in self.__added_reports:
             LOG.debug("Storing bug path positions.")
-            for i, p in enumerate(report.bug_path_positions):
+            for idx, path_pos in enumerate(report.bug_path_positions):
                 session.add(BugReportPoint(
-                    p.range.start_line, p.range.start_col,
-                    p.range.end_line, p.range.end_col,
-                    i, file_path_to_id[p.file.path], db_report.id))
+                    path_pos.range.start_line, path_pos.range.start_col,
+                    path_pos.range.end_line, path_pos.range.end_col,
+                    idx, file_path_to_id[path_pos.file.path], db_report.id))
 
             LOG.debug("Storing bug path events.")
-            for i, event in enumerate(report.bug_path_events):
+            for idx, event in enumerate(report.bug_path_events):
                 session.add(BugPathEvent(
                     event.range.start_line, event.range.start_col,
                     event.range.end_line, event.range.end_col,
-                    i, event.message, file_path_to_id[event.file.path],
+                    idx, event.message, file_path_to_id[event.file.path],
                     db_report.id))
 
             LOG.debug("Storing notes.")
@@ -1405,12 +1405,6 @@ def __validate_and_add_report_annotations(
                     f"'{value}' has wrong format. '{key}' annotations must be "
                     f"'{report_annotation_types[key]['display']}'.")
 
-    def __get_report_limit_for_product(self):
-        with DBSession(self.__config_database) as session:
-            product = session.query(Product).get(self.__product.id)
-            if product.report_limit:
-                self.__report_limit = product.report_limit
-
     def __check_report_count(self):
         """
         This method comparest the already added report count to the report

web/server/codechecker_server/api/report_server.py

@@ -3947,6 +3947,12 @@ def __massStoreRun_common(self, is_async: bool, zipfile_blob: str,
         m: MassStoreRunTask = ih.create_mass_store_task(is_async)
         self._task_manager.push_task(m)
 
+        if not is_async:
+            LOG.info("massStoreRun(): Blocking until task '%s' terminates ...",
+                     m.token)
+            pipe = self._task_manager.get_task_receiver_pipe(m.token)
+            pipe.recv()
+
         return m.token
 
     @exc_to_thrift_reqfail
@@ -3968,27 +3974,21 @@ def massStoreRun(self,
                                          )
         token = self.__massStoreRun_common(False, b64zip, store_opts)
 
-        LOG.info("massStoreRun(): Blocking until task '%s' terminates ...",
-                 token)
-
         # To be compatible with older (<= 6.24, API <= 6.58) clients which
         # may keep using the old API endpoint, simulate awaiting the
         # background task in the API handler.
-        while True:
-            time.sleep(5)
-            t = self._task_manager.get_task_record(token)
-            if t.is_in_terminated_state:
-                if t.status == "failed":
-                    raise codechecker_api_shared.ttypes.RequestFailed(
-                        codechecker_api_shared.ttypes.ErrorCode.GENERAL,
-                        "massStoreRun()'s processing failed. Here follow "
-                        f"the details:\n\n{t.comments}")
-                if t.status == "cancelled":
-                    raise codechecker_api_shared.ttypes.RequestFailed(
-                        codechecker_api_shared.ttypes.ErrorCode.GENERAL,
-                        "Server administrators cancelled the processing of "
-                        "the massStoreRun() request!")
-                break
+        task = self._task_manager.get_task_record(token)
+        if task.is_in_terminated_state:
+            if task.status == "failed":
+                raise codechecker_api_shared.ttypes.RequestFailed(
+                    codechecker_api_shared.ttypes.ErrorCode.GENERAL,
+                    "massStoreRun()'s processing failed. Here follow "
+                    f"the details:\n\n{task.comments}")
+            if task.status == "cancelled":
+                raise codechecker_api_shared.ttypes.RequestFailed(
+                    codechecker_api_shared.ttypes.ErrorCode.GENERAL,
+                    "Server administrators cancelled the processing of "
+                    "the massStoreRun() request!")
 
         # Prior to CodeChecker 6.25.0 (API v6.59), massStoreRun() was
         # completely synchronous and blocking, and the implementation of the

web/server/codechecker_server/server.py

@@ -49,7 +49,7 @@
 
 from codechecker_common import util
 from codechecker_common.compatibility.multiprocessing import \
-    Pool, Process, Queue, Value, cpu_count
+    Pool, Process, Queue, Value, cpu_count, SyncManager
 from codechecker_common.logger import get_logger, signal_log
 
 from codechecker_web.shared import database_status
@@ -635,6 +635,7 @@ def __init__(self,
                  manager: session_manager.SessionManager,
                  machine_id: str,
                  task_queue: Queue,
+                 task_pipes,
                  server_shutdown_flag: Value):
 
         LOG.debug("Initializing HTTP server...")
@@ -658,7 +659,7 @@ def __init__(self,
 
         self.__task_queue = task_queue
         self.task_manager = BackgroundTaskManager(
-            task_queue, self.config_session, self.check_env,
+            task_queue, task_pipes, self.config_session, self.check_env,
             server_shutdown_flag, machine_id,
             pathlib.Path(self.context.codechecker_workspace))
 
@@ -1061,23 +1062,27 @@ def start_server(config_directory: str, workspace_directory: str,
     bg_task_queue: Queue = Queue()
     is_server_shutting_down = Value('B', False)
 
+    sync_manager = SyncManager()
+    sync_manager.start()
+    task_pipes = sync_manager.dict()
+
     def _cleanup_incomplete_tasks(action: str) -> int:
         config_db = config_sql_server.create_engine()
         config_session_factory = sessionmaker(bind=config_db)
-        tm = BackgroundTaskManager(
-            bg_task_queue, config_session_factory, check_env,
+        task_manager = BackgroundTaskManager(
+            bg_task_queue, task_pipes, config_session_factory, check_env,
             is_server_shutting_down, machine_id,
             pathlib.Path(context.codechecker_workspace))
 
         try:
-            tm.destroy_all_temporary_data()
+            task_manager.destroy_all_temporary_data()
         except OSError:
             LOG.warning("Clearing task-temporary storage space failed!")
             import traceback
             traceback.print_exc()
 
         try:
-            return tm.drop_all_incomplete_tasks(action)
+            return task_manager.drop_all_incomplete_tasks(action)
         finally:
             config_db.dispose()
 
@@ -1107,6 +1112,7 @@ def _cleanup_incomplete_tasks(action: str) -> int:
                                manager,
                                machine_id,
                                bg_task_queue,
+                               task_pipes,
                                is_server_shutting_down)
 
     try:
@@ -1190,6 +1196,7 @@ def spawn_bg_process():
         p = _start_process_with_no_signal_handling(
             target=background_task_executor,
             args=(bg_task_queue,
+                  task_pipes,
                   config_sql_server,
                   check_env,
                   is_server_shutting_down,

web/server/codechecker_server/task_executors/abstract_task.py

@@ -116,6 +116,7 @@ def _log_cancel_and_abandon(db_task: DBTask):
                 db_task.set_abandoned(force_dropped_status=False)
 
             task_manager._mutate_task_record(self, _log_cancel_and_abandon)
+            task_manager._send_done_message(self.token)
             return
 
         try:
@@ -137,6 +138,7 @@ def _log_cancel_and_abandon(db_task: DBTask):
                     self, lambda dbt:
                     dbt.set_abandoned(force_dropped_status=True))
             except Exception:
+                task_manager._send_done_message(self.token)
                 return
 
         LOG.debug("Task '%s' running on machine '%s' executor #%d",
@@ -200,3 +202,4 @@ def _log_exception_and_fail(db_task: DBTask):
             task_manager._mutate_task_record(self, _log_exception_and_fail)
         finally:
             self.destroy_data()
+            task_manager._send_done_message(self.token)

web/server/codechecker_server/task_executors/main.py

@@ -30,6 +30,7 @@
 
 
 def executor(queue: Queue,
+             task_pipes,
              config_db_sql_server,
              server_environment,
              server_shutdown_flag: "Value",
@@ -67,7 +68,7 @@ def executor_hangup_handler(signum: int, _frame):
     signal.signal(signal.SIGHUP, executor_hangup_handler)
 
     config_db_engine = config_db_sql_server.create_engine()
-    tm = TaskManager(queue, sessionmaker(bind=config_db_engine),
+    tm = TaskManager(queue, task_pipes, sessionmaker(bind=config_db_engine),
                      server_environment, kill_flag, machine_id)
 
     while not kill_flag.value: