diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 54abfa91..dcd9fb78 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -23,11 +23,8 @@ repos: args: [--exclude, tests] # args: [--check] - - repo: local + - repo: https://github.com/pre-commit/mirrors-isort + rev: v5.10.1 hooks: - id: isort - name: isort - entry: isort - language: system - types: [python] - args: [--profile=black, --skip, tests] \ No newline at end of file + args: [--profile=black, --skip=tests] \ No newline at end of file diff --git a/docs/check.md b/docs/check.md index f2d418c3..37bc3a0b 100644 --- a/docs/check.md +++ b/docs/check.md @@ -2,19 +2,31 @@ `traincheck-check` is the **final stage** of the TrainCheck workflow. It verifies a set of invariants against trace files or streams from target programs, reporting any detected violations—helping you catch silent issues in your ML training pipelines. -## 🔧 Current Status +## 🔧 Checking Modes -`traincheck-check` is designed to support two modes: +TrainCheck supports two checking modes: -- **Offline Checking**: - Perform invariant checking on completed trace files after the training job finishes. ✅ *[Fully Supported]* +- **Post-training Checking (`traincheck-check`)**: + Perform invariant checking on completed trace files after the training job finishes. ✅ -- **Online Checking**: - Perform real-time checking while the target training job is running. 🚧 *[In Development]* +- **On-the-fly Checking (`traincheck-onlinecheck`):** + Perform real-time checking while the target training job is running. ✅ -At present, only **offline checking** is available. Support for online mode is actively being developed. +## How to Use: On-the-fly Checking -## How to Use: Offline Checking +While training is in progress with `traincheck-collect`, run the following command: + +```bash +traincheck-onlinecheck -f -i +``` + +- `-f `: Path to the folder where traces are: + - Already collected, or + - **Actively being collected** by `traincheck-collect` during the training job. + +- `-i `: Path to the JSON file containing inferred invariants. + +## How to Use: Post-training Checking Run the following command: @@ -25,11 +37,12 @@ traincheck-check -f -i - `-f `: Path to the folder containing traces collected by `traincheck-collect`. - `-i `: Path to the JSON file containing inferred invariants. -For details on result format and interpretation, refer to [5. Detection & Diagnosis)](./5-min-tutorial.md#5-detection--diagnosis) in the **5-Minute Tutorial**. +## Interpreting the Results -## How to Use: Online Checking +After running either checking mode, TrainCheck will output a summary of detected invariant violations. Each violation entry typically includes: -**🚧 Coming Soon** -Support for real-time, online checking is under construction. This mode will allow TrainCheck to monitor running training jobs and surface invariant violations as they happen. +- **Trace file or stream name**: Identifies where the issue was found. +- **Invariant description**: Details the specific invariant that was violated. +- **Violation details**: Provides context, such as the step or epoch where the violation occurred. -Stay tuned for updates in future releases. +Review these results to pinpoint silent errors or unexpected behaviors in your ML training pipeline. For more information on result formats and how to diagnose issues, see [5. Detection & Diagnosis](./5-min-tutorial.md#5-detection--diagnosis) in the **5-Minute Tutorial**. \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 89876f6d..3dc8a57d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -52,3 +52,4 @@ homepage = "https://github.com/OrderLab/TrainCheck" traincheck-collect = "traincheck:collect_trace.main" traincheck-infer = "traincheck:infer_engine.main" traincheck-check = "traincheck:checker.main" +traincheck-onlinecheck = "traincheck:checker_online.main" diff --git a/tests/test_utils.py b/tests/test_utils.py index a7ae27fc..f38221c0 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -16,6 +16,7 @@ def test_typename_builtin_function(): assert typename(len) == "len" assert typename(print) == "print" + def test_typename_tensor_and_parameter(): t = torch.tensor([1.0]) assert typename(t) == t.type() diff --git a/traincheck/checker.py b/traincheck/checker.py index 215ea1f1..62045468 100644 --- a/traincheck/checker.py +++ b/traincheck/checker.py @@ -42,7 +42,6 @@ def check_engine( inv.precondition is not None ), "Invariant precondition is None. It should at least be 'Unconditional' or an empty list. Please check the invariant file and the inference process." logger.info("=====================================") - # logger.debug("Checking invariant %s on trace %s", inv, trace) res = inv.check(trace, check_relation_first) res.calc_and_set_time_precentage(trace.get_start_time(), trace.get_end_time()) logger.info("Invariant %s on trace %s: %s", inv, trace, res) @@ -147,7 +146,7 @@ def main(): for inv_file in args.invariants: os.system(f"cp {inv_file} {args.output_dir}/invariants.json") - logger.info("Reading invaraints from %s", "\n".join(args.invariants)) + logger.info("Reading invariants from %s", "\n".join(args.invariants)) invs = read_inv_file(args.invariants) traces = [] diff --git a/traincheck/checker_online.py b/traincheck/checker_online.py new file mode 100644 index 00000000..d71d5248 --- /dev/null +++ b/traincheck/checker_online.py @@ -0,0 +1,371 @@ +import argparse +import datetime +import json +import logging +import os +import signal +import sys +import time + +from traincheck.config import config +from traincheck.invariant import read_inv_file +from traincheck.invariant.base_cls import APIParam, Invariant, Param, VarTypeParam +from traincheck.onlinechecker.streamhandler_filesystem import run_stream_monitor +from traincheck.onlinechecker.utils import Checker_data +from traincheck.trace import MDNONEJSONEncoder +from traincheck.trace.types import VarInstId + +OBSERVER = None +KILLING_PROCESS = ( + False # True indicates that SIGTERM has been sent to the running process +) +NUM_VIOLATIONS = 0 +FAILED_INV = dict() + +ORIGINAL_SIGINT_HANDLER = signal.getsignal(signal.SIGINT) +ORIGINAL_SIGTERM_HANDLER = signal.getsignal(signal.SIGTERM) + + +def handle_SIGINT(signum, frame): + global KILLING_PROCESS + + print("Received SIGINT") + if KILLING_PROCESS: + exit(130) + return + KILLING_PROCESS = True + try: + stop_checker() + except Exception as e: + print(f"Error when stopping checker: {e}") + # if callable(ORIGINAL_SIGINT_HANDLER): + # ORIGINAL_SIGINT_HANDLER(signum, frame) + exit(130) + + +def handle_SIGTERM(signum, frame): + global KILLING_PROCESS + + print("Received SIGTERM") + if KILLING_PROCESS: + exit(143) + return + KILLING_PROCESS = True + try: + stop_checker() + except Exception as e: + print(f"Error when stopping checker: {e}") + if callable(ORIGINAL_SIGTERM_HANDLER): + ORIGINAL_SIGTERM_HANDLER(signum, frame) + else: + exit(143) + + +curr_excepthook = sys.excepthook + + +def kill_running_process_on_except(typ, value, tb): + stop_checker() + curr_excepthook(typ, value, tb) + + +def register_hook_closing_program(): + signal.signal(signal.SIGTERM, handle_SIGTERM) + signal.signal(signal.SIGINT, handle_SIGINT) + sys.excepthook = kill_running_process_on_except + + +def sort_inv_file(invariants): + """Sort the invariants by their parameters. Also collect the needed data for online checking. + Return: + param_to_invs: dict[Param, list[Invariant]] + vartype_to_invs: dict[str, dict[str, list[Invariant]]] + needed_data: (set[str], set[str], set[str]) + """ + logger = logging.getLogger(__name__) + logger.info("Reading invariants from file: %s", invariants) + + invs = read_inv_file(invariants) + logger.info("Total %d invariants read from file: %s", len(invs), invariants) + logger.info("Sorting invariants by parameters") + + param_to_invs: dict[Param, list[Invariant]] = {} + vartype_to_invs: dict[str, dict[str, list[Invariant]]] = {} + needed_vars = set() + needed_apis = set() + _get_api_args_map_to_check = set() + for inv in invs: + assert ( + inv.precondition is not None + ), "Invariant precondition is None. It should at least be 'Unconditional' or an empty list. Please check the invariant file and the inference process." + params = inv._get_identifying_params() + needed_var, needed_api, needed_args_api = ( + inv._get_information_sources_to_check() + ) + if needed_var is not None: + needed_vars.update(needed_var) + if needed_api is not None: + needed_apis.update(needed_api) + if needed_args_api is not None: + _get_api_args_map_to_check.update(needed_args_api) + for param in params: + if isinstance(param, VarTypeParam): + if param.var_type not in vartype_to_invs: + vartype_to_invs[param.var_type] = {} + if param.attr_name not in vartype_to_invs[param.var_type]: + vartype_to_invs[param.var_type][param.attr_name] = [] + vartype_to_invs[param.var_type][param.attr_name].append(inv) + else: + if param not in param_to_invs: + param_to_invs[param] = [] + param_to_invs[param].append(inv) + logger.info("Sorting done.") + needed_data = (needed_vars, needed_apis, _get_api_args_map_to_check) + return param_to_invs, vartype_to_invs, needed_data + + +def get_violated_pair_hash(trace_pair): + from traincheck.invariant.base_cls import make_hashable + + h1 = hash(make_hashable(trace_pair[0])) + h2 = hash(make_hashable(trace_pair[1])) + return tuple(sorted((h1, h2), reverse=True)) + + +def check( + invariants, traces, trace_folders, output_dir: str, check_relation_first: bool +): + global OBSERVER + global NUM_VIOLATIONS + global FAILED_INV + + register_hook_closing_program() + + logger = logging.getLogger(__name__) + logger.addHandler(logging.StreamHandler()) + logger.info("Starting online checker") + + param_to_invs, vartype_to_invs, needed_data = sort_inv_file(invariants) + checker_data = Checker_data(needed_data) + OBSERVER = run_stream_monitor(traces, trace_folders, checker_data) + + output_file = os.path.join(output_dir, "failed.log") + violated_pairs = dict[Invariant, set[tuple[int, int]]]() + + while True: + trace_record = checker_data.check_queue.get() + if checker_data.check_queue.empty(): + logger.debug("Check queue is empty") + if trace_record is None: + continue + + with checker_data.cond: + while True: + if trace_record["time"] > checker_data.min_read_time: + logger.debug("Wait for the different trace file to catch up") + checker_data.cond.wait() + logger.debug("Woke up from wait") + else: + break + + if "var_name" in trace_record and trace_record["var_name"] is not None: + varid = VarInstId( + trace_record["process_id"], + trace_record["var_name"], + trace_record["var_type"], + ) + if varid.var_type in vartype_to_invs: + for attr_name, invs in vartype_to_invs[varid.var_type].items(): + attr_name = config.VAR_ATTR_PREFIX + attr_name + if ( + attr_name in trace_record + and trace_record[attr_name] is not None + ): + for inv in invs: + try: + result = inv.online_check( + trace_record, checker_data, check_relation_first + ) + if not result.check_passed: + violated_pair = get_violated_pair_hash(result.trace) + if inv not in violated_pairs: + violated_pairs[inv] = set() + if violated_pair not in violated_pairs[inv]: + violated_pairs[inv].add(violated_pair) + else: + continue + if inv not in FAILED_INV: + FAILED_INV[inv] = 0 + FAILED_INV[inv] += 1 + NUM_VIOLATIONS += 1 + result.set_id_and_detection_time( + NUM_VIOLATIONS, time.monotonic_ns() + ) + logger.error( + f"Violated id {NUM_VIOLATIONS}:\nInvariant {inv} violated near time {trace_record['time']}" + ) + with open(output_file, "a") as f: + json.dump( + result.to_dict(), + f, + indent=4, + cls=MDNONEJSONEncoder, + ) + f.write("\n") + except Exception as e: + logger.error( + f"Error when checking invariant {inv.text_description} with trace {trace_record}: {e}" + ) + + elif ( + "func_call_id" in trace_record and trace_record["func_call_id"] is not None + ): + apiparam = APIParam(trace_record["function"]) + if apiparam in param_to_invs: + for inv in param_to_invs[apiparam]: + try: + result = inv.online_check( + trace_record, checker_data, check_relation_first + ) + if not result.check_passed: + if inv not in FAILED_INV: + FAILED_INV[inv] = 0 + FAILED_INV[inv] += 1 + NUM_VIOLATIONS += 1 + result.set_id_and_detection_time( + NUM_VIOLATIONS, time.monotonic_ns() + ) + logger.error( + f"Violated id {NUM_VIOLATIONS}:\nInvariant {inv} violated near time {trace_record['time']}" + ) + with open(output_file, "a") as f: + json.dump( + result.to_dict(), f, indent=4, cls=MDNONEJSONEncoder + ) + f.write("\n") + except Exception as e: + logger.error( + f"Error when checking invariant {inv.text_description} with trace {trace_record}: {e}" + ) + + +def stop_checker(): + global OBSERVER + if OBSERVER is None: + return + + OBSERVER.stop() + OBSERVER.join() + + global NUM_VIOLATIONS + global FAILED_INV + + logger = logging.getLogger(__name__) + logger.info("Checker stopped") + logger.info(f"Total {NUM_VIOLATIONS} violations found") + logger.info(f"Total {len(FAILED_INV)} invariants violated:") + # for inv, count in failed_inv.items(): + # logger.info(f"Invariant {inv} violated {count} times") + logger.info("Violations are stored") + + +def main(): + parser = argparse.ArgumentParser( + description="(Online) Invariant Checker for ML Pipelines in Python" + ) + parser.add_argument( + "-t", + "--traces", + nargs="+", + required=False, + help="Traces files to infer invariants on", + ) + parser.add_argument( + "-f", + "--trace-folders", + nargs="+", + help='Folders containing traces files to infer invariants on. Trace files should start with "trace_" or "proxy_log.json"', + ) + parser.add_argument( + "-i", + "--invariants", + nargs="+", + required=True, + help="Invariants files to check on traces", + ) + parser.add_argument( + "-d", + "--debug", + action="store_true", + help="Enable debug logging", + ) + parser.add_argument( + "--check-relation-first", + action="store_true", + help="""Check the relation first, otherwise, the precondition will be checked first. + Enabling this flag will make the checker slower, but enables the checker to catch + the cases where the invariant still holds even if the precondition is not satisfied, + which opens opportunity for precondition refinement. Note that the precondition + refinement algorithm is not implemented yet.""", + ) + parser.add_argument( + "-o", + "--output-dir", + type=str, + help="Output folder to store the results, defaulted to traincheck_checker_results_{timestamp}/", + ) + + args = parser.parse_args() + + # check if either traces or trace folders are provided + if args.traces is None and args.trace_folders is None: + # print help message if neither traces nor trace folders are provided + parser.print_help() + parser.error( + "Please provide either traces or trace folders to infer invariants" + ) + + if args.invariants is None: + parser.print_help() + parser.error("Please provide exactly one invariant file to check") + + if args.debug: + log_level = logging.DEBUG + else: + log_level = logging.INFO + + time_now = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S") + + ## DEBUG + time_now = f"{time_now}_relation_first_{args.check_relation_first}" + # set logging to a file + logging.basicConfig( + filename=f"traincheck_onlinechecker_{time_now}.log", + level=log_level, + ) + + logger = logging.getLogger(__name__) + # log all the arguments + logger.info("Checker started with Arguments:") + for arg, val in vars(args).items(): + logger.info("%s: %s", arg, val) + + if not args.output_dir: + args.output_dir = f"traincheck_onlinechecker_results_{time_now}" + os.makedirs(args.output_dir, exist_ok=True) + + # copy the invariants to the output folder + for inv_file in args.invariants: + os.system(f"cp {inv_file} {args.output_dir}/invariants.json") + + check( + args.invariants, + args.traces, + args.trace_folders, + args.output_dir, + args.check_relation_first, + ) + + +if __name__ == "__main__": + main() diff --git a/traincheck/invariant/DistinctArgumentRelation.py b/traincheck/invariant/DistinctArgumentRelation.py index f0c2c45c..025f7c69 100644 --- a/traincheck/invariant/DistinctArgumentRelation.py +++ b/traincheck/invariant/DistinctArgumentRelation.py @@ -3,6 +3,7 @@ from tqdm import tqdm +from traincheck.instrumentor.tracer import TraceLineType from traincheck.invariant.base_cls import ( # GroupedPreconditions, APIParam, CheckerResult, @@ -11,9 +12,12 @@ FailedHypothesis, Hypothesis, Invariant, + OnlineCheckerResult, + Param, Relation, ) from traincheck.invariant.precondition import find_precondition +from traincheck.onlinechecker.utils import Checker_data, set_meta_vars_online from traincheck.trace.trace import Trace from traincheck.utils import safe_isnan @@ -449,6 +453,91 @@ def static_check_all( triggered=True, ) + @staticmethod + def _get_identifying_params(inv: Invariant) -> list[Param]: + return [inv.params[0]] + + @staticmethod + def _get_variables_to_check(inv: Invariant): + return None + + @staticmethod + def _get_apis_to_check(inv: Invariant): + return None + + @staticmethod + def _get_api_args_map_to_check(inv): + return [inv.params[0].api_full_name] + + @staticmethod + def online_check( + check_relation_first: bool, + inv: Invariant, + trace_record: dict, + checker_data: Checker_data, + ): + if trace_record["type"] != TraceLineType.FUNC_CALL_PRE: + return OnlineCheckerResult( + trace=None, + invariant=inv, + check_passed=True, + ) + + assert inv.precondition is not None, "Invariant should have a precondition." + + func_param = inv.params[0] + assert isinstance( + func_param, APIParam + ), "Invariant parameters should be APIParam." + func_name = func_param.api_full_name + + assert ( + func_name == trace_record["function"] + ), "Function name in the invariant should match the function name in the trace record." + + with checker_data.lock: + [trace_record] = set_meta_vars_online([trace_record], checker_data) + + if not inv.precondition.verify([trace_record], EXP_GROUP_NAME, None): + return OnlineCheckerResult( + trace=None, + invariant=inv, + check_passed=True, + ) + + if "meta_vars.step" not in trace_record: + step = -1 + else: + step = trace_record["meta_vars.step"] + + with checker_data.lock: + check_func_list = checker_data.args_map[func_name][step] + + for PT_pair1, PT_pair2 in combinations(check_func_list.keys(), 2): + for event1 in check_func_list[PT_pair1]: + for event2 in check_func_list[PT_pair2]: + if is_arguments_list_same(event1["args"], event2["args"]): + return OnlineCheckerResult( + trace=[event1, event2], + invariant=inv, + check_passed=False, + ) + + for PT_pair in check_func_list.keys(): + for event1, event2 in combinations(check_func_list[PT_pair], 2): + if is_arguments_list_same(event1["args"], event2["args"]): + return OnlineCheckerResult( + trace=[event1, event2], + invariant=inv, + check_passed=False, + ) + + return OnlineCheckerResult( + trace=None, + invariant=inv, + check_passed=True, + ) + @staticmethod def get_precondition_infer_keys_to_skip(hypothesis: Hypothesis) -> list[str]: return [] diff --git a/traincheck/invariant/base_cls.py b/traincheck/invariant/base_cls.py index 6ea4c53b..d8649377 100644 --- a/traincheck/invariant/base_cls.py +++ b/traincheck/invariant/base_cls.py @@ -375,6 +375,12 @@ def check_event_match(self, event: HighLevelEvent) -> bool: "Check if the high level event contains the required information for the param." raise NotImplementedError("check_event_match method is not implemented yet.") + def check_event_match_online(self, event) -> bool: + "Check if the high level event contains the required information for the param." + raise NotImplementedError( + "check_event_match_online method is not implemented yet." + ) + def get_customizable_field_names(self) -> set[str]: """Returns the field names that can be customized for the param.""" raise NotImplementedError( @@ -433,6 +439,33 @@ def check_event_match(self, event: HighLevelEvent) -> bool: return matched + def check_event_match_online(self, event) -> bool: + if not isinstance( + event, (FuncCallEvent, FuncCallExceptionEvent, IncompleteFuncCallEvent) + ): + return False + + # TODO: Handle Stop Iteration Exception!!! + matched = True + if self.exception != _NOT_SET: + matched = ( + isinstance(event, FuncCallExceptionEvent) + and event.exception == self.exception + ) + else: + matched = not isinstance(event, FuncCallExceptionEvent) + + if not matched: + return False + + # check the arguments if they are provided + if isinstance(self.arguments, Arguments): + # current_args should not violate the provided arguments (i.e., self.arguments should be a subset of current_args) + current_args = Arguments(event.args, event.kwargs, event.func_name) + matched = matched and not self.arguments.check_for_violation(current_args) + + return matched + def with_no_customization(self) -> APIParam: return APIParam(self.api_full_name) @@ -552,6 +585,40 @@ def check_event_match(self, event: HighLevelEvent) -> bool: return pre_and_post_value_matched + def check_event_match_online(self, event) -> bool: + if self.const_value != _NOT_SET: + logger = logging.getLogger(__name__) + logger.warning( + "Const value is set for VarTypeParam, this should be checked in the relation's evaluate method instead of the check_event_match method." + ) + + pre_and_post_value_matched = True + if self.pre_value != _NOT_SET: + if self.pre_value != event[0].value: + if self.pre_value in GENERALIZED_TYPES: + pre_and_post_value_matched = ( + pre_and_post_value_matched + and check_generalized_value_match( + self.pre_value, event[0].value + ) + ) + else: + return False + + if self.post_value != _NOT_SET: + if self.post_value != event[1].value: + if self.post_value in GENERALIZED_TYPES: + pre_and_post_value_matched = ( + pre_and_post_value_matched + and check_generalized_value_match( + self.post_value, event[1].value + ) + ) + else: + return False + + return pre_and_post_value_matched + def check_var_id_match(self, var_id: VarInstId) -> bool: return var_id.var_type == self.var_type @@ -651,6 +718,40 @@ def check_event_match(self, event: HighLevelEvent) -> bool: return var_attr_matched and pre_and_post_value_matched + def check_event_match_online(self, event) -> bool: + if self.const_value != _NOT_SET: + logger = logging.getLogger(__name__) + logger.warning( + "Const value is set for VarTypeParam, this should be checked in the relation's evaluate method instead of the check_event_match method." + ) + + pre_and_post_value_matched = True + if self.pre_value != _NOT_SET: + if self.pre_value != event[0].value: + if self.pre_value in GENERALIZED_TYPES: + pre_and_post_value_matched = ( + pre_and_post_value_matched + and check_generalized_value_match( + self.pre_value, event[0].value + ) + ) + else: + return False + + if self.post_value != _NOT_SET: + if self.post_value != event[1].value: + if self.post_value in GENERALIZED_TYPES: + pre_and_post_value_matched = ( + pre_and_post_value_matched + and check_generalized_value_match( + self.post_value, event[1].value + ) + ) + else: + return False + + return pre_and_post_value_matched + def check_var_id_match(self, var_id: VarInstId) -> bool: return var_id.var_type == self.var_type and var_id.var_name == self.var_name @@ -1349,11 +1450,40 @@ def check(self, trace: Trace, check_relation_first: bool) -> CheckerResult: logging.getLogger(__name__).info( f"Checking invariant: {self.text_description} of relation {self.relation}" ) - print( - f"Checking invariant: {self.text_description} of relation {self.relation}" - ) return self.relation.static_check_all(trace, self, check_relation_first) + def online_check( + self, trace: dict, checker_data, check_relation_first: bool + ) -> OnlineCheckerResult: + assert ( + self.precondition is not None + ), "Invariant precondition is None. It should at least be 'Unconditional' or an empty list. Please check the invariant file and the inference process." + + return self.relation.online_check( + check_relation_first, self, trace, checker_data + ) + + def _get_identifying_params(self) -> list[Param]: + """Retrieves the identifying parameters for the invariant instance. + + This internal method is used to obtain the parameters that uniquely identify + the invariant in the context of online checking. These parameters are essential + for grouping invariants that pertain to the same API calls or variables, thereby + improving the efficiency of invariant management and evaluation. + + Returns: + list[Param]: A list of parameters that uniquely identify the invariant. + """ + return self.relation._get_identifying_params(self) + + def _get_information_sources_to_check(self): + """Retrieves the information sources that the invariant will need to check.""" + return ( + self.relation._get_variables_to_check(self), + self.relation._get_apis_to_check(self), + self.relation._get_api_args_map_to_check(self), + ) + class CheckerResult: def __init__( @@ -1427,6 +1557,58 @@ def to_dict(self): return result_dict +class OnlineCheckerResult: + def __init__( + self, + trace: Optional[list[dict]], + invariant: Invariant, + check_passed: bool, + ): + if trace is None: + assert check_passed, "Check passed should be True for None trace" + else: + assert len(trace) > 0, "Trace should not be empty" + self.trace = trace + self.invariant = invariant + self.check_passed = check_passed + + def __str__(self) -> str: + return f"Trace: {self.trace}\nInvariant: {self.invariant}\nResult: {self.check_passed}" + + def __repr__(self): + return self.__str__() + + def set_id_and_detection_time(self, id: int, detection_time: float): + self.id = id + self.detection_time = detection_time + + def to_dict(self): + """Convert the OnlineCheckerResult object to a json serializable dictionary.""" + assert hasattr( + self, "id" + ), "ID NOT SET for OnlineCheckerResult, please set it before converting to dict" + assert hasattr( + self, "detection_time" + ), "Detection time NOT SET for OnlineCheckerResult, please set it before converting to dict" + result_dict = { + "violated_id": self.id, + "invariant": self.invariant.to_dict(), + "check_passed": self.check_passed, + } + + trace = self.trace.copy() + MD_NONE.replace_with_none(trace) + + result_dict.update( + { + "detection_time": self.detection_time, + "trace": trace, + } + ) + + return result_dict + + def make_hashable(value): """Recursively convert a value into a hashable form.""" if isinstance(value, dict): @@ -1707,6 +1889,38 @@ def static_check_all( """ pass + @staticmethod + @abc.abstractmethod + def _get_identifying_params(inv: Invariant) -> list[Param]: + """Given an invariant, should return a list of Param objects that should be checked.""" + pass + + @staticmethod + @abc.abstractmethod + def _get_variables_to_check(inv: Invariant): + """Given an invariant, should return a list of variable names or variable type that are needed to check the invariant.""" + pass + + @staticmethod + @abc.abstractmethod + def _get_apis_to_check(inv: Invariant): + """Given an invariant, should return a list of API names that are needed to check the invariant.""" + pass + + @staticmethod + @abc.abstractmethod + def _get_api_args_map_to_check(inv: Invariant): + """Given an invariant, should return a list of API names that needs the args map.""" + pass + + @staticmethod + @abc.abstractmethod + def online_check( + check_relation_first: bool, inv: Invariant, trace: dict, checker_data + ) -> OnlineCheckerResult: + """Check the invariant online, i.e. during the trace collection process.""" + pass + def read_inv_file(file_path: str | list[str]) -> list[Invariant]: if isinstance(file_path, str): diff --git a/traincheck/invariant/consistency_relation.py b/traincheck/invariant/consistency_relation.py index 1748e9db..06b24714 100644 --- a/traincheck/invariant/consistency_relation.py +++ b/traincheck/invariant/consistency_relation.py @@ -12,12 +12,15 @@ FailedHypothesis, Hypothesis, Invariant, + OnlineCheckerResult, + Param, Relation, VarTypeParam, ) from traincheck.invariant.precondition import find_precondition +from traincheck.onlinechecker.utils import Checker_data, set_meta_vars_online from traincheck.trace.trace import Trace -from traincheck.trace.types import Liveness +from traincheck.trace.types import Liveness, VarInstId tracker_var_field_prefix = "attributes." @@ -525,6 +528,148 @@ def static_check_all( triggered=inv_triggered, ) + @staticmethod + def _get_identifying_params(inv: Invariant) -> list[Param]: + if inv.params[0] == inv.params[1]: + return [inv.params[0]] + return [inv.params[0], inv.params[1]] + + @staticmethod + def _get_variables_to_check(inv): + if inv.params[0].var_type == inv.params[1].var_type: + return [inv.params[0].var_type] + return [inv.params[0].var_type, inv.params[1].var_type] + + @staticmethod + def _get_apis_to_check(inv: Invariant): + return None + + @staticmethod + def _get_api_args_map_to_check(inv): + return None + + @staticmethod + def online_check( + check_relation_first: bool, + inv: Invariant, + trace_record: dict, + checker_data: Checker_data, + ): + assert len(inv.params) == 2, "Invariant should have exactly two parameters." + assert inv.precondition is not None, "Invariant should have a precondition." + + logger = logging.getLogger(__name__) + + param1 = inv.params[0] + param2 = inv.params[1] + assert isinstance(param1, VarTypeParam) and isinstance( + param2, VarTypeParam + ), "Invariant parameters should be VarTypeParam." + + varid = VarInstId( + trace_record["process_id"], + trace_record["var_name"], + trace_record["var_type"], + ) + + def get_check_attr(param, varid): + if param.var_type != varid.var_type: + return None + if ( + param.attr_name not in checker_data.varid_map[varid] + or len(checker_data.varid_map[varid][param.attr_name]) < 2 + ): + return None + for attr in reversed(checker_data.varid_map[varid][param.attr_name]): + if attr.liveness.start_time < trace_record["time"]: + return attr + + return None + + with checker_data.lock: + check_attr1 = get_check_attr(param1, varid) + check_attr2 = get_check_attr(param2, varid) + if param1 == param2: + check_attr2 = None + + for check_attr, ref_param in [(check_attr1, param2), (check_attr2, param1)]: + if check_attr is None: + continue + with checker_data.lock: + if ref_param.var_type not in checker_data.type_map: + logger.debug( + f"Variable type {ref_param.var_type} not found in checker_data" + ) + continue + + for var2 in checker_data.type_map[ref_param.var_type]: + if var2 == varid: + continue + if checker_data.varid_map[var2][ref_param.attr_name] is None: + logger.debug( + f"Attribute {ref_param.attr_name} not found in variable {var2}" + ) + continue + + for attr in reversed( + checker_data.varid_map[var2][ref_param.attr_name] + ): + if attr.liveness.start_time > trace_record["time"]: + continue + if attr.liveness.end_time is None: + continue + if attr.liveness.end_time <= check_attr.liveness.start_time: + break + if attr.liveness.start_time >= check_attr.liveness.end_time: + continue + overlap = calc_liveness_overlap( + check_attr.liveness, attr.liveness + ) + if overlap <= config.LIVENESS_OVERLAP_THRESHOLD: + continue + if check_relation_first: + compare_result = ConsistencyRelation.evaluate( + [check_attr.value, attr.value] + ) + if not compare_result: + if inv.precondition.verify( + set_meta_vars_online( + [check_attr.traces[-1], attr.traces[-1]], + checker_data, + ), + VAR_GROUP_NAME, + None, + ): + return OnlineCheckerResult( + trace=[check_attr.traces[-1], attr.traces[-1]], + invariant=inv, + check_passed=False, + ) + else: + if inv.precondition.verify( + set_meta_vars_online( + [check_attr.traces[-1], attr.traces[-1]], + checker_data, + ), + VAR_GROUP_NAME, + None, + ): + compare_result = ConsistencyRelation.evaluate( + [check_attr.value, attr.value] + ) + if not compare_result: + return OnlineCheckerResult( + trace=[check_attr.traces[-1], attr.traces[-1]], + invariant=inv, + check_passed=False, + ) + + return OnlineCheckerResult( + trace=None, + invariant=inv, + check_passed=True, + ) + @staticmethod def get_precondition_infer_keys_to_skip(hypothesis: Hypothesis) -> list[str]: assert ( diff --git a/traincheck/invariant/consistency_transient_vars.py b/traincheck/invariant/consistency_transient_vars.py index faeb060a..570ed42f 100644 --- a/traincheck/invariant/consistency_transient_vars.py +++ b/traincheck/invariant/consistency_transient_vars.py @@ -5,6 +5,7 @@ import pandas as pd from tqdm import tqdm +from traincheck.instrumentor.tracer import TraceLineType from traincheck.invariant.base_cls import ( APIParam, Arguments, @@ -15,11 +16,14 @@ Hypothesis, InputOutputParam, Invariant, + OnlineCheckerResult, + Param, Relation, VarTypeParam, make_hashable, ) from traincheck.invariant.precondition import find_precondition +from traincheck.onlinechecker.utils import Checker_data, set_meta_vars_online from traincheck.trace.trace import Trace from traincheck.trace.types import ( FuncCallEvent, @@ -576,6 +580,83 @@ def static_check_all( ) # raise NotImplementedError + @staticmethod + def _get_identifying_params(inv: Invariant) -> list[Param]: + return [inv.params[0]] + + @staticmethod + def _get_variables_to_check(inv): + return None + + @staticmethod + def _get_apis_to_check(inv: Invariant): + assert isinstance(inv.params[0], APIParam) + return [inv.params[0].api_full_name] + + @staticmethod + def _get_api_args_map_to_check(inv): + return None + + @staticmethod + def online_check( + check_relation_first: bool, + inv: Invariant, + trace_record: dict, + checker_data: Checker_data, + ): + if trace_record["type"] != TraceLineType.FUNC_CALL_POST: + return OnlineCheckerResult( + trace=None, + invariant=inv, + check_passed=True, + ) + + assert inv.precondition is not None, "The precondition should not be None." + + assert len(inv.params) == 2 + assert isinstance(inv.params[0], APIParam) + assert isinstance(inv.params[1], VarTypeParam) + + func_name = trace_record["function"] + func_call_id = trace_record["func_call_id"] + process_id = trace_record["process_id"] + thread_id = trace_record["thread_id"] + ptname = (process_id, thread_id, func_name) + with checker_data.lock: + func_call_event = checker_data.pt_map[ptname][func_call_id] + func_pre_record = func_call_event.pre_record + [func_pre_record] = set_meta_vars_online([func_pre_record], checker_data) + + check_passed = True + + if inv.precondition.verify([func_pre_record], "pre_event", None): + returned_tensors = get_returned_tensors(func_call_event) + if len(returned_tensors) == 0: + check_passed = False + else: + for returned_tensor in returned_tensors: + prop = inv.params[1].attr_name + prop_val = inv.params[1].const_value + if ( + prop not in returned_tensor + or make_hashable(returned_tensor[prop]) != prop_val + ): + check_passed = False + break + + if not check_passed: + return OnlineCheckerResult( + trace=[func_pre_record], + invariant=inv, + check_passed=False, + ) + + return OnlineCheckerResult( + trace=None, + invariant=inv, + check_passed=True, + ) + @staticmethod def get_precondition_infer_keys_to_skip(hypothesis: Hypothesis) -> list[str]: return [] @@ -884,6 +965,92 @@ def static_check_all(trace, inv, check_relation_first): triggered=triggered, ) + @staticmethod + def _get_identifying_params(inv: Invariant) -> list[Param]: + return [inv.params[1]] + + @staticmethod + def _get_variables_to_check(inv): + return None + + @staticmethod + def _get_apis_to_check(inv: Invariant): + assert isinstance(inv.params[1], APIParam) + return [inv.params[1].api_full_name] + + @staticmethod + def _get_api_args_map_to_check(inv): + return None + + @staticmethod + def online_check( + check_relation_first: bool, + inv: Invariant, + trace_record: dict, + checker_data: Checker_data, + ): + if trace_record["type"] != TraceLineType.FUNC_CALL_POST: + return OnlineCheckerResult( + trace=None, + invariant=inv, + check_passed=True, + ) + + assert inv.precondition is not None, "The precondition should not be None." + assert len(inv.params) == 3 + + input_param, api_param, output_param = inv.params + + assert isinstance(input_param, InputOutputParam) + assert isinstance(api_param, APIParam) + assert isinstance(output_param, InputOutputParam) + assert input_param.is_input + assert not output_param.is_input + + logger = logging.getLogger(__name__) + + func_name = trace_record["function"] + func_call_id = trace_record["func_call_id"] + process_id = trace_record["process_id"] + thread_id = trace_record["thread_id"] + ptname = (process_id, thread_id, func_name) + with checker_data.lock: + func_call_event = checker_data.pt_map[ptname][func_call_id] + func_pre_record = func_call_event.pre_record + [func_pre_record] = set_meta_vars_online([func_pre_record], checker_data) + + check_passed = True + + if inv.precondition.verify([func_pre_record], "pre_event", None): + + input_tensors = get_input_tensors(func_call_event) + output_tensors = get_returned_tensors(func_call_event) + + try: + input_value = input_param.get_value_from_list_of_tensors(input_tensors) + output_value = output_param.get_value_from_list_of_tensors( + output_tensors + ) + if input_value != output_value: + check_passed = False + except (IndexError, KeyError): + logger.warning( + f"Could not find the value to be checked in input or output tensors for the hypothesis {inv}, skipping this function call." + ) + + if not check_passed: + return OnlineCheckerResult( + trace=[func_pre_record], + invariant=inv, + check_passed=False, + ) + + return OnlineCheckerResult( + trace=None, + invariant=inv, + check_passed=True, + ) + @staticmethod def get_precondition_infer_keys_to_skip(hypothesis: Hypothesis) -> list[str]: return [] @@ -1273,6 +1440,109 @@ def static_check_all(trace, inv, check_relation_first): triggered=triggered, ) + @staticmethod + def _get_identifying_params(inv: Invariant) -> list[Param]: + return [inv.params[1]] + + @staticmethod + def _get_variables_to_check(inv): + return None + + @staticmethod + def _get_apis_to_check(inv: Invariant): + assert isinstance(inv.params[1], APIParam) + return [inv.params[1].api_full_name] + + @staticmethod + def _get_api_args_map_to_check(inv): + return None + + @staticmethod + def online_check( + check_relation_first: bool, + inv: Invariant, + trace_record: dict, + checker_data: Checker_data, + ): + # get the first param and the second param, the first param should be larger or equal to the second param + # the first param should be larger or equal to the second param + if trace_record["type"] != TraceLineType.FUNC_CALL_POST: + return OnlineCheckerResult( + trace=None, + invariant=inv, + check_passed=True, + ) + + assert inv.precondition is not None, "The precondition should not be None." + assert len(inv.params) == 3 + max_param, api_param, min_param = inv.params + assert isinstance(max_param, InputOutputParam) + assert isinstance(api_param, APIParam) + assert isinstance(min_param, InputOutputParam) + + if max_param.is_input: + assert not min_param.is_input + is_threshold_min = False + input_param = max_param + output_param = min_param + else: + assert min_param.is_input + is_threshold_min = True + input_param = min_param + output_param = max_param + + # get all function calls for the function + process_id = trace_record["process_id"] + thread_id = trace_record["thread_id"] + func_name = trace_record["function"] + func_id = trace_record["func_call_id"] + ptname = (process_id, thread_id, func_name) + with checker_data.lock: + func_call_event = checker_data.pt_map[ptname][func_id] + func_pre_record = func_call_event.pre_record + [func_pre_record] = set_meta_vars_online([func_pre_record], checker_data) + + assert not isinstance( + func_call_event, (FuncCallExceptionEvent, IncompleteFuncCallEvent) + ), "The function call event should not be an exception or incomplete." + + check_passed = True + + # check for precondition here + if inv.precondition.verify([func_pre_record], "pre_event", None): + check_passed = False + threshold_value = input_param.get_value_from_arguments( + Arguments( + func_call_event.args, + func_call_event.kwargs, + func_call_event.func_name, + consider_default_values=True, + ) + ) + output_value = output_param.get_value_from_list_of_tensors( + get_returned_tensors(func_call_event) + ) + + if is_threshold_min: + if output_value >= threshold_value: + check_passed = True + else: + if output_value <= threshold_value: + check_passed = True + + if not check_passed: + return OnlineCheckerResult( + trace=[func_pre_record], + invariant=inv, + check_passed=False, + ) + + return OnlineCheckerResult( + trace=None, + invariant=inv, + check_passed=True, + ) + @staticmethod def get_precondition_infer_keys_to_skip(hypothesis: Hypothesis) -> list[str]: return [] diff --git a/traincheck/invariant/contain_relation.py b/traincheck/invariant/contain_relation.py index ad17732b..7e81aee8 100644 --- a/traincheck/invariant/contain_relation.py +++ b/traincheck/invariant/contain_relation.py @@ -1,3 +1,4 @@ +import copy import logging import random import time @@ -17,6 +18,7 @@ FailedHypothesis, Hypothesis, Invariant, + OnlineCheckerResult, Param, Relation, VarNameParam, @@ -28,6 +30,12 @@ ) from traincheck.invariant.precondition import find_precondition from traincheck.invariant.symbolic_value import generalize_values +from traincheck.onlinechecker.utils import ( + Checker_data, + get_var_ids_unchanged_but_causally_related, + get_var_raw_event_before_time, + set_meta_vars_online, +) from traincheck.trace.trace import Trace from traincheck.trace.types import ( ALL_EVENT_TYPES, @@ -70,6 +78,21 @@ def can_func_be_bound_method( return True +def can_func_be_bound_method_online( + trace_record: dict, + checker_data: Checker_data, + func_name: str, + var_type: str, + attr_name: str, +) -> bool: + func_id = trace_record["func_call_id"] + if not get_var_ids_unchanged_but_causally_related( + func_id, var_type, attr_name, trace_record, checker_data + ): + return False + return True + + cache_events_scanner: dict[ str, list[FuncCallEvent | FuncCallExceptionEvent | VarChangeEvent] ] = {} @@ -1135,6 +1158,263 @@ def static_check_all( triggered=inv_triggered, ) + @staticmethod + def _get_identifying_params(inv: Invariant) -> list[Param]: + return [inv.params[0]] + + @staticmethod + def _get_variables_to_check(inv: Invariant): + param = inv.params[1] + if isinstance(param, VarTypeParam): + return [param.var_type] + elif isinstance(param, VarNameParam): + return [param.var_name] + elif isinstance(param, APIParam): + return None + + @staticmethod + def _get_apis_to_check(inv: Invariant): + assert isinstance(inv.params[0], APIParam) + if isinstance(inv.params[1], APIParam): + return [inv.params[0].api_full_name, inv.params[1].api_full_name] + return [inv.params[0].api_full_name] + + @staticmethod + def _get_api_args_map_to_check(inv): + return None + + @staticmethod + def online_check( + check_relation_first: bool, + inv: Invariant, + trace_record: dict, + checker_data: Checker_data, + ): + if ( + trace_record["type"] != TraceLineType.FUNC_CALL_POST + and trace_record["type"] != TraceLineType.FUNC_CALL_POST_EXCEPTION + ): + return OnlineCheckerResult( + trace=None, + invariant=inv, + check_passed=True, + ) + + assert ( + len(inv.params) == 2 + ), f"Expected 2 parameters for APIContainRelation, one for the parent function name, and one for the child event name: {inv.params[0].to_dict()}" + + parent_param, child_param = inv.params[0], inv.params[1] + assert isinstance( + parent_param, APIParam + ), "Expected the first parameter to be an APIParam" + assert isinstance( + child_param, (APIParam, VarTypeParam, VarNameParam) + ), "Expected the second parameter to be an APIParam or VarTypeParam (VarNameParam not supported yet)" + + process_id = trace_record["process_id"] + thread_id = trace_record["thread_id"] + func_name = trace_record["function"] + func_id = trace_record["func_call_id"] + ptname = (process_id, thread_id, func_name) + with checker_data.lock: + func_call_event = checker_data.pt_map[ptname][func_id] + if func_call_event.pre_record is None: + return OnlineCheckerResult( + trace=None, + invariant=inv, + check_passed=True, + ) + func_pre_record = func_call_event.pre_record + func_post_record = func_call_event.post_record + [func_pre_record] = set_meta_vars_online([func_pre_record], checker_data) + + pre_time = func_pre_record["time"] + post_time = func_post_record["time"] + + preconditions = inv.precondition + assert ( + preconditions is not None + ), "Expected the precondition to be set for the invariant" + + skip_var_unchanged_check = ( + VAR_GROUP_NAME not in preconditions.get_group_names() + or len(preconditions.get_group(VAR_GROUP_NAME)) == 0 + or preconditions.is_group_unconditional(VAR_GROUP_NAME) + ) + + if not skip_var_unchanged_check: + assert isinstance( + child_param, VarTypeParam + ), "Expected the child parameter to be a VarTypeParam" + if not can_func_be_bound_method_online( + trace_record, + checker_data, + func_name, + child_param.var_type, + child_param.attr_name, + ): + skip_var_unchanged_check = True + + var_unchanged_check_passed = True + found_expected_child_event = False + precondition_check_passed = True + + if isinstance(child_param, (VarTypeParam, VarNameParam)): + events = [] + with checker_data.lock: + if child_param.var_type in checker_data.type_map: + for varid in checker_data.type_map[child_param.var_type]: + if isinstance(child_param, VarNameParam): + if varid.var_name != child_param.var_name: + continue + attr_name = child_param.attr_name + elif isinstance(child_param, VarTypeParam): + attr_name = child_param.attr_name + for i in reversed( + range(1, len(checker_data.varid_map[varid][attr_name])) + ): + + change_time = checker_data.varid_map[varid][attr_name][ + i + ].liveness.start_time + if change_time <= pre_time: + break + if change_time > post_time: + continue + new_state = checker_data.varid_map[varid][attr_name][i] + old_state = checker_data.varid_map[varid][attr_name][i - 1] + if new_state.value == old_state.value: + continue + if new_state.liveness.end_time is None: + new_state = copy.deepcopy(new_state) + events.append((old_state, new_state)) + + if check_relation_first: + for event in events: + if child_param.check_event_match_online(event): + found_expected_child_event = True + break + + if not preconditions.verify([func_pre_record], PARENT_GROUP_NAME, None): + precondition_check_passed = False + + else: + if preconditions.verify([func_pre_record], PARENT_GROUP_NAME, None): + for event in events: + if child_param.check_event_match_online(event): + found_expected_child_event = True + break + else: + precondition_check_passed = False + + if not precondition_check_passed: + return OnlineCheckerResult( + trace=None, + invariant=inv, + check_passed=True, + ) + + if not found_expected_child_event: + return OnlineCheckerResult( + trace=[func_pre_record], + invariant=inv, + check_passed=False, + ) + + if not skip_var_unchanged_check: + assert isinstance( + child_param, VarTypeParam + ), "Expected the child parameter to be a VarTypeParam" + # get the unchanged vars that are causally related to the parent function + # TODO: this function called twice, can be cached + unchanged_var_ids = get_var_ids_unchanged_but_causally_related( + func_id, + child_param.var_type, + child_param.attr_name, + trace_record, + checker_data, + ) + assert ( + len(unchanged_var_ids) > 0 + ), f"Internal error: can_func_be_bound_method returned True but no unchanged vars found for the parent function: {func_name} at {func_id}: {pre_time} at {time.monotonic_ns()}" + # get the var change events for the unchanged vars + unchanged_var_states = [ + get_var_raw_event_before_time(var_id, pre_time, checker_data) + for var_id in unchanged_var_ids + ] + for unchanged_var_state in unchanged_var_states: + # verify that no precondition is met for the unchanged vars + # MARK: precondition 2 + with checker_data.lock: + [unchanged_var_state] = set_meta_vars_online( + [unchanged_var_state], checker_data + ) + if not preconditions.verify( + [unchanged_var_state], VAR_GROUP_NAME, None + ): + var_unchanged_check_passed = False + break + + if not var_unchanged_check_passed: + return OnlineCheckerResult( + trace=[func_pre_record], + invariant=inv, + check_passed=False, + ) + + elif isinstance(child_param, APIParam): + events = [] + api_full_name = child_param.api_full_name + child_event_ptname = (process_id, thread_id, api_full_name) + with checker_data.lock: + if child_event_ptname in checker_data.pt_map: + for child_event in checker_data.pt_map[child_event_ptname].values(): + if ( + child_event.pre_record is None + or child_event.post_record is None + ): + continue + child_pre_time = child_event.pre_record["time"] + child_post_time = child_event.post_record["time"] + if pre_time > child_pre_time: + continue + if child_post_time > post_time: + continue + events.append(child_event) + + check_passed = False + + if check_relation_first: + for event in events: + if child_param.check_event_match_online(event): + check_passed = True + break + + if not preconditions.verify([func_pre_record], PARENT_GROUP_NAME, None): + check_passed = True + else: + if preconditions.verify([func_pre_record], PARENT_GROUP_NAME, None): + for event in events: + if child_param.check_event_match_online(event): + check_passed = True + break + else: + check_passed = True + + if not check_passed: + return OnlineCheckerResult( + trace=[func_pre_record], + invariant=inv, + check_passed=False, + ) + + return OnlineCheckerResult( + trace=None, + invariant=inv, + check_passed=True, + ) + @staticmethod def get_precondition_infer_keys_to_skip(hypothesis: Hypothesis) -> list[str]: return [] diff --git a/traincheck/invariant/cover_relation.py b/traincheck/invariant/cover_relation.py index d8c2fac0..d4e7c2d9 100644 --- a/traincheck/invariant/cover_relation.py +++ b/traincheck/invariant/cover_relation.py @@ -4,6 +4,7 @@ from tqdm import tqdm +from traincheck.instrumentor.tracer import TraceLineType from traincheck.invariant.base_cls import ( APIParam, CheckerResult, @@ -13,14 +14,19 @@ GroupedPreconditions, Hypothesis, Invariant, + OnlineCheckerResult, + Param, Relation, ) from traincheck.invariant.lead_relation import ( check_same_level, + get_func_A_B_events, get_func_data_per_PT, get_func_names_to_deal_with, + get_post_func_event, ) from traincheck.invariant.precondition import find_precondition +from traincheck.onlinechecker.utils import Checker_data, set_meta_vars_online from traincheck.trace.trace import Trace from traincheck.trace.trace_pandas import TracePandas @@ -226,26 +232,81 @@ def generate_hypothesis(trace) -> list[Hypothesis]: ].negative_examples.add_example(example) continue - flag_A = None - for event in events_list: - if event["type"] != "function_call (pre)": - continue + # find all A and B events in the current process and thread + events_A_pre, events_A_post, events_B_pre, events_B_post = ( + get_func_A_B_events(events_list, func_A, func_B) + ) + + event_A_idx = 0 + event_B_idx = 0 + + pre_event_B_time = None + + for event_B_pre in events_B_pre[event_B_idx:]: + invocation_id = event_B_pre["func_call_id"] + example = Example() + example.add_group(EXP_GROUP_NAME, [event_B_pre]) + + if event_A_idx >= len(events_A_pre): + # If we have exhausted all A events, skip the rest of B events + break - if func_A == event["function"]: - flag_A = event["time"] + event_B_post = get_post_func_event(events_B_post, invocation_id) - if func_B == event["function"]: - example = Example() - example.add_group(EXP_GROUP_NAME, [event]) - if flag_A is None: + if pre_event_B_time is not None: + if event_B_pre["time"] <= pre_event_B_time: hypothesis_with_examples[ (func_A, func_B) ].negative_examples.add_example(example) - else: - hypothesis_with_examples[ - (func_A, func_B) - ].positive_examples.add_example(example) - flag_A = None # reset flag_A + + pre_event_B_time = event_B_post["time"] + event_B_idx += 1 + continue + + found_A_before_B = False + # First B post time <= A pre time <= A post time <= next B pre time + while event_A_idx < len(events_A_pre): + event_A_pre = events_A_pre[event_A_idx] + event_A_time = event_A_pre["time"] + + if event_A_time > event_B_pre["time"]: + break + + if pre_event_B_time is not None: + if event_A_time <= pre_event_B_time: + event_A_idx += 1 + continue + + A_invocation_id = event_A_pre["func_call_id"] + event_A_post = get_post_func_event( + events_A_post, A_invocation_id + ) + if event_A_post["time"] > event_B_pre["time"]: + event_A_idx += 1 + continue + + found_A_before_B = True + event_A_idx += 1 + break + + if found_A_before_B: + hypothesis_with_examples[ + (func_A, func_B) + ].positive_examples.add_example(example) + else: + hypothesis_with_examples[ + (func_A, func_B) + ].negative_examples.add_example(example) + + pre_event_B_time = event_B_post["time"] + event_B_idx += 1 + # add the rest of the A events as negative examples + for event_B_pre in events_B_pre[event_B_idx:]: + example = Example() + example.add_group(EXP_GROUP_NAME, [event_B_pre]) + hypothesis_with_examples[ + (func_A, func_B) + ].negative_examples.add_example(example) print("End adding examples") @@ -344,7 +405,8 @@ def collect_examples(trace, hypothesis): ), "Invariant parameters should be APIParam." function_pool_temp.append(func.api_full_name) - function_pool = list(set(function_pool).intersection(function_pool_temp)) + # function_pool = list(set(function_pool).intersection(function_pool_temp)) + function_pool = set(function_pool).intersection(function_pool_temp) if len(function_pool) == 0: print( @@ -353,7 +415,8 @@ def collect_examples(trace, hypothesis): return print("Starting collecting iteration...") - for i in tqdm(range(invariant_length - 1)): + # for i in tqdm(range(invariant_length - 1)): + for i in range(invariant_length - 1): param_A = inv.params[i] param_B = inv.params[i + 1] @@ -380,25 +443,79 @@ def collect_examples(trace, hypothesis): hypothesis.negative_examples.add_example(example) continue - # check - flag_A = None - for event in events_list: - if event["type"] != "function_call (pre)": - continue + # find all A and B events in the current process and thread + events_A_pre, events_A_post, events_B_pre, events_B_post = ( + get_func_A_B_events(events_list, func_A, func_B) + ) - if func_A == event["function"]: - flag_A = event["time"] + event_A_idx = 0 + event_B_idx = 0 - if func_B == event["function"]: - example = Example() - example.add_group(EXP_GROUP_NAME, [event]) - if flag_A is None: - hypothesis.negative_examples.add_example(example) - else: - hypothesis.positive_examples.add_example(example) - flag_A = None # reset flag_A + pre_event_B_time = None + + for event_B_pre in events_B_pre[event_B_idx:]: + invocation_id = event_B_pre["func_call_id"] + example = Example() + example.add_group(EXP_GROUP_NAME, [event_B_pre]) + + if event_A_idx >= len(events_A_pre): + # If we have exhausted all A events, skip the rest of B events + break + + event_B_post = get_post_func_event(events_B_post, invocation_id) - print("End collecting iteration...") + if pre_event_B_time is not None: + if event_B_pre["time"] <= pre_event_B_time: + hypothesis[(func_A, func_B)].negative_examples.add_example( + example + ) + + pre_event_B_time = event_B_post["time"] + event_B_idx += 1 + continue + + found_A_before_B = False + # First B post time <= A pre time <= A post time <= next B pre time + while event_A_idx < len(events_A_pre): + event_A_pre = events_A_pre[event_A_idx] + event_A_time = event_A_pre["time"] + + if event_A_time > event_B_pre["time"]: + break + + if pre_event_B_time is not None: + if event_A_time <= pre_event_B_time: + event_A_idx += 1 + continue + + A_invocation_id = event_A_pre["func_call_id"] + event_A_post = get_post_func_event( + events_A_post, A_invocation_id + ) + if event_A_post["time"] > event_B_pre["time"]: + event_A_idx += 1 + continue + + found_A_before_B = True + event_A_idx += 1 + break + + if found_A_before_B: + hypothesis[(func_A, func_B)].positive_examples.add_example( + example + ) + else: + hypothesis[(func_A, func_B)].negative_examples.add_example( + example + ) + + pre_event_B_time = event_B_post["time"] + event_B_idx += 1 + # add the rest of the A events as negative examples + for event_B_pre in events_B_pre[event_B_idx:]: + example = Example() + example.add_group(EXP_GROUP_NAME, [event_B_pre]) + hypothesis[(func_A, func_B)].negative_examples.add_example(example) @staticmethod def infer(trace: Trace) -> Tuple[List[Invariant], List[FailedHypothesis]]: @@ -620,12 +737,13 @@ def static_check_all( continue if func_A not in same_level_func[(process_id, thread_id)][func_B]: - # no A invoked, all B should be invalid + # all B invocations in this process and thread are negative examples + # directly find the first B and return the result for event in events_list: - if ( - event["type"] == "function_call (pre)" - and event["function"] == func_B - ): + if event["type"] != "function_call (pre)": + continue + + if func_B == event["function"]: if not inv.precondition.verify( [event], EXP_GROUP_NAME, trace ): @@ -638,34 +756,71 @@ def static_check_all( check_passed=False, triggered=True, ) + # if we have not returned in this branch, lets check the next process and thread continue - # check - unmatched_A_exist = False - for event in events_list: - if event["type"] != "function_call (pre)": - continue + events_A_pre, events_A_post, events_B_pre, events_B_post = ( + get_func_A_B_events(events_list, func_A, func_B) + ) - if func_A == event["function"]: - unmatched_A_exist = True + event_A_idx = 0 + event_B_idx = 0 - if func_B == event["function"]: - if not inv.precondition.verify([event], EXP_GROUP_NAME, trace): - continue + pre_event_B_time = None - inv_triggered = True - if unmatched_A_exist is False: - inv_triggered = True + for event_B_pre in events_B_pre[event_B_idx:]: + if not inv.precondition.verify( + [event_B_pre], EXP_GROUP_NAME, trace + ): + continue + + if pre_event_B_time is not None: + if event_B_pre["time"] <= pre_event_B_time: return CheckerResult( - trace=[event], + trace=[event_B_pre], invariant=inv, check_passed=False, triggered=True, ) - else: - unmatched_A_exist = False # consumed the last A, a new A should be found before the next B - # FIXME: triggered is always False for passing invariants + event_B_post = get_post_func_event( + events_B_post, event_B_pre["func_call_id"] + ) + + found_A_before_B = False + while event_A_idx < len(events_A_pre): + event_A_pre = events_A_pre[event_A_idx] + event_A_time = event_A_pre["time"] + + if event_A_time > event_B_pre["time"]: + break + + if pre_event_B_time is not None: + if event_A_time <= pre_event_B_time: + event_A_idx += 1 + continue + + A_invocation_id = event_A_pre["func_call_id"] + event_A_post = get_post_func_event( + events_A_post, A_invocation_id + ) + if event_A_post["time"] > event_B_pre["time"]: + event_A_idx += 1 + continue + + found_A_before_B = True + event_A_idx += 1 + break + + if not found_A_before_B: + return CheckerResult( + trace=[event_B_pre], + invariant=inv, + check_passed=False, + triggered=True, + ) + pre_event_B_time = event_B_post["time"] + return CheckerResult( trace=None, invariant=inv, @@ -673,6 +828,122 @@ def static_check_all( triggered=inv_triggered, ) + @staticmethod + def _get_identifying_params(inv: Invariant) -> list[Param]: + params = [] + for i in range(len(inv.params) - 1): + params.append(inv.params[i + 1]) + return params + + @staticmethod + def _get_variables_to_check(inv): + return None + + @staticmethod + def _get_apis_to_check(inv: Invariant): + api_name_list = [] + for param in inv.params: + assert isinstance(param, APIParam) + api_name_list.append(param.api_full_name) + return api_name_list + + @staticmethod + def _get_api_args_map_to_check(inv): + return None + + @staticmethod + def online_check( + check_relation_first: bool, + inv: Invariant, + trace_record: dict, + checker_data: Checker_data, + ): + if trace_record["type"] != TraceLineType.FUNC_CALL_PRE: + return OnlineCheckerResult( + trace=None, + invariant=inv, + check_passed=True, + ) + + assert inv.precondition is not None, "Invariant should have a precondition." + + checker_param = APIParam(trace_record["function"]) + cover_param = None + for i in range(len(inv.params)): + if inv.params[i] == checker_param: + if i == 0: + cover_param = None + break + cover_param = inv.params[i - 1] + break + + if cover_param is None: + return OnlineCheckerResult( + trace=None, + invariant=inv, + check_passed=True, + ) + + assert isinstance(cover_param, APIParam) + + process_id = trace_record["process_id"] + thread_id = trace_record["thread_id"] + func_name = trace_record["function"] + ptname = (process_id, thread_id, func_name) + + start_time = None + end_time = trace_record["time"] + + with checker_data.lock: + [trace_record] = set_meta_vars_online([trace_record], checker_data) + + if not inv.precondition.verify([trace_record], EXP_GROUP_NAME, None): + return OnlineCheckerResult( + trace=None, + invariant=inv, + check_passed=True, + ) + + with checker_data.lock: + for func_id, func_event in checker_data.pt_map[ptname].items(): + if func_event.post_record is None: + continue + time = func_event.post_record["time"] + if time >= end_time: + continue + if not inv.precondition.verify( + set_meta_vars_online([func_event.pre_record], checker_data), + EXP_GROUP_NAME, + None, + ): + continue + if start_time is None or time > start_time: + start_time = time + + if start_time is None: + start_time = 0 + + cover_func_name = cover_param.api_full_name + cover_ptname = (process_id, thread_id, cover_func_name) + with checker_data.lock: + if cover_ptname in checker_data.pt_map: + for func_id, func_event in checker_data.pt_map[cover_ptname].items(): + if func_event.pre_record is None or func_event.post_record is None: + continue + pre_time = func_event.pre_record["time"] + post_time = func_event.post_record["time"] + if pre_time >= start_time and post_time <= end_time: + return OnlineCheckerResult( + trace=None, + invariant=inv, + check_passed=True, + ) + return OnlineCheckerResult( + trace=[trace_record], + invariant=inv, + check_passed=False, + ) + @staticmethod def get_precondition_infer_keys_to_skip(hypothesis: Hypothesis) -> list[str]: return ["function"] diff --git a/traincheck/invariant/lead_relation.py b/traincheck/invariant/lead_relation.py index 641f3e2a..27dc427d 100644 --- a/traincheck/invariant/lead_relation.py +++ b/traincheck/invariant/lead_relation.py @@ -4,6 +4,7 @@ from tqdm import tqdm +from traincheck.instrumentor.tracer import TraceLineType from traincheck.invariant.base_cls import ( APIParam, CheckerResult, @@ -13,9 +14,12 @@ GroupedPreconditions, Hypothesis, Invariant, + OnlineCheckerResult, + Param, Relation, ) from traincheck.invariant.precondition import find_precondition +from traincheck.onlinechecker.utils import Checker_data, set_meta_vars_online from traincheck.trace.trace import Trace from traincheck.trace.trace_pandas import TracePandas @@ -111,10 +115,8 @@ def get_func_data_per_PT(trace: Trace, function_pool: Iterable[str]): function_id_map: Dict[Tuple[str, str], Dict[str, List[str]]] = ( {} ) # map from (process_id, thread_id) to function name to function call ids - listed_events: Dict[Tuple[str, str], List[dict[str, Any]]] = ( - {} - ) # map from (process_id, thread_id) to all events - # for all func_ids, get their corresponding events + listed_events: Dict[Tuple[str, str], List[dict[str, Any]]] = {} + events = trace.events filtered_events = events[events["function"].isin(function_pool)] @@ -165,6 +167,39 @@ def get_func_data_per_PT(trace: Trace, function_pool: Iterable[str]): return function_times, function_id_map, listed_events +def get_func_A_B_events(events_list: List[dict[str, Any]], func_A: str, func_B: str): + events_A = [event for event in events_list if event["function"] == func_A] + events_A_pre = [ + event for event in events_A if event["type"] == "function_call (pre)" + ] + events_A_post = [ + event + for event in events_A + if event["type"] == "function_call (post)" + or event["type"] == "function_call (post) (exception)" + ] + events_B = [event for event in events_list if event["function"] == func_B] + events_B_pre = [ + event for event in events_B if event["type"] == "function_call (pre)" + ] + events_B_post = [ + event + for event in events_B + if event["type"] == "function_call (post)" + or event["type"] == "function_call (post) (exception)" + ] + return (events_A_pre, events_A_post, events_B_pre, events_B_post) + + +def get_post_func_event(events_list: List[dict[str, Any]], func_call_id: str): + event_posts = [ + event for event in events_list if event["func_call_id"] == func_call_id + ] + assert event_posts is not None, "Post event not found" + event_post = event_posts[0] + return event_post + + def is_complete_subgraph( path: List[APIParam], new_node: APIParam, graph: Dict[APIParam, List[APIParam]] ) -> bool: @@ -198,9 +233,6 @@ def merge_relations(pairs: List[Tuple[APIParam, APIParam]]) -> List[List[APIPara paths: List[List[APIParam]] = [] - def is_subset(path1: List[APIParam], path2: List[APIParam]) -> bool: - return set(path1).issubset(set(path2)) - def add_path(new_path: List[APIParam]) -> None: nonlocal paths # for existing_path in paths[:]: @@ -367,36 +399,117 @@ def generate_hypothesis(trace) -> list[Hypothesis]: ].negative_examples.add_example(example) continue - time_last_unmatched_A = None - last_pre_record_A = None + # find all A and B events in the current process and thread + events_A_pre, events_A_post, events_B_pre, events_B_post = ( + get_func_A_B_events(events_list, func_A, func_B) + ) + + event_A_idx = 0 + event_B_idx = 0 + + pre_event_A_idx = None + pre_event_A_time = None + last_example = None - hypothesis = hypothesis_with_examples[(func_A, func_B)] - for event in events_list: - if event["type"] != "function_call (pre)": + + for event_A_pre in events_A_pre: + invocation_id = event_A_pre["func_call_id"] + event_A_post = get_post_func_event(events_A_post, invocation_id) + pre_event_A_idx = event_A_idx + pre_event_A_time = event_A_post["time"] + event_A_idx += 1 + example = Example() + example.add_group(EXP_GROUP_NAME, [event_A_pre]) + last_example = example + break + + assert pre_event_A_idx is not None + assert pre_event_A_time is not None + assert last_example is not None + + if event_A_idx >= len(events_A_pre): + max_time = events_B_post["time"].max() + if pre_event_A_time <= max_time: + hypothesis_with_examples[ + (func_A, func_B) + ].positive_examples.add_example(last_example) + else: + hypothesis_with_examples[ + (func_A, func_B) + ].negative_examples.add_example(last_example) + + continue + + for event_A_pre in events_A_pre[event_A_idx:]: + invocation_id = event_A_pre["func_call_id"] + example = Example() + example.add_group(EXP_GROUP_NAME, [event_A_pre]) + + if event_B_idx >= len(events_B_pre): + # If we have exhausted all B events, skip the rest of A events + break + + event_A_post = get_post_func_event(events_A_post, invocation_id) + + if event_A_pre["time"] <= pre_event_A_time: + if last_example is not None: + hypothesis_with_examples[ + (func_A, func_B) + ].negative_examples.add_example(last_example) + + pre_event_A_idx = event_A_idx + pre_event_A_time = event_A_post["time"] + event_A_idx += 1 + last_example = example continue - if func_A == event["function"]: - if time_last_unmatched_A: - # the last A has not been followed by a B, a negative example: - assert last_example - hypothesis.negative_examples.add_example(last_example) + found_B_after_A = False + # First A post time <= B pre time <= B post time <= next A pre time + while event_B_idx < len(events_B_pre): + event_B_pre = events_B_pre[event_B_idx] + event_B_time = event_B_pre["time"] - time_last_unmatched_A = event["time"] - last_pre_record_A = event - last_example = Example() - last_example.add_group(EXP_GROUP_NAME, [last_pre_record_A]) + if event_B_time > event_A_pre["time"]: + break - if func_B == event["function"]: - if time_last_unmatched_A: - assert ( - last_example - ), "Raising an alarm for an A without B, but A's record is None, likely a bug" - hypothesis.positive_examples.add_example(last_example) - time_last_unmatched_A = None + if event_B_time <= pre_event_A_time: + event_B_idx += 1 + continue - if time_last_unmatched_A is not None: - assert last_example - hypothesis.negative_examples.add_example(last_example) + B_invocation_id = event_B_pre["func_call_id"] + event_B_post = get_post_func_event( + events_B_post, B_invocation_id + ) + if event_B_post["time"] > event_A_pre["time"]: + event_B_idx += 1 + continue + + found_B_after_A = True + event_B_idx += 1 + break + + if last_example is not None: + if found_B_after_A: + # Check if there's a B event after the current A event + hypothesis_with_examples[ + (func_A, func_B) + ].positive_examples.add_example(last_example) + else: + hypothesis_with_examples[ + (func_A, func_B) + ].negative_examples.add_example(last_example) + + pre_event_A_idx = event_A_idx + pre_event_A_time = event_A_post["time"] + event_A_idx += 1 + last_example = example + # add the rest of the A events as negative examples + for event_A_pre in events_A_pre[event_A_idx:]: + example = Example() + example.add_group(EXP_GROUP_NAME, [event_A_pre]) + hypothesis_with_examples[ + (func_A, func_B) + ].negative_examples.add_example(example) print("End adding examples") @@ -534,34 +647,113 @@ def collect_examples(trace, hypothesis): hypothesis.negative_examples.add_example(last_example) continue - time_last_unmatched_A = None - last_pre_record_A = None + # find all A and B events in the current process and thread + events_A_pre, events_A_post, events_B_pre, events_B_post = ( + get_func_A_B_events(events_list, func_A, func_B) + ) + # print(f"Found {len(events_A_pre)} A events and {len(events_B_pre)} B events") + + event_A_idx = 0 + event_B_idx = 0 + + pre_event_A_idx = None + pre_event_A_time = None + last_example = None - for event in events_list: - if event["type"] != "function_call (pre)": + + for event_A_pre in events_A_pre: + invocation_id = event_A_pre["func_call_id"] + event_A_post = get_post_func_event(events_A_post, invocation_id) + pre_event_A_idx = event_A_idx + pre_event_A_time = event_A_post["time"] + event_A_idx += 1 + example = Example() + example.add_group(EXP_GROUP_NAME, [event_A_pre]) + last_example = example + break + + if event_A_idx >= len(events_A_pre): + max_time = events_B_post["time"].max() + if pre_event_A_time <= max_time: + hypothesis[(func_A, func_B)].positive_examples.add_example( + last_example + ) + else: + hypothesis[(func_A, func_B)].negative_examples.add_example( + last_example + ) + + continue + + assert pre_event_A_idx is not None + assert pre_event_A_time is not None + + for event_A_pre in events_A_pre[event_A_idx:]: + invocation_id = event_A_pre["func_call_id"] + example = Example() + example.add_group(EXP_GROUP_NAME, [event_A_pre]) + + if event_B_idx >= len(events_B_pre): + # If we have exhausted all B events, skip the rest of A events + break + + event_A_post = get_post_func_event(events_A_post, invocation_id) + + if event_A_pre["time"] <= pre_event_A_time: + hypothesis[(func_A, func_B)].negative_examples.add_example( + last_example + ) + + pre_event_A_idx = event_A_idx + pre_event_A_time = event_A_post["time"] + event_A_idx += 1 + last_example = example continue - if func_A == event["function"]: - if time_last_unmatched_A: - # the last A has not been followed by a B, a negative example: - assert last_example - hypothesis.negative_examples.add_example(last_example) + found_B_after_A = False + # First A post time <= B pre time <= B post time <= next A pre time + while event_B_idx < len(events_B_pre): + event_B_pre = events_B_pre[event_B_idx] + event_B_time = event_B_pre["time"] - time_last_unmatched_A = event["time"] - last_pre_record_A = event - last_example = Example() - last_example.add_group(EXP_GROUP_NAME, [last_pre_record_A]) + if event_B_time > event_A_pre["time"]: + break - if func_B == event["function"]: - if time_last_unmatched_A: - assert ( - last_example - ), "Raising an alarm for an A without B, but A's record is None, likely a bug" - hypothesis.positive_examples.add_example(last_example) - time_last_unmatched_A = None + if event_B_time <= pre_event_A_time: + event_B_idx += 1 + continue - if time_last_unmatched_A is not None: - hypothesis.negative_examples.add_example(last_example) + B_invocation_id = event_B_pre["func_call_id"] + event_B_post = get_post_func_event( + events_B_post, B_invocation_id + ) + if event_B_post["time"] > event_A_pre["time"]: + event_B_idx += 1 + continue + + found_B_after_A = True + event_B_idx += 1 + break + + if found_B_after_A: + # Check if there's a B event after the current A event + hypothesis[(func_A, func_B)].positive_examples.add_example( + last_example + ) + else: + hypothesis[(func_A, func_B)].negative_examples.add_example( + last_example + ) + + pre_event_A_idx = event_A_idx + pre_event_A_time = event_A_post["time"] + event_A_idx += 1 + last_example = example + # add the rest of the A events as negative examples + for event_A_pre in events_A_pre[event_A_idx:]: + example = Example() + example.add_group(EXP_GROUP_NAME, [event_A_pre]) + hypothesis[(func_A, func_B)].negative_examples.add_example(example) @staticmethod def infer(trace: Trace) -> Tuple[List[Invariant], List[FailedHypothesis]]: @@ -805,36 +997,84 @@ def static_check_all( # if we have not returned in this branch, lets check the next process and thread continue - # check - has_B_showup_for_last_A = True # initialize the flag to True - last_A_pre_record = None - for event in events_list: + events_A_pre, events_A_post, events_B_pre, events_B_post = ( + get_func_A_B_events(events_list, func_A, func_B) + ) - if event["type"] != "function_call (pre)": - continue + event_A_idx = 0 + event_B_idx = 0 - if func_A == event["function"]: - if not inv.precondition.verify([event], EXP_GROUP_NAME, trace): - continue + pre_event_A = None + pre_event_A_time = None - inv_triggered = True + for event_A_pre in events_A_pre: + if not inv.precondition.verify( + [event_A_pre], EXP_GROUP_NAME, trace + ): + event_A_idx += 1 + continue + inv_triggered = True + invocation_id = event_A_pre["func_call_id"] + event_A_post = get_post_func_event(events_A_post, invocation_id) + pre_event_A = event_A_pre + pre_event_A_time = event_A_post["time"] + event_A_idx += 1 + break + + for event_A_pre in events_A_pre[event_A_idx:]: + if not inv.precondition.verify( + [event_A_pre], EXP_GROUP_NAME, trace + ): + continue - if has_B_showup_for_last_A: - # check passed for the last A, reset the flag - has_B_showup_for_last_A = False - last_A_pre_record = event - continue - else: - # we encountered an new A, but the last A has not been followed by a B - assert last_A_pre_record is not None + if pre_event_A_time is not None: + if event_A_pre["time"] <= pre_event_A_time: + assert pre_event_A is not None return CheckerResult( - trace=[last_A_pre_record], + trace=[pre_event_A], invariant=inv, check_passed=False, triggered=True, ) - if func_B == event["function"]: - has_B_showup_for_last_A = True + + event_A_post = get_post_func_event( + events_A_post, event_A_pre["func_call_id"] + ) + + found_B_after_A = False + while event_B_idx < len(events_B_pre): + event_B_pre = events_B_pre[event_B_idx] + event_B_time = event_B_pre["time"] + + if event_B_time > event_A_pre["time"]: + break + + if event_B_time <= pre_event_A_time: + event_B_idx += 1 + continue + + B_invocation_id = event_B_pre["func_call_id"] + event_B_post = get_post_func_event( + events_B_post, B_invocation_id + ) + if event_B_post["time"] > event_A_pre["time"]: + event_B_idx += 1 + continue + + found_B_after_A = True + event_B_idx += 1 + break + + if not found_B_after_A: + assert pre_event_A is not None + return CheckerResult( + trace=[pre_event_A], + invariant=inv, + check_passed=False, + triggered=True, + ) + pre_event_A_time = event_A_post["time"] + pre_event_A = event_A_pre return CheckerResult( trace=None, @@ -843,6 +1083,126 @@ def static_check_all( triggered=inv_triggered, ) + @staticmethod + def _get_identifying_params(inv: Invariant) -> list[Param]: + params = [] + for i in range(len(inv.params) - 1): + params.append(inv.params[i]) + return params + + @staticmethod + def _get_variables_to_check(inv): + return None + + @staticmethod + def _get_apis_to_check(inv: Invariant): + api_name_list = [] + for param in inv.params: + assert isinstance(param, APIParam) + api_name_list.append(param.api_full_name) + return api_name_list + + @staticmethod + def _get_api_args_map_to_check(inv): + return None + + @staticmethod + def online_check( + check_relation_first: bool, + inv: Invariant, + trace_record: dict, + checker_data: Checker_data, + ): + if trace_record["type"] != TraceLineType.FUNC_CALL_PRE: + return OnlineCheckerResult( + trace=None, + invariant=inv, + check_passed=True, + ) + + assert inv.precondition is not None, "Invariant should have a precondition." + + checker_param = APIParam(trace_record["function"]) + lead_param = None + for i in range(len(inv.params)): + if inv.params[i] == checker_param: + if i == len(inv.params) - 1: + lead_param = None + break + lead_param = inv.params[i + 1] + break + if lead_param is None: + return OnlineCheckerResult( + trace=None, + invariant=inv, + check_passed=True, + ) + + assert isinstance(lead_param, APIParam) + + process_id = trace_record["process_id"] + thread_id = trace_record["thread_id"] + func_name = trace_record["function"] + ptname = (process_id, thread_id, func_name) + + start_time = None + end_time = trace_record["time"] + + with checker_data.lock: + [trace_record] = set_meta_vars_online([trace_record], checker_data) + + if not inv.precondition.verify([trace_record], EXP_GROUP_NAME, None): + return OnlineCheckerResult( + trace=None, + invariant=inv, + check_passed=True, + ) + + with checker_data.lock: + for func_id, func_event in checker_data.pt_map[ptname].items(): + if func_event.post_record is None: + continue + time = func_event.post_record["time"] + if time >= end_time: + continue + if not inv.precondition.verify( + set_meta_vars_online([func_event.pre_record], checker_data), + EXP_GROUP_NAME, + None, + ): + continue + if start_time is None or time > start_time: + start_time = time + + if start_time is None: + return OnlineCheckerResult( + trace=None, + invariant=inv, + check_passed=True, + ) + + lead_func_name = lead_param.api_full_name + lead_ptname = (process_id, thread_id, lead_func_name) + with checker_data.lock: + if lead_ptname in checker_data.pt_map: + for func_id, func_event in checker_data.pt_map[lead_ptname].items(): + if func_event.pre_record is None or func_event.post_record is None: + continue + pre_time = func_event.pre_record["time"] + post_time = func_event.post_record["time"] + if pre_time >= start_time and post_time <= end_time: + return OnlineCheckerResult( + trace=None, + invariant=inv, + check_passed=True, + ) + + return OnlineCheckerResult( + trace=[trace_record], + invariant=inv, + check_passed=False, + ) + @staticmethod def get_precondition_infer_keys_to_skip(hypothesis: Hypothesis) -> list[str]: return ["function"] diff --git a/traincheck/onlinechecker/__init__.py b/traincheck/onlinechecker/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/traincheck/onlinechecker/streamhandler_filesystem.py b/traincheck/onlinechecker/streamhandler_filesystem.py new file mode 100644 index 00000000..310fc70e --- /dev/null +++ b/traincheck/onlinechecker/streamhandler_filesystem.py @@ -0,0 +1,337 @@ +import json +import logging +import os +import re +import time + +from watchdog.events import FileSystemEventHandler +from watchdog.observers.polling import PollingObserver + +from traincheck.config import config +from traincheck.instrumentor.tracer import TraceLineType +from traincheck.instrumentor.types import PTID +from traincheck.trace.types import AttrState, ContextManagerState, Liveness, VarInstId +from traincheck.trace.utils import ( + BindedFuncInput, + bind_args_kwargs_to_signature, + flatten_dict, + load_signature_from_class_method_name, + replace_none_with_md_none, +) +from traincheck.utils import safe_isnan + +from .utils import Checker_data, OnlineFuncCallEvent + + +class StreamLogHandler(FileSystemEventHandler): + """A file system handler to monitor the trace log file changes.""" + + def __init__(self, file_path, checker_data: Checker_data): + self.file_path = file_path + self.fp = open(file_path, "r") + + self.queue = checker_data.check_queue + + self.varid_map = checker_data.varid_map + self.type_map = checker_data.type_map + self.pt_map = checker_data.pt_map + self.process_to_vars = checker_data.process_to_vars + self.args_map = checker_data.args_map + + self.context_map = checker_data.context_map + self.init_map = checker_data.init_map + + self.needed_vars = checker_data.needed_vars + self.needed_apis = checker_data.needed_apis + self._get_api_args_map_to_check = checker_data._get_api_args_map_to_check + + self.min_read_time = checker_data.min_read_time + self.lock = checker_data.lock + self.cond = checker_data.cond + self.checker_data = checker_data + + logger = logging.getLogger(__name__) + self.logger = logger + + self._save_initial_content() + + self.fp.seek(0, 2) + + def _save_initial_content(self): + self.logger.info(f"Processing initial content from {self.file_path}") + self.fp.seek(0) + lines = self.fp.readlines() + if not lines: + return + + self._handle_line(lines) + self.logger.info(f"Initial content from {self.file_path} processed.") + + def on_modified(self, event): + if os.path.abspath(event.src_path) != os.path.abspath(self.file_path): + return + self.logger.debug(f"File {self.file_path} modified at {time.monotonic_ns()}") + new_lines = self.fp.readlines() + if new_lines: + self._handle_line(new_lines) + + def _handle_line(self, lines): + for line in lines: + trace_record = None + try: + flat_dict = flatten_dict( + json.loads(line, object_hook=replace_none_with_md_none), + skip_fields=["args", "kwargs", "return_values"], + ) + trace_record = flat_dict + self._set_maps(trace_record) + self.queue.put(trace_record) + + except Exception as e: + self.logger.error( + f"Error processing line in {self.file_path}: {e}. Line content: {line}" + ) + continue + + def _set_maps(self, trace_record): + """Set the variable map and function call map based on the trace record.""" + if "var_name" in trace_record and trace_record["var_name"] is not None: + self._set_var_map(trace_record) + elif ( + "func_call_id" in trace_record and trace_record["func_call_id"] is not None + ): + self._set_func_map(trace_record) + + self._set_read_time(trace_record) + + def _set_var_map(self, trace_record): + with self.lock: + var_name = trace_record["var_name"] + var_type = trace_record["var_type"] + if var_name in self.needed_vars or var_type in self.needed_vars: + varid = VarInstId( + trace_record["process_id"], + trace_record["var_name"], + trace_record["var_type"], + ) + if varid not in self.varid_map: + self.varid_map[varid] = {} + + if varid.process_id not in self.process_to_vars: + self.process_to_vars[varid.process_id] = set() + + self.process_to_vars[varid.process_id].add(varid) + + for attr_name, value in trace_record.items(): + if value is None: + continue + + if attr_name.startswith(config.VAR_ATTR_PREFIX): + attr_name = attr_name[len(config.VAR_ATTR_PREFIX) :] + else: + continue + + from traincheck.invariant.base_cls import make_hashable + + curr_value = make_hashable(value) + if any( + [ + re.match(pattern, attr_name) is not None + for pattern in config.PROP_ATTR_PATTERNS + ] + ): + continue + + if attr_name not in self.varid_map[varid]: + self.varid_map[varid][attr_name] = [] + else: + self.varid_map[varid][attr_name][-1].liveness.end_time = ( + trace_record["time"] + ) + + self.varid_map[varid][attr_name].append( + AttrState( + curr_value, + Liveness(trace_record["time"], None), + [trace_record], + ) + ) + + if trace_record["var_type"] is not None: + if trace_record["var_type"] not in self.type_map: + self.type_map[trace_record["var_type"]] = set() + self.type_map[trace_record["var_type"]].add(varid) + + def _set_func_map(self, trace_record): + with self.lock: + function_name = trace_record["function"] + process_id = trace_record["process_id"] + thread_id = trace_record["thread_id"] + ptid = (process_id, thread_id) + func_call_id = trace_record["func_call_id"] + ptname = (process_id, thread_id, function_name) + trace_type = trace_record["type"] + if function_name in self.needed_apis: + if ptname not in self.pt_map: + self.pt_map[ptname] = {} + if func_call_id not in self.pt_map[ptname]: + # TODO: check whether dict is necessary here, can be a list? + self.pt_map[ptname][func_call_id] = OnlineFuncCallEvent( + function_name + ) + if trace_type == TraceLineType.FUNC_CALL_PRE: + self.pt_map[ptname][func_call_id].pre_record = trace_record + self.pt_map[ptname][func_call_id].args = trace_record["args"] + self.pt_map[ptname][func_call_id].kwargs = trace_record["kwargs"] + elif trace_type == TraceLineType.FUNC_CALL_POST: + self.pt_map[ptname][func_call_id].post_record = trace_record + self.pt_map[ptname][func_call_id].return_values = trace_record[ + "return_values" + ] + elif trace_type == TraceLineType.FUNC_CALL_POST_EXCEPTION: + self.pt_map[ptname][func_call_id].post_record = trace_record + self.pt_map[ptname][func_call_id].exception = trace_record[ + "exception" + ] + + if trace_type == TraceLineType.FUNC_CALL_PRE: + if function_name in self.checker_data._get_api_args_map_to_check: + if "args" in trace_record: + if "meta_vars.step" not in trace_record: + trace_record["meta_vars.step"] = -1 + step = trace_record["meta_vars.step"] + if function_name not in self.args_map: + self.args_map[function_name] = {} + if step not in self.args_map[function_name]: + self.args_map[function_name][step] = {} + if ptid not in self.args_map[function_name][step]: + self.args_map[function_name][step][ptid] = [] + self.args_map[function_name][step][ptid].append(trace_record) + + if ( + ".__enter__" in function_name + or ".__exit__" in function_name + or ".__init__" in function_name + ): + if ( + "torch.autograd.grad_mode" not in function_name + and "torch.autograd.profiler.record_function" not in function_name + ): + self._set_context_map(trace_record) + + def _set_context_map(self, trace_record): + function_name = trace_record["function"] + process_id = trace_record["process_id"] + thread_id = trace_record["thread_id"] + ptid = PTID(process_id, thread_id) + ptname = (process_id, thread_id, function_name) + trace_type = trace_record["type"] + if ".__init__" in function_name and trace_type == TraceLineType.FUNC_CALL_PRE: + context_manager_name = function_name.removesuffix(".__init__") + ptname = (process_id, thread_id, context_manager_name) + if ptname not in self.init_map: + self.init_map[ptname] = [] + self.init_map[ptname].append(trace_record) + + elif ( + ".__enter__" in function_name and trace_type == TraceLineType.FUNC_CALL_POST + ): + context_manager_name = function_name.removesuffix(".__enter__") + ptname = (process_id, thread_id, context_manager_name) + closest_init_record = None + closest_init_time = None + if ptname in self.init_map: + for init_record in reversed(self.init_map[ptname]): + if init_record["time"] < trace_record["time"]: + if ( + closest_init_time is None + or init_record["time"] > closest_init_time + ): + closest_init_time = init_record["time"] + closest_init_record = init_record + + start_time = trace_record["time"] + args = closest_init_record["args"] + kwargs = closest_init_record["kwargs"] + + if not safe_isnan(args): + signature = load_signature_from_class_method_name( + closest_init_record["function"] + ) + + binded_args_and_kwargs = bind_args_kwargs_to_signature( + args, kwargs, signature + ) + else: + # create an empty BindedFuncInput if args is NaN, as it indicates + # that we did not record the args and kwargs for this function call + binded_args_and_kwargs = BindedFuncInput({}) + + if ptid not in self.context_map: + self.context_map[ptid] = {} + if context_manager_name not in self.context_map[ptid]: + self.context_map[ptid][context_manager_name] = [] + self.context_map[ptid][context_manager_name].append( + ContextManagerState( + name=context_manager_name, + ptid=ptid, + liveness=Liveness(start_time, None), + input=binded_args_and_kwargs, + ) + ) + elif ".__exit__" in function_name and trace_type == TraceLineType.FUNC_CALL_PRE: + context_manager_name = function_name.removesuffix(".__exit__") + contextmanagerstate = None + if ptid in self.context_map: + if context_manager_name in self.context_map[ptid]: + for state in reversed(self.context_map[ptid][context_manager_name]): + if state.liveness.start_time < trace_record["time"]: + break + if state.liveness.end_time is not None: + break + contextmanagerstate = state + if contextmanagerstate is not None: + contextmanagerstate.liveness.end_time = trace_record["time"] + + def _set_read_time(self, trace_record): + with self.cond: + self.checker_data.read_time_map[self.file_path] = trace_record["time"] + recalc_needed = ( + self.checker_data.min_read_path == self.file_path + or self.checker_data.min_read_time is None + ) + if recalc_needed: + pre_min_read_time = self.checker_data.min_read_time + self.checker_data.min_read_path, self.checker_data.min_read_time = min( + self.checker_data.read_time_map.items(), default=(None, None) + ) + if pre_min_read_time != self.checker_data.min_read_time: + self.checker_data.cond.notify_all() + + +def run_stream_monitor(traces, trace_folders, checker_data: Checker_data): + """Run the stream monitor to watch the trace files and folders.""" + logger = logging.getLogger(__name__) + observer = PollingObserver() + handlers = [] + if traces is not None: + file_path = os.path.abspath(traces[0]) + handler = StreamLogHandler(file_path, checker_data) + handlers.append(handler) + watch_dir = os.path.dirname(file_path) + observer.schedule(handler, path=watch_dir, recursive=False) + logger.info(f"Watching: {file_path}") + + if trace_folders is not None: + for trace_folder in trace_folders: + for file in os.listdir(trace_folder): + if file.startswith("trace_") or file.endswith("proxy_log.json"): + file_path = os.path.join(trace_folder, file) + handler = StreamLogHandler(file_path, checker_data) + handlers.append(handler) + watch_dir = os.path.dirname(file_path) + observer.schedule(handler, path=watch_dir, recursive=False) + logger.info(f"Watching: {file_path}") + + observer.start() + return observer diff --git a/traincheck/onlinechecker/utils.py b/traincheck/onlinechecker/utils.py new file mode 100644 index 00000000..987f22a6 --- /dev/null +++ b/traincheck/onlinechecker/utils.py @@ -0,0 +1,293 @@ +import queue +import threading + +from traincheck.instrumentor.types import PTID +from traincheck.trace.types import FuncCallEvent, VarInstId + + +class Checker_data: + """Data structure for online checker threads. Holds the needed data and the queue for processing.""" + + def __init__(self, needed_data): + needed_vars, needed_apis, _get_api_args_map_to_check = needed_data + self.needed_vars = needed_vars + self.needed_apis = needed_apis + self._get_api_args_map_to_check = _get_api_args_map_to_check + + self.check_queue = queue.Queue() + self.varid_map = {} + self.type_map = {} + self.pt_map = {} + self.process_to_vars = {} + self.args_map = {} + + self.context_map = {} + self.init_map = {} + + self.read_time_map = {} + self.min_read_time = None + self.min_read_path = None + self.lock = threading.Lock() + self.cond = threading.Condition(self.lock) + + +class OnlineFuncCallEvent(FuncCallEvent): + """A function call event for online checking.""" + + def __init__(self, func_name): + self.func_name = func_name + self.pre_record = None + self.post_record = None + self.exception = None + + self.args = None + self.kwargs = None + self.return_values = None + + def get_traces(self): + return [self.pre_record, self.post_record] + + def __hash__(self) -> int: + return super().__hash__() + + def __eq__(self, other) -> bool: + return super().__eq__(other) + + +def get_var_ids_unchanged_but_causally_related( + func_call_id: str, + var_type: str, + attr_name: str, + trace_record: dict, + checker_data: Checker_data, +) -> list[VarInstId]: + """Find all variables that are causally related to a function call but not changed within the function call. + + Casually related vars: Variables are accessed or modified by the object that the function call is bound to. + """ + related_vars = get_causally_related_vars(func_call_id, trace_record, checker_data) + changed_vars = query_var_changes_within_func_call( + func_call_id, var_type, attr_name, trace_record, checker_data + ) + + related_vars_not_changed = [] + if var_type is not None: + related_vars = { + var_id for var_id in related_vars if var_id.var_type == var_type + } + changed_vars = [ + var_change + for var_change in changed_vars + if var_change[0].var_type == var_type + ] + if attr_name is not None: + changed_vars = [ + var_change for var_change in changed_vars if var_change[1] == attr_name + ] + + for var_id in related_vars: + if any([var_change[0] == var_id for var_change in changed_vars]): + continue + related_vars_not_changed.append(var_id) + return related_vars_not_changed + + +def get_causally_related_vars( + func_call_id, trace_record, checker_data +) -> set[VarInstId]: + """Find all variables that are causally related to a function call. + By causally related, we mean that the variables have been accessed or modified by the object (with another method) that the function call is made on. + """ + + process_id = trace_record["process_id"] + thread_id = trace_record["thread_id"] + func_name = trace_record["function"] + func_id = trace_record["func_call_id"] + ptname = (process_id, thread_id, func_name) + with checker_data.lock: + func_call_pre_event = checker_data.pt_map[ptname][func_id].pre_record + + if func_call_pre_event is None: + raise ValueError( + f"Function call pre-event not found for func_call_id: {func_call_id}" + ) + + assert func_call_pre_event[ + "is_bound_method" + ], f"Causal relation extraction is only supported for bound methods, got {func_call_pre_event['function']} which is not" + + obj_id = func_call_pre_event["obj_id"] + + causally_related_var_ids: set[VarInstId] = set() + + with checker_data.lock: + for _, calls in checker_data.pt_map.items(): + for call_id, record in calls.items(): + if ( + record.pre_record["obj_id"] == obj_id + and record.pre_record["time"] < func_call_pre_event["time"] + ): + assert ( + record.pre_record["process_id"] + == func_call_pre_event["process_id"] + ), "Related function call is on a different process." + assert ( + record.pre_record["thread_id"] + == func_call_pre_event["thread_id"] + ), "Related function call is on a different thread." + + for var_name, var_type in record.pre_record["proxy_obj_names"]: + if var_name == "" and var_type == "": + continue + causally_related_var_ids.add( + VarInstId( + record.pre_record["process_id"], var_name, var_type + ) + ) + + return causally_related_var_ids + + +def query_var_changes_within_func_call( + func_call_id: str, + var_type: str, + attr_name: str, + trace_record: dict, + checker_data: Checker_data, +) -> list: + """Extract all variable change events from the trace, within the duration of a specific function call.""" + process_id = trace_record["process_id"] + thread_id = trace_record["thread_id"] + func_name = trace_record["function"] + func_id = trace_record["func_call_id"] + ptname = (process_id, thread_id, func_name) + with checker_data.lock: + func_call_event = checker_data.pt_map[ptname][func_id] + pre_record = func_call_event.pre_record + + post_record = func_call_event.post_record + + start_time = pre_record["time"] + end_time = post_record["time"] + + return query_var_changes_within_time_and_process( + (start_time, end_time), + var_type, + attr_name, + trace_record["process_id"], + checker_data, + ) + + +def query_var_changes_within_time_and_process( + time_range: tuple[int | float, int | float], + var_type: str, + attr_name: str, + process_id: int, + checker_data: Checker_data, +) -> list: + """Extract all variable change events from the trace, within a specific time range and process.""" + events = [] + with checker_data.lock: + for varid in checker_data.type_map[var_type]: + for i in reversed(range(1, len(checker_data.varid_map[varid][attr_name]))): + + change_time = checker_data.varid_map[varid][attr_name][ + i + ].liveness.start_time + if change_time <= time_range[0]: + break + if change_time > time_range[1]: + continue + new_state = checker_data.varid_map[varid][attr_name][i] + old_state = checker_data.varid_map[varid][attr_name][i - 1] + if new_state.value == old_state.value: + continue + events.append((varid, attr_name)) + return events + + +def get_var_raw_event_before_time( + var_id: VarInstId, time: int, checker_data: Checker_data +) -> list[dict]: + """Get all original trace records of a variable before the specified time.""" + + raw_events = [] + with checker_data.lock: + for attr_name, records in checker_data.varid_map[var_id].items(): + for record in records: + if record.liveness.start_time < time: + raw_events.append(record.traces[-1]) + + return raw_events + + +def get_meta_vars_online( + time, + process_id, + thread_id, + checker_data, +): + ptid = PTID(process_id, thread_id) + active_context_managers = [] + meta_vars = {} + + if ptid not in checker_data.context_map: + return None + context_managers = checker_data.context_map[ptid] + for context_manager_name, context_manager_states in context_managers.items(): + for context_manager_state in reversed(context_manager_states): + if context_manager_state.liveness.start_time <= time and ( + context_manager_state.liveness.end_time is None + or context_manager_state.liveness.end_time >= time + ): + active_context_managers.append(context_manager_state) + + prefix = "context_managers" + for _, context_manager in enumerate(active_context_managers): + meta_vars[f"{prefix}.{context_manager.name}"] = context_manager.to_dict()[ + "input" + ] + + return meta_vars + + +# TODO: move set_meta_vars from online check part to set map part +def set_meta_vars_online(records: list, checker_data: Checker_data): + earliest_time = None + earliest_process_id = None + earliest_thread_id = None + for record in records: + if earliest_time is None or record["time"] < earliest_time: + earliest_time = record["time"] + earliest_process_id = record["process_id"] + earliest_thread_id = record["thread_id"] + meta_vars = get_meta_vars_online( + earliest_time, earliest_process_id, earliest_thread_id, checker_data + ) + + if meta_vars: + for key in meta_vars: + for i in range(len(records)): + records[i][f"meta_vars.{key}"] = meta_vars[key] + return records + + +# use for time analysis +# timing_info = {} +# lock = threading.Lock() + +# def profile_section(name): +# def decorator(func): +# def wrapper(*args, **kwargs): +# start = time.perf_counter() +# result = func(*args, **kwargs) +# end = time.perf_counter() +# duration = end - start +# with lock: +# if name not in timing_info: +# timing_info[name] = [] +# timing_info[name].append(duration) +# return result +# return wrapper +# return decorator