o "j@sbddlZddlmZddlmZmZGdddejjjj Z GdddZ dd d Z dd d Z dS)N)Variable) LayerHelpercorecs,eZdZdZfddZfddZZS)BlockGuardServzl BlockGuardServ class. BlockGuardServ class is used to create an op with a block in a program. cs,t|ts tdt|jj||_dS)Nz$BlockGuardServ takes a ListenAndServ) isinstance ListenAndServ TypeErrorsuper__init__helper main_programserver)selfr __class__\/var/www/html/Deteccion_Ine/venv/lib/python3.10/site-packages/paddle/incubate/nn/layer/io.pyr s  zBlockGuardServ.__init__cs&|durdS|jt|||S)NF)r complete_opr __exit__)rexc_typeexc_valexc_tbrrrr!s zBlockGuardServ.__exit__)__name__ __module__ __qualname____doc__r r __classcell__rrrrrs rc@s:eZdZdZdddZddZdd Zd d Zd d ZdS)ra **ListenAndServ Layer** ListenAndServ is used to create a rpc server bind and listen on specific TCP port, this server will run the sub-block when received variables from clients. Args: endpoint(string): IP:port string which the server will listen on. inputs(list): a list of variables that the server will get from clients. fan_in(int): how many client are expected to report to this server, default: 1. optimizer_mode(bool): whether to run the server as a parameter server, default: True. Examples: .. code-block:: python >>> # doctest: +REQUIRES(env:DISTRIBUTED) >>> from paddle.incubate.nn.layer.io import ListenAndServ >>> import paddle >>> paddle.enable_static() >>> place = paddle.CPUPlace() >>> main = paddle.static.Program() >>> with paddle.static.program_guard(main): ... serv = ListenAndServ( ... "127.0.0.1:6170", ["X"], optimizer_mode=False) ... with serv.do(): ... x = paddle.static.data( ... shape=[32, 32], ... dtype='float32', ... name="X") ... paddle.nn.initializer.Constant(value=1.0)(x, main.global_block()) ... paddle.scale(x=x, scale=10.0) >>> exe = paddle.static.Executor(place) >>> exe.run(main) TcCs,td|_||_g|_||_||_||_dS)Nlisten_and_serv)rr inputsoutputsendpointfan_inoptimizer_mode)rr"r r#r$rrrr Os  zListenAndServ.__init__cCst|S)N)r)rrrrdoYszListenAndServ.doc Cs|jj}|}|}g}g}|jD]@}|jr5d|jvr4d|jvr4||jdj||jdjq|j D]}| |D]}|| ||| |q?q8q||fS)NZGradParam) r r current_block parent_blockopsr$r appendnameZ input_namesinputvar) rr r'r(paramsZgradsopZinameZ in_var_namerrrget_params_and_grads\s$  z"ListenAndServ.get_params_and_gradscCs,|jj}|j}|dksJ||}|S)Nr)r r r' parent_idxblock)rprogr1r(rrrr(rs    zListenAndServ.parent_blockc CsVddlm}|jj}|}|}|jdd|jii|j|j |g|j dgdddS)Nr)DistributedModerX)r"ZFaninZoptimize_blocksZdistributed_modeZgrad_to_block_idtyper r!attrs) Z7paddle.incubate.distributed.fleet.parameter_server.moder4r r r'r( append_opr r"r#ZSYNC)rr4r r'r(rrrrys  zListenAndServ.complete_opN)rT) rrrrr r%r0r(rrrrrr)s %  rTc Cst|tksJ|durg}nt|tr|g}t|tksJ|d}tt|}td it}tj }|j dd|id|id|d||tj j j id |r`|j d d|idgid|id dSdS) ak Send variables to the server side, and get vars from server side when server have finished running server side program. Args: endpoints (str): comma separated IP:PORT pairs in the order of send_vars to send send_vars (list): variables to send to server sync (bool): whether to wait the request finish N,Sendsendr5Out endpointsepmapr7Z send_barrier)r<)r8listrrsplitsetrlocalsrZop_proto_and_checker_makerZkOpRoleAttrNamer:ZOpRoleZRPC)r?Z send_varsZ dummy_outputsyncr@r Zrpc_op_role_namerrrr<s6        r<cCst|tksJ|durg}nt|tr|g}t|tksJ|d}tt|}td it}|jdd|id|i||dd|rP|jd d|id |id |S) aY Receive variables from server side Args: endpoints (str): comma separated IP:PORT pairs in the order of send_vars to send get_vars (list): vars to get from server after send completes. sync (bool): whether to wait the request finish Returns: list: list of received variables Nr;Recvrecvr5r>)r?r@r7Z fetch_barrierr?)r8r!r9)rF) r8rArrrBrCrrDr:)r?Zget_varsZ dummy_inputrEr@r rrrrFs,    rF)NT)ZpaddleZbase.frameworkrZ frameworkrrZstaticnnZ control_flowZ BlockGuardrrr<rFrrrrs  i.