a &j-@sTdZddlmZddlZddlZddlmZddlmZddl m Z ddl m Z ddl m Z dd l m Z dd l mZdd l mZdd l mZdd l mZddlmZddlmZddlmZddlmZddlmZddlmZddlmZddlmZe r ddlmZeZ GdddZ!GdddZ"ddddd d!Z#e!Z$e$j%Z%e$j&Z&dS)"zk Managing Gateway Groups and interactions with multiple channels. (c) 2008-2014, Holger Krekel and others ) annotationsNpartial)Lock) TYPE_CHECKING)Any)Callable)Iterable)Iterator)Literal)Sequence)overload)gateway_bootstrap) gateway_io)Channel) ExecModel) WorkerPool) get_execmodel)trace)XSpec)Gatewayc@seZdZdZdZd=ddddd d Zed d d dZed d ddZd>ddddddZ dd ddZ dddddZ dddddZ d d d!d"Z d#d d$d%Zd?d&dd'd(d)Zd*dd'd+d,Zddd-d.d/Zddd-d0d1Zdd d2d3Zd@d4dd5d6d7Zd8d9d:d;d<ZdS)AGroupzGateway Group.popenthreadzIterable[XSpec | str | None]strNone)xspecs execmodelreturncCsHg|_d|_t|_g|_|||D]}||q(t|j dS)z|Initialize a group and make gateways as specified. execmodel can be one of the supported execution models. rN) _gateways_autoidcounterr _autoidlock_gateways_to_join set_execmodel makegatewayatexitregister_cleanup_atexit)selfrrxspecrr3E:\soccer\ml_env\Lib\site-packages\execnet/multi.py__init__+s  zGroup.__init__rr cCs|jSN) _execmodelr*rrr,r@szGroup.execmodelcCs|jSr/)_remote_execmodelr1rrr,remote_execmodelDszGroup.remote_execmodelNz str | None)rr3r cCs2|jrtd|dur|}t||_t||_dS)aeSet the execution model for local and remote site. execmodel can be one of the supported execution models. It determines the execution model for any newly created gateway. If remote_execmodel is not specified it takes on the value of execmodel. NOTE: Execution models can only be set before any gateway is created. zBcan not set execution models if gateways have been created alreadyN)r! ValueErrorrr0r2)r*rr3rrr,r%Hs  zGroup.set_execmodelcCsdd|D}d|S)NcSsg|] }|jqSr)id.0gwrrr, ]z"Group.__repr__..z r)r*Z idgatewaysrrr,__repr__\szGroup.__repr__zint | str | Gatewayrkeyr cCsFt|tr|j|S|jD]}||ks0|j|kr|Sqt|dSr/) isinstanceintr!r5KeyError)r*r=r8rrr, __getitem__`s     zGroup.__getitem__boolcCs(z||WdSty"YdS0dS)NTF)r@r*r=rrr, __contains__hs  zGroup.__contains__r?cCs t|jSr/)lenr!r1rrr,__len__osz Group.__len__zIterator[Gateway]cCstt|jSr/)iterlistr!r1rrr,__iter__rszGroup.__iter__zXSpec | str | None)specr c Csd|s |j}t|tst|}|||jdur:|jj|_|jr|jrJJ||j}| t }| t |t ||j}t||}nr|js|js|jrt j||jd}t||}nB|jrddlm}|j|||jd}t||}ntd|j||_|||js"|js"|jr`| d} |jr@t|jpBd} | |j| |jf| |S)aCreate and configure a gateway to a Python interpreter. The ``spec`` string encodes the target gateway type and configuration information. The general format is:: key1=value1//key2=value2//... If you leave out the ``=value`` part a True value is assumed. Valid types: ``popen``, ``ssh=hostname``, ``socket=host:port``. Valid configuration:: id= specifies the gateway id python= specifies which python interpreter to execute execmodel=model 'thread', 'main_thread_only', 'eventlet', 'gevent' execution model chdir= specifies to which directory to change nice= specifies process priority of new process env:NAME=value specifies a remote environment variable setting. If no spec is given, self.defaultspec is used. N)rr)gateway_socketzno gateway type found for a import os path, nice, env = channel.receive() if path: if not os.path.exists(path): os.mkdir(path) os.chdir(path) if nice and hasattr(os, 'nice'): os.nice(nice) if env: for name, value in env.items(): os.environ[name] = value r) defaultspecr>r allocate_idrr3backendviasocket remote_execrsendvarsZProxyIOrZ bootstraprsshZ vagrant_sshZ create_iorKr4_specrJ _registerchdirniceenvr? waitclose) r*rJZmasterZ proxy_channelZproxy_io_masterr8iorKsiochannelrYrrr,r&us@         zGroup.makegatewayrcCsl|jdurh|jHdt|j}|jd7_||vrDtd|||_Wdn1s^0YdS)z4(re-entrant) allocate id for the given xspec object.Nr8rzalready have gateway with id )r5r#rr"r4)r*rJr5rrr,rMs zGroup.allocate_id)gatewayr cCs<t|drJ|jsJ|j|vs&J|j|||_dS)N_group)hasattrr5r!appendr`r*r_rrr,rWs   zGroup._registercCs|j||j|dSr/)r!remover$rbrcrrr, _unregisters zGroup._unregistercCs td|d|jdddS)Nz=== atexit cleanup z ===g?timeout)r terminater1rrr,r)szGroup._cleanup_atexit float | None)rgr cs|rt}|D]}|jjr||jjq|D]}|j|vr.|q.ddddddddddt|j|fdd |jDg|jd d <qd S) aBTrigger exit of member gateways and wait for termination of member gateways and associated subprocesses. After waiting timeout seconds try to to kill local sub processes of popen- and ssh-gateways. Timeout defaults to None meaning open-ended waiting and no kill attempts. rr)r8r cSs||jdSr/)join_iowaitr8rrr, join_waitsz"Group.terminate..join_waitcSstd||jdS)Nz,Gateways did not come down after timeout: %r)rrkkillrmrrr,ros zGroup.terminate..killcs g|]}t|t|fqSrrr6rnrorr,r9sz#Group.terminate..N) setrJrOaddr5exitsafe_terminaterr$)r*rgZviasr8rrpr,rhs$    zGroup.terminatezCstr | types.FunctionType | Callable[..., object] | types.ModuleType MultiChannel)sourcer cKs.g}|D]}||j|fi|qt|S)zlremote_exec source on all member gateways and return a MultiChannel connecting to all sub processes.)rbrQru)r*rvkwargschannelsr8rrr,rQszGroup.remote_exec)rr)N)N)N)__name__ __module__ __qualname____doc__rLr-propertyrr3r%r;rArDrFrIr&rMrWrer)rhrQrrrr,r&s,D %rc@seZdZdddddZdddd Zd dd d Zdd dddZd ddddZdddddZe d+dddddZ e dd dd!dZ d,dd#dd$dZ e fdd%d&d'Z ddd(d)Z d*S)-ruzSequence[Channel]r)rxr cCs ||_dSr/ _channels)r*rxrrr,r-szMultiChannel.__init__r?r.cCs t|jSr/)rErr1rrr,rFszMultiChannel.__len__zIterator[Channel]cCs t|jSr/)rGrr1rrr,rI szMultiChannel.__iter__rr<cCs |j|Sr/r~rCrrr,rAszMultiChannel.__getitem__rB)chanr cCs ||jvSr/r~)r*rrrr,rDszMultiChannel.__contains__object)itemr cCs|jD]}||qdSr/)rrR)r*rchrrr, send_eachs zMultiChannel.send_each.zLiteral[False]z list[Any]) withchannelr cCsdSr/rr*rrrr, receive_eachszMultiChannel.receive_eachz Literal[True]zlist[tuple[Channel, Any]]cCsdSr/rrrrr,rsFz%list[tuple[Channel, Any]] | list[Any]cCsHt|drJg}|jD]*}|}|r8|||fq||q|S)N_queue)rarZreceiverb)r*rlrobjrrr,r s   endmarkerc szjWStyd_jD]X}jdur@|jjj_|fdddfdd }|turl||q"|j||dq"jYS0dS)Nrr)r^r csj||fdSr/)rput)rr^r1rr, putreceived6sz4MultiChannel.make_receive_queue..putreceivedr) rAttributeErrorrr_rqueueQueueNO_ENDMARKER_WANTEDZ setcallback)r*rrrrr1r,make_receive_queue-s    zMultiChannel.make_receive_queuec Cs\d}|jD]D}z |Wq |jyL}z|dur8|}WYd}~q d}~00q |rX|dSr/)rr[Z RemoteError)r*firstrexcrrr,r[?s  zMultiChannel.waitcloseN).)F)ryrzr{r-rFrIrArDrr rrrr[rrrr,rus rurrir)rrgr csht|ddfdd }g}|D] \}}|||}||q$|D] }|qJjddS)Nrr.cs8|}z|jdWnty2|Yn0dS)Nrf)spawngetOSError)termfunckillfuncZ termreplyrgZ workerpoolrr,termkillPs   z safe_terminate..termkillrf)rrrbrZwaitall)rrgZlist_of_paired_functionsrZ replylistrrZreplyrrr,rtKs   rt)'r| __future__rr'types functoolsr threadingrtypingrrrr r r r r rUrrZ gateway_baserrrrrr+rr_rrrrrurtZ default_groupr&r%rrrr,s>                    _G