from__future__importabsolute_importimporttypesfrompypy.tool.ansi_printimportansi_logfrompypy.tool.pairtypeimportpairfrompypy.tool.errorimport(format_blocked_annotation_error,AnnotatorError,gather_error,ErrorWrapper)frompypy.objspace.flow.modelimport(Variable,Constant,FunctionGraph,c_last_exception,checkgraph)frompypy.translatorimportsimplify,transformfrompypy.annotationimportmodelasannmodel,signaturefrompypy.annotation.bookkeeperimportBookkeeperimportpylog=py.log.Producer("annrpython")py.log.setconsumer("annrpython",ansi_log)FAIL=object()classRPythonAnnotator(object):"""Block annotator for RPython. See description in doc/translation.txt."""def__init__(self,translator=None,policy=None,bookkeeper=None):importpypy.rpython.ootypesystem.ooregistry# has side effectsimportpypy.rpython.extfuncregistry# has side effectsimportpypy.rlib.nonconst# has side effectsiftranslatorisNone:# interface for testsfrompypy.translator.translatorimportTranslationContexttranslator=TranslationContext()translator.annotator=selfself.translator=translatorself.pendingblocks={}# map {block: graph-containing-it}self.bindings={}# map Variables to SomeValuesself.annotated={}# set of blocks already seenself.added_blocks=None# see processblock() belowself.links_followed={}# set of links that have ever been followedself.notify={}# {block: {positions-to-reflow-from-when-done}}self.fixed_graphs={}# set of graphs not to annotate againself.blocked_blocks={}# set of {blocked_block: (graph, index)}# --- the following information is recorded for debugging ---self.blocked_graphs={}# set of graphs that have blocked blocks# --- end of debugging information ---self.frozen=FalseifpolicyisNone:frompypy.annotation.policyimportAnnotatorPolicyself.policy=AnnotatorPolicy()else:self.policy=policyifbookkeeperisNone:bookkeeper=Bookkeeper(self)self.bookkeeper=bookkeeperdef__getstate__(self):attrs="""translator pendingblocks bindings annotated links_followed notify bookkeeper frozen policy added_blocks""".split()ret=self.__dict__.copy()forkey,valueinret.items():ifkeynotinattrs:asserttype(value)isdict,("%r is not dict. please update %s.__getstate__"%(key,self.__class__.__name__))ret[key]={}returnret#___ convenience high-level interface __________________defbuild_types(self,function,input_arg_types,complete_now=True,main_entry_point=False):"""Recursively build annotations about the specific entry point."""assertisinstance(function,types.FunctionType),"fix that!"frompypy.annotation.policyimportAnnotatorPolicypolicy=AnnotatorPolicy()# make input arguments and set their typeargs_s=[self.typeannotation(t)fortininput_arg_types]# XXX hackannmodel.TLS.check_str_without_nul=(self.translator.config.translation.check_str_without_nul)flowgraph,inputcells=self.get_call_parameters(function,args_s,policy)ifnotisinstance(flowgraph,FunctionGraph):assertisinstance(flowgraph,annmodel.SomeObject)returnflowgraphifmain_entry_point:self.translator.entry_point_graph=flowgraphreturnself.build_graph_types(flowgraph,inputcells,complete_now=complete_now)defget_call_parameters(self,function,args_s,policy):desc=self.bookkeeper.getdesc(function)args=self.bookkeeper.build_args("simple_call",args_s[:])result=[]defschedule(graph,inputcells):result.append((graph,inputcells))returnannmodel.s_ImpossibleValueprevpolicy=self.policyself.policy=policyself.bookkeeper.enter(None)try:desc.pycall(schedule,args,annmodel.s_ImpossibleValue)finally:self.bookkeeper.leave()self.policy=prevpolicy[(graph,inputcells)]=resultreturngraph,inputcellsdefannotate_helper(self,function,args_s,policy=None):ifpolicyisNone:frompypy.annotation.policyimportAnnotatorPolicypolicy=AnnotatorPolicy()graph,inputcells=self.get_call_parameters(function,args_s,policy)self.build_graph_types(graph,inputcells,complete_now=False)self.complete_helpers(policy)returngraphdefcomplete_helpers(self,policy):saved=self.policy,self.added_blocksself.policy=policytry:self.added_blocks={}self.complete()# invoke annotation simplifications for the new blocksself.simplify(block_subset=self.added_blocks)finally:self.policy,self.added_blocks=saveddefbuild_graph_types(self,flowgraph,inputcells,complete_now=True):checkgraph(flowgraph)nbarg=len(flowgraph.getargs())iflen(inputcells)!=nbarg:raiseTypeError("%s expects %d args, got %d"%(flowgraph,nbarg,len(inputcells)))# register the entry pointself.addpendinggraph(flowgraph,inputcells)# recursively proceed until no more pending block is leftifcomplete_now:self.complete()returnself.binding(flowgraph.getreturnvar(),None)defgettype(self,variable):"""Return the known type of a control flow graph variable, defaulting to 'object'."""ifisinstance(variable,Constant):returntype(variable.value)elifisinstance(variable,Variable):cell=self.bindings.get(variable)ifcell:returncell.knowntypeelse:returnobjectelse:raiseTypeError,("Variable or Constant instance expected, ""got %r"%(variable,))defgetuserclassdefinitions(self):"""Return a list of ClassDefs."""returnself.bookkeeper.classdefs#___ medium-level interface ____________________________defaddpendinggraph(self,flowgraph,inputcells):self.addpendingblock(flowgraph,flowgraph.startblock,inputcells)defaddpendingblock(self,graph,block,cells):"""Register an entry point into block with the given input cells."""ifgraphinself.fixed_graphs:# special case for annotating/rtyping in several phases: calling# a graph that has already been rtyped. Safety-check the new# annotations that are passed in, and don't annotate the old# graph -- it's already low-level operations!fora,s_newarginzip(graph.getargs(),cells):s_oldarg=self.binding(a)assertannmodel.unionof(s_oldarg,s_newarg)==s_oldargelse:assertnotself.frozenforaincells:assertisinstance(a,annmodel.SomeObject)ifblocknotinself.annotated:self.bindinputargs(graph,block,cells)else:self.mergeinputargs(graph,block,cells)ifnotself.annotated[block]:self.pendingblocks[block]=graphdefcomplete(self):"""Process pending blocks until none is left."""whileTrue:whileself.pendingblocks:block,graph=self.pendingblocks.popitem()self.processblock(graph,block)self.policy.no_more_blocks_to_annotate(self)ifnotself.pendingblocks:break# finished# make sure that the return variables of all graphs is annotatedifself.added_blocksisnotNone:newgraphs=[self.annotated[block]forblockinself.added_blocks]newgraphs=dict.fromkeys(newgraphs)got_blocked_blocks=Falseinnewgraphselse:newgraphs=self.translator.graphs#all of themgot_blocked_blocks=Falseinself.annotated.values()ifgot_blocked_blocks:forgraphinself.blocked_graphs.values():self.blocked_graphs[graph]=Trueblocked_blocks=[blockforblock,doneinself.annotated.items()ifdoneisFalse]assertlen(blocked_blocks)==len(self.blocked_blocks)text=format_blocked_annotation_error(self,self.blocked_blocks)#raise SystemExit()raiseAnnotatorError(text)forgraphinnewgraphs:v=graph.getreturnvar()ifvnotinself.bindings:self.setbinding(v,annmodel.s_ImpossibleValue)# policy-dependent computationself.bookkeeper.compute_at_fixpoint()defbinding(self,arg,default=FAIL):"Gives the SomeValue corresponding to the given Variable or Constant."ifisinstance(arg,Variable):try:returnself.bindings[arg]exceptKeyError:ifdefaultisnotFAIL:returndefaultelse:raiseelifisinstance(arg,Constant):#if arg.value is undefined_value: # undefined local variables# return annmodel.s_ImpossibleValuereturnself.bookkeeper.immutableconstant(arg)else:raiseTypeError,'Variable or Constant expected, got %r'%(arg,)deftypeannotation(self,t):returnsignature.annotation(t,self.bookkeeper)defsetbinding(self,arg,s_value):ifarginself.bindings:asserts_value.contains(self.bindings[arg])self.bindings[arg]=s_valuedeftransfer_binding(self,v_target,v_source):assertv_sourceinself.bindingsself.bindings[v_target]=self.bindings[v_source]defwarning(self,msg,pos=None):ifposisNone:try:pos=self.bookkeeper.position_keyexceptAttributeError:pos='?'ifpos!='?':pos=self.whereami(pos)log.WARNING("%s/ %s"%(pos,msg))#___ interface for annotator.bookkeeper _______defrecursivecall(self,graph,whence,inputcells):ifisinstance(whence,tuple):parent_graph,parent_block,parent_index=whencetag=parent_block,parent_indexself.translator.update_call_graph(parent_graph,graph,tag)# self.notify[graph.returnblock] is a dictionary of call# points to this func which triggers a reflow whenever the# return block of this graph has been analysed.callpositions=self.notify.setdefault(graph.returnblock,{})ifwhenceisnotNone:ifcallable(whence):defcallback():whence(self,graph)else:callback=whencecallpositions[callback]=True# generalize the function's input argumentsself.addpendingblock(graph,graph.startblock,inputcells)# get the (current) return valuev=graph.getreturnvar()try:returnself.bindings[v]exceptKeyError:# the function didn't reach any return statement so far.# (some functions actually never do, they always raise exceptions)returnannmodel.s_ImpossibleValuedefreflowfromposition(self,position_key):graph,block,index=position_keyself.reflowpendingblock(graph,block)#___ simplification (should be moved elsewhere?) _______defsimplify(self,block_subset=None,extra_passes=None):# Generic simplificationstransform.transform_graph(self,block_subset=block_subset,extra_passes=extra_passes)ifblock_subsetisNone:graphs=self.translator.graphselse:graphs={}forblockinblock_subset:graph=self.annotated.get(block)ifgraph:graphs[graph]=Trueforgraphingraphs:simplify.eliminate_empty_blocks(graph)#___ flowing annotations in blocks _____________________defprocessblock(self,graph,block):# Important: this is not called recursively.# self.flowin() can only issue calls to self.addpendingblock().# The analysis of a block can be in three states:# * block not in self.annotated:# never seen the block.# * self.annotated[block] == False:# the input variables of the block are in self.bindings but we# still have to consider all the operations in the block.# * self.annotated[block] == graph-containing-block:# analysis done (at least until we find we must generalize the# input variables).#print '* processblock', block, cellsself.annotated[block]=graphifblockinself.blocked_blocks:delself.blocked_blocks[block]try:self.flowin(graph,block)exceptBlockedInference,e:self.annotated[block]=False# failed, hopefully temporarilyself.blocked_blocks[block]=(graph,e.opindex)exceptException,e:# hack for debug tools onlyifnothasattr(e,'__annotator_block'):setattr(e,'__annotator_block',block)raise# The dict 'added_blocks' is used by rpython.annlowlevel to# detect which are the new blocks that annotating an additional# small helper creates.ifself.added_blocksisnotNone:self.added_blocks[block]=Truedefreflowpendingblock(self,graph,block):assertnotself.frozenassertgraphnotinself.fixed_graphsself.pendingblocks[block]=graphassertblockinself.annotatedself.annotated[block]=False# must re-flowself.blocked_blocks[block]=(graph,None)defbindinputargs(self,graph,block,inputcells):# Create the initial bindings for the input args of a block.assertlen(block.inputargs)==len(inputcells)fora,cellinzip(block.inputargs,inputcells):self.setbinding(a,cell)self.annotated[block]=False# must flowin.self.blocked_blocks[block]=(graph,None)defmergeinputargs(self,graph,block,inputcells):# Merge the new 'cells' with each of the block's existing input# variables.oldcells=[self.binding(a)forainblock.inputargs]try:unions=[annmodel.unionof(c1,c2)forc1,c2inzip(oldcells,inputcells)]exceptannmodel.UnionError,e:e.args=e.args+(ErrorWrapper(gather_error(self,graph,block,None)),)raise# if the merged cells changed, we must redo the analysisifunions!=oldcells:self.bindinputargs(graph,block,unions)defwhereami(self,position_key):graph,block,i=position_keyblk=""ifblock:at=block.at()ifat:blk=" block"+atopid=""ifiisnotNone:opid=" op=%d"%ireturnrepr(graph)+blk+opiddefflowin(self,graph,block):#print 'Flowing', block, [self.binding(a) for a in block.inputargs]try:foriinrange(len(block.operations)):try:self.bookkeeper.enter((graph,block,i))self.consider_op(block,i)finally:self.bookkeeper.leave()exceptBlockedInference,e:if(e.opisblock.operations[-1]andblock.exitswitch==c_last_exception):# this is the case where the last operation of the block will# always raise an exception which is immediately caught by# an exception handler. We then only follow the exceptional# branches.exits=[linkforlinkinblock.exitsiflink.exitcaseisnotNone]elife.op.opnamein('simple_call','call_args','next'):# XXX warning, keep the name of the call operations in sync# with the flow object space. These are the operations for# which it is fine to always raise an exception. We then# swallow the BlockedInference and that's it.# About 'next': see test_annotate_iter_empty_container().returnelse:# other cases are problematic (but will hopefully be solved# later by reflowing). Throw the BlockedInference up to# processblock().raiseexceptannmodel.HarmlesslyBlocked:returnelse:# dead code removal: don't follow all exits if the exitswitch# is knownexits=block.exitsifisinstance(block.exitswitch,Variable):s_exitswitch=self.bindings[block.exitswitch]ifs_exitswitch.is_constant():exits=[linkforlinkinexitsiflink.exitcase==s_exitswitch.const]# mapping (exitcase, variable) -> s_annotation# that can be attached to booleans, exitswitchesknowntypedata=getattr(self.bindings.get(block.exitswitch),"knowntypedata",{})# filter out those exceptions which cannot# occour for this specific, typed operation.ifblock.exitswitch==c_last_exception:op=block.operations[-1]ifop.opnameinannmodel.BINARY_OPERATIONS:arg1=self.binding(op.args[0])arg2=self.binding(op.args[1])binop=getattr(pair(arg1,arg2),op.opname,None)can_only_throw=annmodel.read_can_only_throw(binop,arg1,arg2)elifop.opnameinannmodel.UNARY_OPERATIONS:arg1=self.binding(op.args[0])opname=op.opnameifopname=='contains':opname='op_contains'unop=getattr(arg1,opname,None)can_only_throw=annmodel.read_can_only_throw(unop,arg1)else:can_only_throw=Noneifcan_only_throwisnotNone:candidates=can_only_throwcandidate_exits=exitsexits=[]forlinkincandidate_exits:case=link.exitcaseifcaseisNone:exits.append(link)continuecovered=[cforcincandidatesifissubclass(c,case)]ifcovered:exits.append(link)candidates=[cforcincandidatesifcnotincovered]forlinkinexits:in_except_block=Falselast_exception_var=link.last_exception# may be None for non-exception linklast_exc_value_var=link.last_exc_value# may be None for non-exception linkifisinstance(link.exitcase,(types.ClassType,type)) \
andissubclass(link.exitcase,py.builtin.BaseException):assertlast_exception_varandlast_exc_value_varlast_exc_value_object=self.bookkeeper.valueoftype(link.exitcase)last_exception_object=annmodel.SomeType()ifisinstance(last_exception_var,Constant):last_exception_object.const=last_exception_var.valuelast_exception_object.is_type_of=[last_exc_value_var]ifisinstance(last_exception_var,Variable):self.setbinding(last_exception_var,last_exception_object)ifisinstance(last_exc_value_var,Variable):self.setbinding(last_exc_value_var,last_exc_value_object)last_exception_object=annmodel.SomeType()ifisinstance(last_exception_var,Constant):last_exception_object.const=last_exception_var.value#if link.exitcase is Exception:# last_exc_value_object = annmodel.SomeObject()#else:last_exc_value_vars=[]in_except_block=Trueignore_link=Falsecells=[]renaming={}fora,vinzip(link.args,link.target.inputargs):renaming.setdefault(a,[]).append(v)fora,vinzip(link.args,link.target.inputargs):ifa==last_exception_var:assertin_except_blockcells.append(last_exception_object)elifa==last_exc_value_var:assertin_except_blockcells.append(last_exc_value_object)last_exc_value_vars.append(v)else:cell=self.binding(a)if(link.exitcase,a)inknowntypedata:knownvarvalue=knowntypedata[(link.exitcase,a)]cell=pair(cell,knownvarvalue).improve()# ignore links that try to pass impossible valuesifcell==annmodel.s_ImpossibleValue:ignore_link=Trueifhasattr(cell,'is_type_of'):renamed_is_type_of=[]forvincell.is_type_of:new_vs=renaming.get(v,[])renamed_is_type_of+=new_vsassertcell.knowntypeistypenewcell=annmodel.SomeType()ifcell.is_constant():newcell.const=cell.constcell=newcellcell.is_type_of=renamed_is_type_ofifhasattr(cell,'knowntypedata'):renamed_knowntypedata={}for(value,v),sincell.knowntypedata.items():new_vs=renaming.get(v,[])fornew_vinnew_vs:renamed_knowntypedata[value,new_v]=sassertisinstance(cell,annmodel.SomeBool)newcell=annmodel.SomeBool()ifcell.is_constant():newcell.const=cell.constcell=newcellcell.set_knowntypedata(renamed_knowntypedata)cells.append(cell)ifignore_link:continueifin_except_block:last_exception_object.is_type_of=last_exc_value_varsself.links_followed[link]=Trueself.addpendingblock(graph,link.target,cells)ifblockinself.notify:# reflow from certain positions when this block is doneforcallbackinself.notify[block]:ifisinstance(callback,tuple):self.reflowfromposition(callback)# callback is a positionelse:callback()#___ creating the annotations based on operations ______defconsider_op(self,block,opindex):op=block.operations[opindex]argcells=[self.binding(a)forainop.args]consider_meth=getattr(self,'consider_op_'+op.opname,None)ifnotconsider_meth:raiseException,"unknown op: %r"%op# let's be careful about avoiding propagated SomeImpossibleValues# to enter an op; the latter can result in violations of the# more general results invariant: e.g. if SomeImpossibleValue enters is_# is_(SomeImpossibleValue, None) -> SomeBool# is_(SomeInstance(not None), None) -> SomeBool(const=False) ...# boom -- in the assert of setbinding()forarginargcells:ifisinstance(arg,annmodel.SomeImpossibleValue):raiseBlockedInference(self,op,opindex)try:resultcell=consider_meth(*argcells)exceptException,e:graph=self.bookkeeper.position_key[0]e.args=e.args+(ErrorWrapper(gather_error(self,graph,block,opindex)),)raiseifresultcellisNone:resultcell=self.noreturnvalue(op)elifresultcell==annmodel.s_ImpossibleValue:raiseBlockedInference(self,op,opindex)# the operation cannot succeedassertisinstance(resultcell,annmodel.SomeObject)assertisinstance(op.result,Variable)self.setbinding(op.result,resultcell)# bind resultcell to op.resultdefnoreturnvalue(self,op):returnannmodel.s_ImpossibleValue# no return value (hook method)# XXX "contains" clash with SomeObject methoddefconsider_op_contains(self,seq,elem):self.bookkeeper.count("contains",seq)returnseq.op_contains(elem)defconsider_op_newtuple(self,*args):returnannmodel.SomeTuple(items=args)defconsider_op_newlist(self,*args):returnself.bookkeeper.newlist(*args)defconsider_op_newdict(self):returnself.bookkeeper.newdict()def_registeroperations(cls,model):# All unary operationsd={}foropnameinmodel.UNARY_OPERATIONS:fnname='consider_op_'+opnameexecpy.code.Source("""def consider_op_%s(self, arg, *args): return arg.%s(*args)"""%(opname,opname)).compile()inglobals(),dsetattr(cls,fnname,d[fnname])# All binary operationsforopnameinmodel.BINARY_OPERATIONS:fnname='consider_op_'+opnameexecpy.code.Source("""def consider_op_%s(self, arg1, arg2, *args): return pair(arg1,arg2).%s(*args)"""%(opname,opname)).compile()inglobals(),dsetattr(cls,fnname,d[fnname])_registeroperations=classmethod(_registeroperations)# register simple operations handlingRPythonAnnotator._registeroperations(annmodel)classBlockedInference(Exception):"""This exception signals the type inference engine that the situation is currently blocked, and that it should try to progress elsewhere."""def__init__(self,annotator,op,opindex):self.annotator=annotatortry:self.break_at=annotator.bookkeeper.position_keyexceptAttributeError:self.break_at=Noneself.op=opself.opindex=opindexdef__repr__(self):ifnotself.break_at:break_at="?"else:break_at=self.annotator.whereami(self.break_at)return"<BlockedInference break_at %s [%s]>"%(break_at,self.op)__str__=__repr__