| 
						
						
							
								
							
						
						
					 | 
					 | 
					@ -13,10 +13,8 @@ else: | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					import cereal.messaging as messaging | 
					 | 
					 | 
					 | 
					import cereal.messaging as messaging | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					from collections import namedtuple | 
					 | 
					 | 
					 | 
					from collections import namedtuple | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					from tools.lib.logreader import LogReader | 
					 | 
					 | 
					 | 
					from tools.lib.logreader import LogReader | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					
 | 
					 | 
					 | 
					 | 
					from selfdrive.test.process_replay.test_processes import get_segment | 
				
			
			
				
				
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					ProcessConfig = namedtuple('ProcessConfig', ['proc_name', 'pub_sub', 'ignore', 'command', 'path']) | 
					 | 
					 | 
					 | 
					ProcessConfig = namedtuple('ProcessConfig', ['proc_name', 'pub_sub', 'ignore', 'command', 'path', 'segment', 'wait_for_response']) | 
				
			
			
				
				
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					
 | 
					 | 
					 | 
					 | 
					 | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					BASE_URL = "https://commadataci.blob.core.windows.net/openpilotci/" | 
					 | 
					 | 
					 | 
					 | 
				
			
			
		
	
		
		
	
		
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					
 | 
					 | 
					 | 
					 | 
					
 | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					CONFIGS = [ | 
					 | 
					 | 
					 | 
					CONFIGS = [ | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					  ProcessConfig( | 
					 | 
					 | 
					 | 
					  ProcessConfig( | 
				
			
			
		
	
	
		
		
			
				
					| 
						
						
						
							
								
							
						
					 | 
					 | 
					@ -25,28 +23,20 @@ CONFIGS = [ | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					      "ubloxRaw": ["ubloxGnss", "gpsLocationExternal"], | 
					 | 
					 | 
					 | 
					      "ubloxRaw": ["ubloxGnss", "gpsLocationExternal"], | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    }, | 
					 | 
					 | 
					 | 
					    }, | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    ignore=[], | 
					 | 
					 | 
					 | 
					    ignore=[], | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    command="./ubloxd & sleep 20; kill $!", | 
					 | 
					 | 
					 | 
					    command="./ubloxd", | 
				
			
			
				
				
			
		
	
		
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    path="../locationd", | 
					 | 
					 | 
					 | 
					    path="../locationd", | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					 | 
					 | 
					 | 
					 | 
					    segment="0375fdf7b1ce594d|2019-06-13--08-32-25--3", | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					 | 
					 | 
					 | 
					 | 
					    wait_for_response=True | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					  ), | 
					 | 
					 | 
					 | 
					  ), | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					] | 
					 | 
					 | 
					 | 
					] | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					
 | 
					 | 
					 | 
					 | 
					
 | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					class SimplePubMaster(): | 
					 | 
					 | 
					 | 
					 | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					  def __init__(self, services):  # pylint: disable=super-init-not-called | 
					 | 
					 | 
					 | 
					 | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    self.sock = {} | 
					 | 
					 | 
					 | 
					 | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    for s in services: | 
					 | 
					 | 
					 | 
					 | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					      self.sock[s] = messaging.pub_sock(s) | 
					 | 
					 | 
					 | 
					 | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					
 | 
					 | 
					 | 
					 | 
					 | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					  def send(self, s, dat): | 
					 | 
					 | 
					 | 
					 | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    # print(dat) | 
					 | 
					 | 
					 | 
					 | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    self.sock[s].send(dat.to_bytes()) | 
					 | 
					 | 
					 | 
					 | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					
 | 
					 | 
					 | 
					 | 
					 | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					class TestValgrind(unittest.TestCase): | 
					 | 
					 | 
					 | 
					class TestValgrind(unittest.TestCase): | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					
 | 
					 | 
					 | 
					 | 
					
 | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					  def valgrindlauncher(self, arg, cwd): | 
					 | 
					 | 
					 | 
					  def valgrindlauncher(self, arg, cwd): | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    os.chdir(cwd) | 
					 | 
					 | 
					 | 
					    os.chdir(cwd) | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					
 | 
					 | 
					 | 
					 | 
					
 | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    # Run valgrind on a process | 
					 | 
					 | 
					 | 
					    # Run valgrind on a process | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    command = "valgrind --leak-check=full " + arg | 
					 | 
					 | 
					 | 
					    command = "valgrind --leak-check=full " + arg + " & sleep 20; kill $!" | 
				
			
			
				
				
			
		
	
		
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					
 | 
					 | 
					 | 
					 | 
					
 | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) | 
					 | 
					 | 
					 | 
					    p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    _, err = p.communicate() | 
					 | 
					 | 
					 | 
					    _, err = p.communicate() | 
				
			
			
		
	
	
		
		
			
				
					| 
						
						
						
							
								
							
						
					 | 
					 | 
					@ -59,45 +49,45 @@ class TestValgrind(unittest.TestCase): | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    possibly_lost_amount = int(err_lost3.split(" ")[0]) | 
					 | 
					 | 
					 | 
					    possibly_lost_amount = int(err_lost3.split(" ")[0]) | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    if max(definitely_lost_amount, indirectly_lost_amount, possibly_lost_amount) > 0: | 
					 | 
					 | 
					 | 
					    if max(definitely_lost_amount, indirectly_lost_amount, possibly_lost_amount) > 0: | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					      self.leak = True | 
					 | 
					 | 
					 | 
					      self.leak = True | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					      print(err) | 
					 | 
					 | 
					 | 
					      print(definitely_lost_amount, indirectly_lost_amount, possibly_lost_amount) | 
				
			
			
				
				
			
		
	
		
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					 | 
					 | 
					 | 
					 | 
					      # print(err) | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					      return | 
					 | 
					 | 
					 | 
					      return | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    self.leak = False | 
					 | 
					 | 
					 | 
					    self.leak = False | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					
 | 
					 | 
					 | 
					 | 
					
 | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					  def replay_process(self, cfg, lr): | 
					 | 
					 | 
					 | 
					  def replay_process(self, config, logreader): | 
				
			
			
				
				
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    pub_sockets = [s for s in cfg.pub_sub.keys() if s != 'can']  # We dump data from logs here | 
					 | 
					 | 
					 | 
					    pub_sockets = [s for s in config.pub_sub.keys()]  # We dump data from logs here | 
				
			
			
				
				
			
		
	
		
		
	
		
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					 | 
					 | 
					 | 
					 | 
					    sub_sockets = [s for _, sub in config.pub_sub.items() for s in sub]  # We get responses here | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					 | 
					 | 
					 | 
					 | 
					    pm = messaging.PubMaster(pub_sockets) | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					 | 
					 | 
					 | 
					 | 
					    sm = messaging.SubMaster(sub_sockets) | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					
 | 
					 | 
					 | 
					 | 
					
 | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    pm = SimplePubMaster(pub_sockets) | 
					 | 
					 | 
					 | 
					 | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    print("Sorting logs") | 
					 | 
					 | 
					 | 
					    print("Sorting logs") | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    all_msgs = sorted(lr, key=lambda msg: msg.logMonoTime) | 
					 | 
					 | 
					 | 
					    all_msgs = sorted(logreader, key=lambda msg: msg.logMonoTime) | 
				
			
			
				
				
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    pub_msgs = [msg for msg in all_msgs if msg.which() in list(cfg.pub_sub.keys())] | 
					 | 
					 | 
					 | 
					    pub_msgs = [msg for msg in all_msgs if msg.which() in list(config.pub_sub.keys())] | 
				
			
			
				
				
			
		
	
		
		
	
		
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					
 | 
					 | 
					 | 
					 | 
					
 | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    thread = threading.Thread(target=self.valgrindlauncher, args=(cfg.command, cfg.path)) | 
					 | 
					 | 
					 | 
					    thread = threading.Thread(target=self.valgrindlauncher, args=(config.command, config.path)) | 
				
			
			
				
				
			
		
	
		
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    thread.daemon = True | 
					 | 
					 | 
					 | 
					    thread.daemon = True | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    thread.start() | 
					 | 
					 | 
					 | 
					    thread.start() | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					 | 
					 | 
					 | 
					 | 
					
 | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    time.sleep(5)  # We give the process time to start | 
					 | 
					 | 
					 | 
					    time.sleep(5)  # We give the process time to start | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    for msg in tqdm(pub_msgs): | 
					 | 
					 | 
					 | 
					    for msg in tqdm(pub_msgs): | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					      pm.send(msg.which(), msg.as_builder()) | 
					 | 
					 | 
					 | 
					      pm.send(msg.which(), msg.as_builder()) | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					
 | 
					 | 
					 | 
					 | 
					      if config.wait_for_response: | 
				
			
			
				
				
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					  def get_segment(self, segment_name, original=True): | 
					 | 
					 | 
					 | 
					        gotResponse = False | 
				
			
			
				
				
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    route_name, segment_num = segment_name.rsplit("--", 1) | 
					 | 
					 | 
					 | 
					        while not gotResponse: | 
				
			
			
				
				
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    if original: | 
					 | 
					 | 
					 | 
					          sm.update() | 
				
			
			
				
				
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					      rlog_url = BASE_URL + "%s/%s/rlog.bz2" % (route_name.replace("|", "/"), segment_num) | 
					 | 
					 | 
					 | 
					          for s in sub_sockets: | 
				
			
			
				
				
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    else: | 
					 | 
					 | 
					 | 
					            if sm.updated[s]: | 
				
			
			
				
				
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					      process_replay_dir = os.path.dirname(os.path.abspath(__file__)) | 
					 | 
					 | 
					 | 
					              gotResponse = True | 
				
			
			
				
				
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					      model_ref_commit = open(os.path.join(process_replay_dir, "model_ref_commit")).read().strip() | 
					 | 
					 | 
					 | 
					 | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					      rlog_url = BASE_URL + "%s/%s/rlog_%s.bz2" % (route_name.replace("|", "/"), segment_num, model_ref_commit) | 
					 | 
					 | 
					 | 
					 | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					
 | 
					 | 
					 | 
					 | 
					 | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    return rlog_url | 
					 | 
					 | 
					 | 
					 | 
				
			
			
		
	
		
		
	
		
		
	
		
		
	
		
		
	
		
		
	
		
		
	
		
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					
 | 
					 | 
					 | 
					 | 
					
 | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					  def test_config_0(self): | 
					 | 
					 | 
					 | 
					  def test_config_0(self): | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    cfg = CONFIGS[0] | 
					 | 
					 | 
					 | 
					    cfg = CONFIGS[0] | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					
 | 
					 | 
					 | 
					 | 
					
 | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    URL = self.get_segment("0375fdf7b1ce594d|2019-06-13--08-32-25--3") | 
					 | 
					 | 
					 | 
					    URL = cfg.segment | 
				
			
			
				
				
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    lr = LogReader(URL) | 
					 | 
					 | 
					 | 
					    lr = LogReader(get_segment(URL)) | 
				
			
			
				
				
			
		
	
		
		
	
		
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    self.replay_process(cfg, lr) | 
					 | 
					 | 
					 | 
					    self.replay_process(cfg, lr) | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    # Wait for the replay to complete | 
					 | 
					 | 
					 | 
					    # Wait for the replay to complete | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    time.sleep(30) | 
					 | 
					 | 
					 | 
					    time.sleep(30) | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					    assert not self.leak | 
					 | 
					 | 
					 | 
					    self.assertFalse(self.leak) | 
				
			
			
				
				
			
		
	
		
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					
 | 
					 | 
					 | 
					 | 
					
 | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					if __name__ == "__main__": | 
					 | 
					 | 
					 | 
					if __name__ == "__main__": | 
				
			
			
		
	
		
		
			
				
					
					 | 
					 | 
					 | 
					  unittest.main() | 
					 | 
					 | 
					 | 
					  unittest.main() | 
				
			
			
		
	
	
		
		
			
				
					| 
						
						
						
					 | 
					 | 
					
  |