Merge pull request #4 from jkrshnmenon/fix-sym-exc-timeout-perf

Refactored the watchdog mechanism

arbiter/master_chief/symbolic_execution.py

@@ -49,7 +49,8 @@ def __init__(self, sa, constrain, require_dd=None, verbose=False):
         self._stats_filename = 'UCSE.json'
         self._statistics['identified_functions'] = len(self._targets)
 
-        self._timeout_received = False
+        # Used to synchronize with the watchdog
+        self._watchdog_event = threading.Event()
 
         if len(self._targets) <= 0:
             raise FatalError("No targets for SymExec")
@@ -78,20 +79,11 @@ def _dump_stats(self):
             json.dump(self._statistics, f, indent=2)
 
     def _watchdog(self, timeout):
-        logger.debug("Watchdog started")
-        start_time = None
-        if self._timeout_received is True:
-            logger.debug("Timeout_received flag is set to True")
-            logger.debug("Watchdog setting timeout_received flag to False")
-            self._timeout_received = False
-        while not self._timeout_received:
-            cur_time = time.time()
-            if start_time is None or (cur_time - start_time) > timeout:
-                logger.debug("Watchdog setting alarm for %ds" % timeout)
-                signal.alarm(timeout)
-                start_time = time.time()
-            logger.debug("Watchdog sleeping for %ds" % (timeout/10))
-            time.sleep(timeout/10)
+        logger.debug(f"Watchdog started, waiting for {timeout}s")
+        # When we timeout, Event.wait will return False
+        if not self._watchdog_event.wait(timeout):
+            logger.debug(f"Watchdog timed out, sending SIG_ALARM to self")
+            os.kill(os.getpid(), signal.SIGALRM)
 
     def _hook_checkpoint(self, target, cname):
         cfunc = self.cfg.functions.get(cname)
@@ -274,23 +266,23 @@ def _explore_one(self, target, site, init_state):
         counter = 0
         start = time.time()
         logger.info("Starting exploration 0x%0x => 0x%0x" % (init_state.addr, site.bbl))
-        self._timeout_received = False
+        self._watchdog_event.clear()
+
         t = threading.Thread(target=self._watchdog, args=(300,))
         t.start()
-        # self._watchdog(300)
         try:
             pg.explore(find=sorted(block.instruction_addrs)[-1])
 
             if len(pg.found) == 0:
-                self._timeout_received = True
+                self._watchdog_event.set()
                 t.join()
                 raise angr.AngrAnalysisError("No paths found")
 
             if len(pg.active) == 0:
                 logger.debug("Found %d paths" % len(pg.found))
                 self._statistics[target.addr]['paths_found'] += len(pg.found)
                 for pp in pg.found:
-                    self._timeout_received = True
+                    self._watchdog_event.set()
                     if self._check_state(pg.found[0], site, target) is True:
                         logger.debug("Waiting for watchdog to join")
                         t.join()
@@ -304,16 +296,17 @@ def _explore_one(self, target, site, init_state):
                 self._statistics[target.addr]['exploration_time'] = int(end - start)
                 for pp in pg.found:
                     if self._check_state(pp, site, target):
-                        self._timeout_received = True
+                        self._watchdog_event.set()
                         logger.debug("Waiting for watchdog to join")
                         t.join()
                         return
                 pg.drop(stash='found')
-                signal.alarm(300)
+                # adamd: I don't know if this is necessary, the watchdog should be able to wake us up
+                #signal.alarm(300)
                 logger.debug("Wrapping up %d active paths" % len(pg.active))
                 pg.explore(find=sorted(block.instruction_addrs)[-1])
         except (TimeoutException, KeyboardInterrupt) as e:
-            self._timeout_received = True
+            self._watchdog_event.set()
             logger.debug("Waiting for watchdog to join")
             t.join()
             logger.debug("Got an exception")
@@ -327,7 +320,7 @@ def _explore_one(self, target, site, init_state):
         for pp in pg.found:
             self._check_state(pp, site, target)
 
-        self._timeout_received = True
+        self._watchdog_event.set()
         logger.debug("Waiting for watchdog to join")
         t.join()
         del init_state
@@ -443,7 +436,7 @@ def _execute_one(self, target, site):
                 try:
                     self._explore_one(target, site, x)
                 except angr.AngrAnalysisError as e:
-                    logger.error(e)
+                    logger.exeception(e)
                     continue
 
     def run_one(self, target):
@@ -452,10 +445,10 @@ def run_one(self, target):
                 self._statistics[target.addr] = {}
                 self._execute_one(target, target._nodes[x])
             except angr.AngrAnalysisError as e:
-                logger.error(e)
+                logger.exception(e)
                 continue
             except (KeyboardInterrupt, AssertionError, AttributeError) as e:
-                logger.error(e)
+                logger.exception(e)
                 continue
 
     def run_all(self):
@@ -636,10 +629,9 @@ def verify_one(self, report, start, block_dict):
         sat_states = []
 
         pg = self._project.factory.simulation_manager(init_state)
-        self._timeout_received = False
+        self._watchdog_event = threading.Event()
         t = threading.Thread(target=self._watchdog, args=(600,))
         t.start()
-        # signal.alarm(600)
         try:
             logger.info("Starting exploration to 0x%x" % first_target.addr)
             start_time = time.time()
@@ -651,10 +643,10 @@ def verify_one(self, report, start, block_dict):
             end = time.time()
             logger.debug("Found %d paths" % len(pg.found))
             self._statistics[first_target.addr][start]['exploring_callers'] = int(end - start_time)
-            self._timeout_received = True
+            self._watchdog_event.set()
             t.join()
 
-            self._timeout_received = False
+            self._watchdog_event.clear()
             t = threading.Thread(target=self._watchdog, args=(600,))
             t.start()
             start_time = time.time()
@@ -668,13 +660,13 @@ def verify_one(self, report, start, block_dict):
             self._statistics[first_target.addr][start]['reaching_sink'] = int(end - start_time)
             logger.debug("Found %d states" % len(final_states))
 
-            self._timeout_received = True
+            self._watchdog_event.set()
             t.join()
             for state in final_states:
                 if self._check_state(state, report.site, None, self._statistics[first_target.addr][start]) is True:
                     sat_states.append(state)
         except (TimeoutException, KeyboardInterrupt, AssertionError) as e:
-            self._timeout_received = True
+            self._watchdog_event.set()
             t.join()
             logger.debug("Got an exception")
             logger.error(e)
@@ -698,7 +690,7 @@ def verify(self, report, blocks):
                 if output is not None:
                     break
             except AssertionError as e:
-                self._timeout_received = True
+                self._watchdog_event.set()
                 logger.error(e)
                 continue