[docs]classThreading(ExplorationTechnique):""" Enable multithreading. This is only useful in paths where a lot of time is taken inside z3, doing constraint solving. This is because of python's GIL, which says that only one thread at a time may be executing python code. """
[docs]defstep(self,simgr,stash="active",error_list=None,target_stash=None,**kwargs):target_stash=target_stashorstashiferror_listisnotNone:raiseValueError("Can't pass error_list to step with threading enabled. Did you install threading twice?")l.info("Thread-stepping %s of %s",stash,simgr)forstateinsimgr.stashes[stash]:ifstateinself.queued:continue# construct new simgr with new lists# this means that threads won't trample each other's hook stacks# but can still negotiate over shared resourcestsimgr=simgr.copy()tsimgr._stashes={self.local_stash:[state]}tsimgr._errored=[]self.tasks.add(self.executor.submit(self.inner_step,state,tsimgr,target_stash=target_stash,**kwargs))self.queued.add(state)timeout=NonewhileTrue:done,self.tasks=concurrent.futures.wait(self.tasks,timeout=timeout,return_when=concurrent.futures.FIRST_COMPLETED)ifnotdone:breakfordone_futureindone:done_future:concurrent.futures.Futurestate,error_list,tsimgr=done_future.result()simgr.absorb(tsimgr)simgr.errored.extend(error_list)simgr.stashes[stash].remove(state)self.queued.remove(state)timeout=0returnsimgr
[docs]defsuccessors(self,simgr,state,engine=None,**kwargs):engine=engineorself.project.factory.default_engineifnotisinstance(engine,TLSMixin)andonce("tls_engine"):l.error("Using Threading exploration technique but your engine is not thread-safe.")l.error("Do you want to add the TLSMixin to your engine?")returnsimgr.successors(state,engine=engine,**kwargs)