File manager - Edit - /opt/saltstack/salt/lib/python3.10/site-packages/salt/transport/__pycache__/ipc.cpython-310.pyc
Back
o ;jWw � @ s( d Z ddlZddlZddlZddlZddlZddlZddlZddl Zddl Zddl Zddl Zddl ZddlZddl mZ ddl mZ ddlmZmZ ddlmZ e�e�Zdd� ZG d d � d ejj j j �Z G d d � d �Z G d d� d�Z!G dd� de!�Z"G dd� de �Z#G dd� d�Z$G dd� de!�Z%dS )z IPC transport classes � N)�IOLoop)� TimeoutError)�IOStream�StreamClosedError)�Lockc C s | j d ur | j �| � d S d S �N)�_future_with_timeout�_done_callback)�future� r �F/opt/saltstack/salt/lib/python3.10/site-packages/salt/transport/ipc.py� future_with_timeout_callback s �r c s, e Zd Z� fdd�Zdd� Zdd� Z� ZS )�FutureWithTimeoutc s� t � �� || _|| _|d ur$|dk rd}| j�| j�� | | j�| _nd | _t| jd�r?| | j_ | j� � r=t | j� d S d S | | j_ | j� t � d S )Ng�������?r ) �super�__init__�io_loop�_futureZ add_timeout�time�_timeout_callback�_timeout_handle�hasattrr �doner �add_done_callback)�selfr r �timeout�� __class__r r r % s" � �zFutureWithTimeout.__init__c C s d | _ d | j_| �t� � d S r )r r r � set_exception�TornadoTimeoutError�r r r r r = s z#FutureWithTimeout._timeout_callbackc C sd z| j d ur| j�| j � d | _ | �|�� � W d S ty1 } z | �|� W Y d }~d S d }~ww r )r r Zremove_timeout� set_result�result� Exceptionr )r r �excr r r r F s ��z FutureWithTimeout._done_callback)�__name__� __module__� __qualname__r r r � __classcell__r r r r r $ s r c @ sn e Zd ZdZdgZdgZddd�Zdd� Zej j j j e fd d ��Zd d � Zd d� Zdd� Zdd� Zdd� ZdS )� IPCServerz� A Tornado IPC server very similar to Tornado's TCPServer class but using either UNIX domain sockets or TCP sockets � handle_stream�closeNc C s8 || _ d| _|| _d| _|ptjjjj� � | _ d| _ dS )a Create a new Tornado IPC server :param str/int socket_path: Path on the filesystem for the socket to bind to. This socket does not need to exist prior to calling this method, but parent directories should. It may also be of type 'int', in which case it is used as the port for a tcp localhost connection. :param IOLoop io_loop: A Tornado ioloop to handle scheduling :param func payload_handler: A function to customize handling of incoming data. FN) � socket_path�_started�payload_handler�sock�salt�ext�tornado�ioloopr �currentr �_closing)r r+ r r- r r r r ^ s zIPCServer.__init__c C �� t �d| j� t| jt�r6t�tjtj�| _| j� tj tj d� | j� d� | j� d| jf� | j�d� n tjjj�| j�| _tjj�| j�� tjjj�| j| j� W d � n1 s^w Y d| _dS )�y Perform the work necessary to start up a Tornado IPC server Blocks until socket is established z IPCServer: binding to socket: %s� r � 127.0.0.1� NT��log�tracer+ � isinstance�int�socket�AF_INET� SOCK_STREAMr. � setsockopt� SOL_SOCKET� SO_REUSEADDR� setblocking�bind�listenr/ r0 r1 ZnetutilZbind_unix_socket�utils� asynchronous�current_ioloopr Zadd_accept_handler�handle_connectionr, r r r r �startw � � � zIPCServer.startc # sL � t jjjjdd� �� � fdd�}| js�|�� s�z3|�d�V }t� d|�d }|�|�V }t j j j |dd �}|d }| j �| j||||�d i ��� W nK |y\ t�d | j� Y dS ty } z| jdkrot�d | � nt�d| � W Y d} ~ n d} ~ w ty� } z t�d| � W Y d} ~ nd} ~ ww | js�|�� rdS dS dS dS )z� Override this to handle the streams as they arrive :param IOStream stream: An IOStream for processing See https://tornado.readthedocs.io/en/latest/iostream.html#tornado.iostream.IOStream for additional details. c S s t jjj�d ��r )r/ r0 r1 �gen�Return)�msgr r r �_null� s z&IPCServer.handle_stream.<locals>._nullc s, � � d�rtjjjj� �fdd��}|S �S )N�midc 3 s. � t jjj| d� d idd�}��|�V d S )NrR T)�header�raw_body)r/ � transport�frame� frame_msg_ipc�write)rP �pack�rS �streamr r �return_message� s � �zGIPCServer.handle_stream.<locals>.write_callback.<locals>.return_message)�getr/ r0 r1 rN � coroutine)r[ rS r\ �rQ rZ r �write_callback� s z/IPCServer.handle_stream.<locals>.write_callback� �>Ir F��raw�body�head� Client disconnected from IPC %sz>Exception occurred with error number 0, spurious exception: %s�,Exception occurred while handling stream: %sN)r/ r0 r1 rN r^ r4 �closed� read_bytes�struct�unpackrH �msgpack�unpackbr �spawn_callbackr- r] r; r< r+ �OSError�errno�errorr" ) r r[ Z_StreamClosedErrorr` � length_bytes�length�payload� framed_msgre r# r r_ r r) � s@ � � � ��� �zIPCServer.handle_streamc C s� t �d|r|n|� z'tjj�| j�� t|�}W d � n1 s"w Y | j�| j |� W d S t yJ } z t � d|� W Y d }~d S d }~ww )N�-IPCServer: Handling connection to address: %s�IPC streaming error: %s) r; r< r/ rH rI rJ r r ro r) r" rr )r � connection�addressr[ r# r r r rK � s �� ���zIPCServer.handle_connectionc C s. | j rdS d| _ t| jd�r| j�� dS dS ��� Routines to handle any cleanup before the instance shuts down. Sockets and filehandles should be closed explicitly, to prevent leaks. NTr* )r4 r r. r* r r r r r* � s �zIPCServer.closec C �$ z| � � W d S ty Y d S w r �r* � TypeErrorr r r r �__del__� � �zIPCServer.__del__c C � | S r r r r r r � __enter__� � zIPCServer.__enter__c G � | � � d S r �r* �r �argsr r r �__exit__� � zIPCServer.__exit__�NN)r$ r% r& �__doc__� async_methods� close_methodsr rL r/ r0 r1 rN r^ r r) rK r* r� r� r� r r r r r( Q s �� ; r( c @ sb e Zd ZdZddd�Zdd� Zddd�Zejj j j dd d ��Z d d � Z d d� Zdd� Zdd� ZdS )� IPCClienta� A Tornado IPC client very similar to Tornado's TCPClient class but using either UNIX domain sockets or TCP sockets This was written because Tornado does not have its own IPC server/client implementation. :param IOLoop io_loop: A Tornado ioloop to handle scheduling :param str/int socket_path: A path on the filesystem where a socket belonging to a running IPCServer can be found. It may also be of type 'int', in which case it is used as the port for a tcp localhost connection. Nc C s2 |p t jjjj�� | _|| _d| _d| _ d| _ dS )z� Create a new IPC client IPC clients cannot bind to ports, but must connect to existing IPC servers. Clients can then send messages to the server. FN) r/ r0 r1 r2 r r3 r r+ r4 r[ �_connecting_future�r r+ r r r r r s zIPCClient.__init__c C s | j d uo | j �� S r )r[ ri r r r r � connected s zIPCClient.connectedc sr �j dur�j �� s�j }n�j dur�j �� tjjj�� }|�_ ��|� � dur7� �fdd�}|� |� |S )z+ Connect to the IPC socket Nc s | � � }�j�� |� d S r )r! r Z add_callback)r �response��callbackr r r � handle_future+ s z(IPCClient.connect.<locals>.handle_future) r� r � exceptionr/ r0 r1 � concurrent�Future�_connectr )r r� r r r� r r� r �connect s zIPCClient.connectc c st � t | jt�rtj}d| jf}ntj}| j}d| _|dur#t�� | } | jr)dS | jdu rPt j j � | j �� tt�|tj��| _W d � n1 sKw Y z t�d| j� | j�|�V | jdurj| j�d� W dS ty� } z@| j�� r|d| _|du s�t�� |kr�| jdur�| j�� d| _| jdur�| j�|� W Y d}~dS t jjj� d�V W Y d}~nd}~ww q$)z0 Connect to a running IPCServer r8 NTz#IPCClient: Connecting to socket: %sr7 ) r= r+ r> r? r@ �AF_UNIXr[ r r4 r/ rH rI rJ r r rA r; r< r� r� r r" ri r* r r0 r1 rN �sleep)r r Z sock_typeZ sock_addrZ timeout_at�er r r r� 3 sH � � ���zIPCClient._connectc C s� | j rdS d| _ d| _t�d| jj� | jdur@| j�� sBz| j�� W dS t y? } z|j t j kr4� W Y d}~dS d}~ww dS dS )r| NTzClosing %s instance) r4 r� r; �debugr r$ r[ ri r* rp rq �EBADF�r r# r r r r* ^ s ����zIPCClient.closec C r} r r~ r r r r r� u r� zIPCClient.__del__c C r� r r r r r r r� r� zIPCClient.__enter__c G r� r r� r� r r r r� � r� zIPCClient.__exit__r r� )r$ r% r& r� r r� r� r/ r0 r1 rN r^ r� r* r� r� r� r r r r r� � s * r� c @ s4 e Zd ZdZg d�ZdgZejjj j ddd��Z dS )�IPCMessageClienta� Salt IPC message client Create an IPC client to send messages to an IPC server An example of a very simple IPCMessageClient connecting to an IPCServer. This example assumes an already running IPCMessage server. IMPORTANT: The below example also assumes a running IOLoop process. # Import Tornado libs import salt.ext.tornado.ioloop # Import Salt libs import salt.config import salt.transport.ipc io_loop = salt.ext.tornado.ioloop.IOLoop.current() ipc_server_socket_path = '/var/run/ipc_server.ipc' ipc_client = salt.transport.ipc.IPCMessageClient(ipc_server_socket_path, io_loop=io_loop) # Connect to the server ipc_client.connect() # Send some data ipc_client.send('Hello world') )�sendr� r� r* Nc c s8 � | � � s | �� V tjjj|dd�}| j�|�V dS )a Send a message to an IPC socket If the socket is not currently connected, a connection will be established. :param dict msg: The message to be sent :param int timeout: Timeout when sending message (Currently unimplemented) T�rT N)r� r� r/ rU rV rW r[ rX )r rP r ZtriesrY r r r r� � s � zIPCMessageClient.sendr� ) r$ r% r& r� r� r� r/ r0 r1 rN r^ r� r r r r r� � s � r� c @ s e Zd ZdZdS )�IPCMessageServera Salt IPC message server Creates a message server which can create and bind to a socket on a given path and then respond to messages asynchronously. An example of a very simple IPCServer which prints received messages to a console: # Import Tornado libs import salt.ext.tornado.ioloop # Import Salt libs import salt.transport.ipc io_loop = salt.ext.tornado.ioloop.IOLoop.current() ipc_server_socket_path = '/var/run/ipc_server.ipc' ipc_server = salt.transport.ipc.IPCMessageServer(ipc_server_socket_path, io_loop=io_loop, payload_handler=print_to_console) # Bind to the socket and prepare to run ipc_server.start() # Start the server io_loop.start() # This callback is run whenever a message is received def print_to_console(payload): print(payload) See IPCMessageClient() for an example of sending messages to an IPCMessageServer instance N)r$ r% r& r� r r r r r� � s r� c @ sR e Zd ZdZddd�Zdd� Zdd� Zd d � Zd d � Zd d� Z dd� Z dd� Z dS )�IPCMessagePublisherz~ A Tornado IPC Publisher similar to Tornado's TCPServer class but using either UNIX domain sockets or TCP sockets Nc C s8 || _ || _d| _d| _|pt�� | _d| _t� | _ dS )a� Create a new Tornado IPC server :param dict opts: Salt options :param str/int socket_path: Path on the filesystem for the socket to bind to. This socket does not need to exist prior to calling this method, but parent directories should. It may also be of type 'int', in which case it is used as the port for a tcp localhost connection. :param IOLoop io_loop: A Tornado ioloop to handle scheduling FN) �optsr+ r, r. r r3 r r4 �set�streams)r r� r+ r r r r r � s z IPCMessagePublisher.__init__c C r5 )r6 z*IPCMessagePublisher: binding to socket: %sr7 r r8 r9 NTr: r r r r rL rM zIPCMessagePublisher.startc sZ |f� fdd� }z|� |�}W n ty � j�|� Y dS w |dur+|�|� dS dS )aQ Queue a write to ``stream`` and attach a completion callback to handle exceptions. Note: this is intentionally NOT a Tornado @gen.coroutine. When it was a coroutine, every published message produced a long-lived gen.Runner per subscriber stream that waited inside ``yield stream.write(...)`` until the OS drained the bytes. Under high event rates (beacons, command returns, flood_events), Runners piled up faster than the OS could flush, and the Runner/generator/frame/Future quadruple was the dominant minion leak. Returning a non-Awaitable lets stream.write enqueue the bytes in Tornado's own write buffer (which Tornado already manages efficiently) and the done-callback handles the disconnect path without spawning a coroutine. c s� z| � � W d S ty t�d� j� � j�|� Y d S tyD } zt�d|� |� � s3|� � � j�|� W Y d }~d S d }~ww )Nrg rh ) r! r r; r< r+ r� �discardr" rr ri r* )r �_streamr# r r r �_on_done+ s ��z,IPCMessagePublisher._write.<locals>._on_doneN)rX r r� r� r )r r[ rY r� r r r r �_write s ��zIPCMessagePublisher._writec C s< | j sdS tjjj|dd�}t| j �D ]}| �||� qdS )z7 Send message to all connected sockets NTr� )r� r/ rU rV rW �tupler� )r rP rY r[ r r r �publish? s �zIPCMessagePublisher.publishc s� t �d|� zNi }� jd dkr � jd |d< t �d� jd � tjj�� j�� t|fi |���W d � n1 s;w Y � j � �� � �fdd�}�� |� W d S t ym } z t � d|� W Y d }~d S d }~ww ) Nrw Zipc_write_bufferr Zmax_write_buffer_sizez'Setting IPC connection write buffer: %sc s � j ��� d S r )r� r� r �r r[ r r �discard_after_closed^ s zCIPCMessagePublisher.handle_connection.<locals>.discard_after_closedrx )r; r< r� r/ rH rI rJ r r r� �addZset_close_callbackr" rr )r ry rz �kwargsr� r# r r� r rK P s& � � ��z%IPCMessagePublisher.handle_connectionc C sL | j rdS d| _ | jD ]}|�� q | j�� t| jd�r$| j�� dS dS r{ )r4 r� r* �clearr r. r� r r r r* e s �zIPCMessagePublisher.closec C r� r r r r r r r� t r� z IPCMessagePublisher.__enter__c G r� r r� r� r r r r� w r� z IPCMessagePublisher.__exit__r ) r$ r% r& r� r rL r� r� rK r* r� r� r r r r r� � s & r� c s� e Zd ZdZddgZdgZd� fdd� Zejj j j ddd ��Z ejj j j d d � �Z dd d �Zejj j j dd� �Z� fdd�Zdd� Z� ZS )�IPCMessageSubscribera� Salt IPC message subscriber Create an IPC client to receive messages from IPC publisher An example of a very simple IPCMessageSubscriber connecting to an IPCMessagePublisher. This example assumes an already running IPCMessagePublisher. IMPORTANT: The below example also assumes the IOLoop is NOT running. # Import Tornado libs import salt.ext.tornado.ioloop # Import Salt libs import salt.config import salt.transport.ipc # Create a new IO Loop. # We know that this new IO Loop is not currently running. io_loop = salt.ext.tornado.ioloop.IOLoop() ipc_publisher_socket_path = '/var/run/ipc_publisher.ipc' ipc_subscriber = salt.transport.ipc.IPCMessageSubscriber(ipc_server_socket_path, io_loop=io_loop) # Connect to the server # Use the associated IO Loop that isn't running. io_loop.run_sync(ipc_subscriber.connect) # Wait for some data package = ipc_subscriber.read_sync() �readr� r* Nc s. t � j||d� d | _g | _t� | _d| _d S )N)r F)r r �_read_stream_future� _saved_datar �_read_in_progressr4 r� r r r r � s z IPCMessageSubscriber.__init__c c s� � z�z | j jdd�V W n tjjjjy tjjj�d��w d}zf | jdu r0| j � d�| _|du r9| jV }n t | j | j|�V }d| _d}t �d|�d }| j � |�V }tjjj|dd �}t|t�rmd |v rm|d }n t�d t|�j� |ryq#n |r�| j �||� q#|} W n7 ty� d}Y n- ty� t�d | j � d| _Y n t y� } zt�d | � d| _W Y d} ~ nd} ~ ww | j � � tjjj�|�� t y� Y dS w )u� Read framed IPC messages. Each message on the wire is: [4-byte big-endian length][msgpack payload]. We read the length prefix first (applying the caller's timeout there), then read exactly that many bytes for the payload — eliminating the streaming-Unpacker approach that was vulnerable to byte interleaving when large messages exceeded PIPE_BUF on the Unix domain socket. When a ``callback`` is provided, this coroutine loops indefinitely, invoking the callback for every received message until the stream is closed. Without a callback, it returns the body of the first message (or None on timeout / closed stream). g:�0�yE>�r NTra rb r Frc re z3IPC subscriber: malformed frame (type=%s), skippingz#Subscriber disconnected from IPC %sz:Exception occurred in Subscriber while handling stream: %s) r� �acquirer/ r0 r1 rN r rO r� r[ rj r r rk rl rH rm rn r= �dictr; r� �typer$ ro r r r< r+ r" �releaser ) r r r� �retrs rt ru rv re r# r r r �_read� sl �� � �� ��� �zIPCMessageSubscriber._readc c s� � | j r| j �d�}tjjj�|��| �� s]z | jdd�V W n9 t y8 t � d| j � tjjj� d�V Y n! tyX } zt �d|� tjjj� d�V W Y d}~nd}~ww | �� r| �|�V }tjjj�|��)z� Asynchronously read messages and invoke a callback when they are ready. :param callback: A callback with the received data r � r� �1Subscriber closed stream on IPC %s before connectr7 �2Exception occurred while Subscriber connecting: %sN)r� �popr/ r0 r1 rN rO r� r� r r; r< r+ r� r" rr r� )r r �resr# r r r r� � s* � � ��� zIPCMessageSubscriber.readc s( � j r � j �d�S � j�� �fdd��S )aZ Read a message from an IPC socket The socket must already be connected. The associated IO Loop must NOT be running. :param int timeout: Timeout when receiving message :return: message data if successful. None if timed out. Will raise an exception for all other error conditions. r c s � � ��S r )r� r �r r r r �<lambda> s z0IPCMessageSubscriber.read_sync.<locals>.<lambda>)r� r� r Zrun_syncr� r r� r � read_sync s z IPCMessageSubscriber.read_syncc c s� � | � � sLz | jdd�V W n9 ty' t�d| j� tjjj � d�V Y n! t yG } zt� d|� tjjj � d�V W Y d}~nd}~ww | � � r| � d|�V dS )z� Asynchronously read messages and invoke a callback when they are ready. :param callback: A callback with the received data r� r� r� r7 r� N)r� r� r r; r< r+ r/ r0 r1 rN r� r" rr r� )r r� r# r r r � read_async" s" � � ��� z IPCMessageSubscriber.read_asyncc s` | j rdS t� �� | jdur(| j�� r*| j�� }|r,t|t�s.t� d|� dS dS dS dS dS )r| Nz!Read future returned exception %r) r4 r r* r� r r� r= r r; rr r� r r r r* 7 s �zIPCMessageSubscriber.closec C s$ | j stjd| ��t| d� d S d S )Nz unclosed ipc message subscriber )�source)r4 �warnings�warn�ResourceWarningr r r r r� I s ��z IPCMessageSubscriber.__del__r )r$ r% r& r� r� r� r r/ r0 r1 rN r^ r� r� r� r� r* r� r' r r r r r� { s" "�� O r� )&r� rq �loggingr? rk r r� Zsalt.ext.tornador/ Zsalt.ext.tornado.concurrentZsalt.ext.tornado.genZsalt.ext.tornado.ioloopZsalt.ext.tornado.netutilZsalt.transport.frameZsalt.utils.msgpackr r r Zsalt.ext.tornado.iostreamr r Zsalt.ext.tornado.locksr � getLoggerr$ r; r r0 r1 r� r� r r( r� r� r� r� r� r r r r �<module> s< - ( :"
| ver. 1.4 |
Github
|
.
| PHP 8.2.30 | Generation time: 0 |
proxy
|
phpinfo
|
Settings