o jZh3(@sdZddlZddlZddlmZddlmZddlmZddlm Z ddl m Z dd l m Z dd l m Z dd lmZeZGd d d ZGdddZddZeZejZejZdS)zk Managing Gateway Groups and interactions with multiple channels. (c) 2008-2014, Holger Krekel and others Npartial)Lock)gateway_bootstrap) gateway_io) get_execmodel)trace) WorkerPool)XSpecc@seZdZdZdZd&ddZeddZed d Zd'd d Z ddZ ddZ ddZ ddZ ddZd'ddZddZddZddZd d!Zd'd"d#Zd$d%Zd S)(GroupzGateway Groups.popenthreadcCsHg|_d|_t|_g|_|||D]}||qt|j dS)zjinitialize group and make gateways as specified. execmodel can be 'thread' or 'eventlet'. rN) _gateways_autoidcounterr _autoidlock_gateways_to_join set_execmodel makegatewayatexitregister_cleanup_atexit)selfZxspecs execmodelxspecrrD/var/www/html/lang_env/lib/python3.10/site-packages/execnet/multi.py__init__s  zGroup.__init__cC|jSN) _execmodelrrrrr,zGroup.execmodelcCrr)_remote_execmodelr!rrrremote_execmodel0r"zGroup.remote_execmodelNcCs2|jrtd|dur |}t||_t||_dS)asSet the execution model for local and remote site. execmodel can be one of "thread" or "eventlet" (XXX gevent). 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 ValueErrorrr r#)rrr$rrrr4s  zGroup.set_execmodelcCsdd|D}d|S)NcSsg|]}|jqSr)id.0gwrrr Isz"Group.__repr__..z r)rZ idgatewaysrrr__repr__HszGroup.__repr__cCsBt|tr |j|S|jD]}||ks|j|kr|Sq t|r) isinstanceintrr&KeyError)rkeyr)rrr __getitem__Ls   zGroup.__getitem__cCs$z||WdStyYdSw)NTF)r.rr/rrr __contains__Ts  zGroup.__contains__cC t|jSr)lenrr!rrr__len__[ z Group.__len__cCstt|jSr)iterlistrr!rrr__iter__^szGroup.__iter__c CsZ|s|j}t|tst|}|||jdur|jj|_|jrD|jr%J||j}| t }| t |t ||j}t||}n9|jsM|jsM|jr\t j||jd}t||}n!|jruddlm}|j|||jd}t||}ntd|j||_|||js|js|jr| d}|jrt|jpd} | |j| |jf||S)apcreate 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', 'eventlet', 'gevent' model for execution 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_idrr$backendviasocket remote_execrsendvarsZProxyIOrZ bootstrapr sshZ vagrant_sshZ create_ior:r%_specspec _registerchdirniceenvr- waitclose) rrFZmasterZ proxy_channelZproxy_io_masterr)ior:channelrIrrrras@         zGroup.makegatewaycCsr|jdur7|j%dt|j}|jd7_||vr"td|||_WddS1s0wYdSdS)z4(re-entrant) allocate id for the given xspec object.Nr)rzalready have gateway with id )r&rstrrr%)rrFr&rrrr<s "zGroup.allocate_idcCs<t|drJ|js J|j|vsJ|j|||_dS)N_group)hasattrr&rappendrOrgatewayrrrrGs    zGroup._registercCs|j||j|dSr)rremoverrQrRrrr _unregisters zGroup._unregistercCs td|d|jdddS)Nz=== atexit cleanup z ===g?timeout)r terminater!rrrrszGroup._cleanup_atexitcs|rDi}|D] }|jjrd||jj<q|D] }|j|vr |qddddt|j|fdd|jDg|jdd<|sdSdS) a:trigger 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. TcSs||jdSr)join_iowaitr)rrr join_waitsz"Group.terminate..join_waitcSstd||jdS)Nz,Gateways did not come down after timeout: %r)r rZkillr\rrrr^s zGroup.terminate..killcs g|] }t|t|fqSrrr'r]r^rrr*sz#Group.terminate..N)rFr>r&exitsafe_terminaterr)rrWZviasr)rr_rrXs*    zGroup.terminatecKs.g}|D]}||j|fi|qt|S)zsremote_exec source on all member gateways and return MultiChannel connecting to all sub processes. )rQr@ MultiChannel)rsourcekwargschannelsr)rrrr@szGroup.remote_exec)rrr)__name__ __module__ __qualname____doc__r;rpropertyrr$rr+r0r2r5r9rr<rGrUrrXr@rrrrr s*     C   #r c@sZeZdZddZddZddZddZd d Zd d ZdddZ e fddZ ddZ dS)rbcCs ||_dSr _channels)rrerrrrr6zMultiChannel.__init__cCr3r)r4rlr!rrrr5r6zMultiChannel.__len__cCr3r)r7rlr!rrrr9r6zMultiChannel.__iter__cCs |j|Srrkr1rrrr0r6zMultiChannel.__getitem__cCs ||jvSrrk)rZchanrrrr2r6zMultiChannel.__contains__cCs|jD]}||qdSr)rlrA)ritemchrrr send_eachs  zMultiChannel.send_eachFcCsHt|drJg}|jD]}|}|r|||fq ||q |S)N_queue)rPrlZreceiverQ)rZ withchannellrnobjrrr receive_eachs  zMultiChannel.receive_eachcszjWSty?d_jD](}jdur |jjj_|ffdd }|tur2||q|j||dqjYSw)Ncsj||fdSr)rpput)rrrMr!rr putreceivedsz4MultiChannel.make_receive_queue..putreceived) endmarker) rpAttributeErrorrlrSrqueueQueueNO_ENDMARKER_WANTEDZ setcallback)rrvrnrurr!rmake_receive_queue s     zMultiChannel.make_receive_queuec CsZd}|jD]}z|Wq|jy|durt}Yqw|r+|d|ddS)Nr)rlrKZ RemoteErrorsysexc_infowith_traceback)rfirstrnrrrrKs  zMultiChannel.waitcloseN)F) rfrgrhrr5r9r0r2rorsrzr{rKrrrrrbs  rbcsbt|fdd}g}|D]\}}|||}||q|D]}|q"jddS)Ncs8|}z |jdWdSty|YdSw)NrV)spawngetOSError)termfunckillfuncZ termreplyrWZ workerpoolrrtermkill*s   z safe_terminate..termkillrV)r rrQrZwaitall)rrWZlist_of_paired_functionsrZ replylistrrZreplyrrrra's   ra)rirr} functoolsr threadingrrDrrZ gateway_baserr r rr objectrzr rbraZ default_grouprrrrrrs&        V=