File manager - Edit - /home/newsbmcs.com/public_html/static/img/logo/iocpreactor.tar
Back
iocpsupport.py 0000644 00000000671 15030441410 0007501 0 ustar 00 __all__ = [ "CompletionPort", "Event", "accept", "connect", "get_accept_addrs", "have_connectex", "makesockaddr", "maxAddrLen", "recv", "recvfrom", "send", ] from twisted_iocpsupport.iocpsupport import ( # type: ignore[import] CompletionPort, Event, accept, connect, get_accept_addrs, have_connectex, makesockaddr, maxAddrLen, recv, recvfrom, send, ) const.py 0000644 00000001013 15030441410 0006227 0 ustar 00 # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Windows constants for IOCP """ # this stuff should really be gotten from Windows headers via pyrex, but it # probably is not going to change ERROR_PORT_UNREACHABLE = 1234 ERROR_NETWORK_UNREACHABLE = 1231 ERROR_CONNECTION_REFUSED = 1225 ERROR_IO_PENDING = 997 ERROR_OPERATION_ABORTED = 995 WAIT_TIMEOUT = 258 ERROR_NETNAME_DELETED = 64 ERROR_HANDLE_EOF = 38 INFINITE = -1 SO_UPDATE_CONNECT_CONTEXT = 0x7010 SO_UPDATE_ACCEPT_CONTEXT = 0x700B __pycache__/tcp.cpython-310.pyc 0000644 00000041204 15030441410 0012234 0 ustar 00 o �b�M � @ s� d Z ddlZddlZddlZddlmZ ddlmZmZ ddl m Z m Z m Z m Z mZ ddlmZmZ ddlmZmZ ddlmZmZmZmZmZ dd l m Z dd l m Z dd l m!Z"m#Z#m$Z$m%Z%m&Z&m'Z'm(Z( dd l)m*Z*m+Z+m,Z, zdd l-m.Z/ W n e0y� dZ1Y nw e/Z1eej2eej3iZ4ee e j5e j6�G dd� dej7e(e#��Z8e1dur�ee8e j9� G dd� de$e%e8�Z:G dd� de8�Z;G dd� de"�Z!ee j<�G dd� de(e��Z=dS )z TCP support for IOCP reactor � N)�Optional)�classImplements� implementer)�address�defer�error� interfaces�main)� _LogOwner� isIPv6Address)�abstract� iocpsupport)�ERROR_CONNECTION_REFUSED�ERROR_IO_PENDING�ERROR_NETWORK_UNREACHABLE�SO_UPDATE_ACCEPT_CONTEXT�SO_UPDATE_CONNECT_CONTEXT)�IReadWriteHandle)�Protocol)� Connector�_AbortingMixin�_BaseBaseClient�_BaseTCPClient� _getsockname� _resolveIPv6� _SocketCloser)�failure�log�reflect)�startTLSc @ s� e Zd ZdZdZd/dd�Zdd� Zdd � Zd d � Zd d � Z dd� Z dd� Z dd� Z dd� Z dd� Zdd� Zdd� Zd d � ZedurFd0d d �Zd!d"� Zd#d$� Zd/d%d&�Zd'd(� Zd)d*� Zd+d,� Zd-d.� ZdS )1� Connectionz� @ivar TLS: C{False} to indicate the connection is in normal TCP mode, C{True} to indicate that TLS has been started and that operations must be routed through the L{TLSMemoryBIOProtocol} instance. FNc C s& t j�| |� || _|j| _|| _d S �N)r � FileHandle�__init__�socket�fileno� getFileHandle�protocol)�self�sock�proto�reactor� r, �B/usr/lib/python3/dist-packages/twisted/internet/iocpreactor/tcp.pyr# ? s zConnection.__init__c C � | j S r! )r$ �r( r, r, r- � getHandleE s zConnection.getHandlec C s@ t |t�rnt |t�rt|�}ntdt|� ��| j�|� dS )z` @param rbuffer: Data received. @type rbuffer: L{bytes} or L{bytearray} z%data must be bytes or bytearray, not N)� isinstance�bytes� bytearray� TypeError�typer' � dataReceived)r( �rbufferr, r, r- r6 H s zConnection.dataReceivedc C s t �| �� ||�S r! )�_iocp�recvr& )r( �bufflist�evtr, r, r- �readFromHandleW s zConnection.readFromHandlec C s( t |�}t�| �� |d| j� �� |�S )z� Send C{buff} to current file handle using C{_iocp.send}. The buffer sent is limited to a size of C{self.SEND_LIMIT}. r )� memoryviewr8 �sendr& � SEND_LIMIT�tobytes)r( �buffr; � writeViewr, r, r- � writeToHandleZ s �zConnection.writeToHandlec C sx z| j �d� W n ty Y nw t�| jd �}|r:z|�� W d S ty9 t� � }t � � | � |� Y d S w d S )N� ) r$ �shutdown�OSErrorr �IHalfCloseableProtocolr' �writeConnectionLost� BaseExceptionr �Failurer �err�connectionLost)r( �p�fr, r, r- �_closeWriteConnectiond s � ��z Connection._closeWriteConnectionc C sZ t �| jd �}|r&z|�� W d S ty% t�� | �t� � � Y d S w | �|� d S r! ) r rG r' �readConnectionLostrI r rK rL r rJ )r( �reasonrM r, r, r- rP r s �z Connection.readConnectionLostc C sX | j rd S tj�| |� |d u p|�tj� }| �|� | j}| `| ` | ` |�|� d S r! ) � disconnectedr r" rL �checkr �ConnectionAborted� _closeSocketr' r$ r& )r( rQ �isCleanr' r, r, r- rL } s zConnection.connectionLostc C r. )zN Return the prefix to log with when I own the logging thread. )�logstrr/ r, r, r- � logPrefix� � zConnection.logPrefixc C � t | j�tjtj��S r! )�boolr$ � getsockopt� IPPROTO_TCP� TCP_NODELAYr/ r, r, r- � getTcpNoDelay� � zConnection.getTcpNoDelayc C � | j �t jt j|� d S r! )r$ � setsockoptr] r^ �r( �enabledr, r, r- � setTcpNoDelay� � zConnection.setTcpNoDelayc C rZ r! )r[ r$ r\ � SOL_SOCKET� SO_KEEPALIVEr/ r, r, r- �getTcpKeepAlive� r` zConnection.getTcpKeepAlivec C ra r! )r$ rb rg rh rc r, r, r- �setTcpKeepAlive� rf zConnection.setTcpKeepAliveTc C s t | ||tj� dS )z9 @see: L{ITLSTransport.startTLS} N)� _startTLSr r" )r( �contextFactory�normalr, r, r- r � s zConnection.startTLSc C �2 | j rdS | jr| j�|� dS tj�| |� dS )z� Write some data, either directly to the underlying handle or, if TLS has been started, to the L{TLSMemoryBIOProtocol} for it to encrypt and send. @see: L{twisted.internet.interfaces.ITransport.write} N)rR �TLSr' �writer r" )r( �datar, r, r- rp � � zConnection.writec C rn )z� Write some data, either directly to the underlying handle or, if TLS has been started, to the L{TLSMemoryBIOProtocol} for it to encrypt and send. @see: L{twisted.internet.interfaces.ITransport.writeSequence} N)rR ro r' � writeSequencer r" )r( �iovecr, r, r- rs � rr zConnection.writeSequencec C s: | j r| jr| js| j�� dS dS dS tj�| |� dS )z� Close the underlying handle or, if TLS has been started, first shut it down. @see: L{twisted.internet.interfaces.ITransport.loseConnection} N)ro � connected� disconnectingr' �loseConnectionr r" )r( rQ r, r, r- rw � s �zConnection.loseConnectionc C s, | j r | j�||� dS tj�| ||� dS )zc Register a producer. If TLS is enabled, the TLS connection handles this. N)ro r' �registerProducerr r" )r( �producer� streamingr, r, r- rx � s zConnection.registerProducerc C s$ | j r | j�� dS tj�| � dS )ze Unregister a producer. If TLS is enabled, the TLS connection handles this. N)ro r' �unregisterProducerr r" r/ r, r, r- r{ � s z Connection.unregisterProducerc C � d S r! r, r/ r, r, r- �getHost� � zConnection.getHostc C r| r! r, r/ r, r, r- �getPeer� r~ zConnection.getPeerr! )T)�__name__� __module__� __qualname__�__doc__ro r# r0 r6 r< rC rO rP rL rX r_ re ri rj rk r rp rs rw rx r{ r} r r, r, r, r- r 5 s2 r c @ sT e Zd ZdZejZejZdZ e Z dd� Z dd� Z dd� Zd d � Zd d � Zd d� ZdS )�Clientz� @ivar _tlsClientDefault: Always C{True}, indicating that this is a client connection, and by default when TLS is negotiated this class will act as a TLS client. Tc C s* |d u rd}|| _ t�| |||||� d S )N)� r )r+ r r# )r( �host�port� bindAddress� connectorr+ r, r, r- r# s zClient.__init__c C s | j �| j| j�S )zd Create a socket registered with the IOCP reactor. @see: L{_BaseTCPClient} )r+ � createSocket� addressFamily� socketTyper/ r, r, r- �createInternetSocket s zClient.createInternetSocketc C s | ` | `dS )z� Clean up potentially circular references to the socket and to its C{getFileHandle} method. @see: L{_BaseBaseClient} N)r$ r& r/ r, r, r- �_collectSocketDetails s z Client._collectSocketDetailsc C s | j �| � dS )z^ Remove the active handle from the reactor. @see: L{_BaseBaseClient} N)r+ �removeActiveHandler/ r, r, r- �_stopReadingAndWriting s z Client._stopReadingAndWritingc C s� |rt �||�}| �t�|tj�|d�f�� d S | j�tj t t � d| j� � �� | j�| �� �| _d| _| �| j�}|d | _| jd u rOt� | _| �� d S | j�| � | �� d S )Nz Unknown error�PTz,client)�connectExErrors�get�failIfNotConnectedr �getConnectError�errno� errorcoder$ rb rg r �struct�packr% r� � buildProtocolr r' ru � _getLogPrefixrW r rw �makeConnection� startReading)r( �rcrq r; rX r, r, r- � cbConnect" s& �� zClient.cbConnectc C sn t | d�sd S tjs J �| j�| � t�| j| �}t�| j� � | j |�}|r3|t kr5| �|d|� d S d S d S )Nr� r ) �hasattrr8 �have_connectexr+ �addActiveHandle�Eventr� �connectr$ r% � realAddressr )r( r; r� r, r, r- � doConnect@ s �zClient.doConnectN)r� r� r� r� r$ �AF_INETr� � SOCK_STREAMr� �_tlsClientDefaultr �_commonConnectionr# r� r� r� r� r� r, r, r, r- r� � s r� c @ s: e Zd ZdZdZdd� Zdefdd�Zdd � Zd d � Z d S ) �ServeraV Serverside socket-stream connection class. I am a serverside network connection transport; a socket which came from an accept() on a server. @ivar _tlsClientDefault: Always C{False}, indicating that this is a server connection, and by default when TLS is negotiated this class will act as a TLS server. Fc C sv t �| |||� || _|| _|| _| �| j�}|� d|� d| jj� �| _d� | jj j | j| jj �| _ d| _| �� dS )a Server(sock, protocol, client, server, sessionno) Initialize me with a socket, a protocol, a descriptor for my peer (a tuple of host, port describing the other end of the connection), an instance of Port, and a session number. �,z<{} #{} on {}>TN)r r# � serverAddr� clientAddr� sessionnor� r' r� rW �format� __class__r� r� �repstrru r� )r( r) r' r� r� r� r+ rX r, r, r- r# \ s � zServer.__init__�returnc C r. )z= A string representation of this connection. )r� r/ r, r, r- �__repr__r rY zServer.__repr__c C r. )zW Returns an IPv4Address. This indicates the server's address. )r� r/ r, r, r- r} x � zServer.getHostc C r. )zW Returns an IPv4Address. This indicates the client's address. )r� r/ r, r, r- r � r� zServer.getPeerN) r� r� r� r� r� r# �strr� r} r r, r, r, r- r� N s r� c @ s e Zd Zdd� ZdS )r c C s t | j| j| j| | j�S r! )r� r� r� r� r+ r/ r, r, r- �_makeTransport� rf zConnector._makeTransportN)r� r� r� r� r, r, r, r- r � s r c @ s� e Zd ZU dZdZdZejZej Z e j Z dZdZee ed< dZd dd �Zd efd d �Zd d� Ze�ej�fdd�Z e Z dd� Z dd� Z dd� Z dd� Z!dd� Z"dd � Z#d d � Z$dS ) �PortFr N�_realPortNumber�TCP�2 r� c C s> || _ || _|| _|| _|| _t|�r tj| _t j | _ d S d S r! ) r� �factory�backlog� interfacer+ r r$ �AF_INET6r� r � IPv6Address� _addressType)r( r� r� r� r� r+ r, r, r- r# � s �z Port.__init__r� c C s2 | j d urd�| j| jj| j �S d�| j| jj�S )Nz<{} of {} on {}>z<{} of {} (not listening)>)r� r� r� r� r/ r, r, r- r� � s ��z Port.__repr__c C s� z$| j �| j| j�}| jtjkrt| j| j�}n| j| jf}|� |� W n t y9 } z t � | j| j|��d }~ww t �|�� �| _|�� d | _t�d| �| j�| jf � | j�� |�| j� d| _d| _| j � | � || _| jj| _ | � � d S )NrD z%s starting on %sTF) r+ r� r� r� r$ r� r r� r� �bindrF r �CannotListenErrorr8 � maxAddrLenr% �addrLen� getsocknamer� r �msgr� r� �doStart�listenr� ru rR r� r& �doAccept)r( �skt�addr�ler, r, r- �startListening� s2 ���� zPort.startListeningc C s2 d| _ | jrt�� | _| j�d| j|� | jS dS )z� Stop accepting connections on this port. This will shut down my socket and call self.connectionLost(). It returns a deferred which will fire successfully when the port is actually closed. Tr N)rv ru r �Deferred�deferredr+ � callLaterrL )r( �connDoner, r, r- rw � s �zPort.loseConnectionc C s t �d| j� d| j� d�� dS )z. Log message for closing port �(z Port z Closed)N)r r� �_typer� r/ r, r, r- �_logConnectionLostMsg� s zPort._logConnectionLostMsgc C s� | � � d| _d}t| d�r| j}| `d| _| j�| � d| _| �d� | ` | ` z| j � � W n t yH d| _|durG|�t�� � Y dS � w d| _|durW|�d� dS dS )z' Cleans up the socket. Nr� TF)r� r� r� r� rR r+ r� ru rU r$ r& r� �doStoprI rv �errbackr rJ �callback)r( rQ �dr, r, r- rL � s0 ��zPort.connectionLostc C s t �| jj�S )zK Returns the name of my class, to prefix log entries with. )r �qualr� r� r/ r, r, r- rX s zPort.logPrefixc C s | j dgt| j��R � S )zf Returns an IPv4Address or IPv6Address. This indicates the server's address. r� )r� r r$ r/ r, r, r- r} s z Port.getHostc C s, | � ||� | js| js| �� d S d S d S r! )� handleAcceptrv rR r� )r( r� rq r; r, r, r- �cbAccept s �z Port.cbAcceptc C sh | j s| jrdS |rt�dtj�|d�|f � dS |j�t j t t � d| j �� �� t�|j�� |j�\}}}|| jks>J �d|d v rYt|d �d�d �}|d |d d|f}d|d v rtt|d �d�d �}|d |d d|f}| j�| jdg|�R � �}|d u r�|j�� d S | j}|d | _t|j|| jdg|�R � | jdg|�R � || j�} |� | � d S ) NFz*Could not accept new connection -- %s (%s)z unknown errorr� �%r rD r� T) rv rR r r� r� r� r� �newsktrb r$ rg r r� r� r% r8 �get_accept_addrsrA r� �int�splitr� r� r� �closer� r� r+ r� ) r( r� r; �family�lAddr�rAddr�scoper' �s� transportr, r, r- r� " sL ��� � � zPort.handleAcceptc C s| t �| j| �}td| jd � |_}| j�| j| j � |_ }t � | j � � |� � ||�}|r:|tkr<| �||� d S d S d S )N� � )r8 r� r� r3 r� rA r+ r� r� r� r� �acceptr$ r% r r� )r( r; rA r� r� r, r, r- r� U s � �z Port.doAccept)r� r� N)%r� r� r� ru rR rv r$ r� r� r� r� r � IPv4Addressr� r� r� r r� �__annotations__r� r# r� r� r� r rJ r �CONNECTION_DONErw � stopListeningr� rL rX r} r� r� r� r, r, r, r- r� � s, 3r� )>r� r� r$ r� �typingr �zope.interfacer r �twisted.internetr r r r r �twisted.internet.abstractr r � twisted.internet.iocpreactorr r r8 �"twisted.internet.iocpreactor.constr r r r r �'twisted.internet.iocpreactor.interfacesr �twisted.internet.protocolr �twisted.internet.tcpr � TCPConnectorr r r r r r �twisted.pythonr r r �twisted.internet._newtlsr � __startTLS� ImportErrorrk �WSAECONNREFUSED�WSAENETUNREACHr� � ITCPTransport� ISystemHandler" r � ITLSTransportr� r� �IListeningPortr� r, r, r, r- �<module> sB $ �� < Y; __pycache__/reactor.cpython-310.pyc 0000644 00000016667 15030441410 0013124 0 ustar 00 o �b�$ � @ sd U d Z ddlZddlZddlZddlmZmZ ddlmZ ddl m Z m Z m Z m Z ddlmZ ddlmZmZmZ ddlmZ dd lmZ dd lmZmZ zdd l m Z W n e yr dZ d Z eee j! d f e"d< e�#d� Y nw e Z e j!fZ dZ$dZ%e&d�\Z'Z(e �)d�Z*e �)d�Z+ee j,e j-e j.e j/ge �R � G dd� de j0e j1e��Z2dd� Z3ddgZ4dS )z' Reactor that uses IO completion ports � N)�Tuple�Type)� implementer)�base�error� interfaces�main)�Process)� iocpsupport�tcp�udp)� WAIT_TIMEOUT)�_ThreadedWin32EventsMixin)�failure�log)�TLSMemoryBIOFactory� .�_extraInterfacesz|pyOpenSSL 0.10 or newer is required for SSL support in iocpreactor. It is missing, so the reactor will not support SSL APIs.i� �� � z#Handler has no getFileHandle methodzFiledescriptor went awayc @ s� e Zd ZdZdd� Zdd� Zdd� Zdd � Zd d � Zd d � Z dd� Z dd� Z dd� Z d-dd�Z d.dd�ZedurDd-dd �Z d.d d �Zn d-d d �Z d.d d �Zd/d"d#�Z $d0d%d&�Zd'i dddd(dfd)d*�Zd+d,� ZdS )1� IOCPReactorNc C s" t j�| � t�� | _t� | _d S �N)r � ReactorBase�__init__�_iocp�CompletionPort�port�set�handles��selfr r �F/usr/lib/python3/dist-packages/twisted/internet/iocpreactor/reactor.pyr ? s zIOCPReactor.__init__c C � | j �|� d S r )r �add�r �handler r r! �addActiveHandleD � zIOCPReactor.addActiveHandlec C r" r )r �discardr$ r r r! �removeActiveHandleG r' z IOCPReactor.removeActiveHandlec C s� d}|du r t }n tt td| ��}| j�|�\}}}} |tkr#dS |tkr<|tks-J �t� |j | j |||� |d7 }|t krBdS | j�d�\}}}}q )z= Poll the IO completion port for new events. r Nr � ) � MAX_TIMEOUT�min�intr �getEventr � KEY_WAKEUP� KEY_NORMALr �callWithLogger�owner�_callEventCallback�EVENTS_PER_LOOP)r �timeout�processed_events�rc�numBytes�key�evtr r r! � doIterationJ s$ ��zIOCPReactor.doIterationc C s� |j }d }z |�|||� t|dd �}|st}n|� dkr t}|r$W d S W n ty8 t�� d }t� � Y nw |rE|� t � |�� d S d S )N� getFileHandle���r* ) r2 �callback�getattr� _NO_GETHANDLE� _NO_FILEDESC� BaseException�sys�exc_infor �err�loseConnectionr �Failure)r r7 r8 r: r2 �why�handfnr r r! r3 v s& � ��z IOCPReactor._callEventCallbackc C s d S r r r r r r! � installWaker� s zIOCPReactor.installWakerc C s | j �dtd � d S )Nr )r � postEventr/ r r r r! �wakeUp� s zIOCPReactor.wakeUpc C s | j �|t� d S r )r � addHandler0 r$ r r r! �registerHandle� s zIOCPReactor.registerHandlec C s t � ||�}| �|�� � |S r )�socketrN �fileno)r �af�stype�sktr r r! � createSocket� s zIOCPReactor.createSocket�2 � c C � t �||||| �}|�� |S )zI @see: twisted.internet.interfaces.IReactorTCP.listenTCP )r �Port�startListening)r r �factory�backlog� interface�pr r r! � listenTCP� s zIOCPReactor.listenTCP� c C s t �|||||| �}|�� |S )zJ @see: twisted.internet.interfaces.IReactorTCP.connectTCP )r � Connector�connect)r �hostr rZ r5 � bindAddress�cr r r! � connectTCP� s zIOCPReactor.connectTCPc C s" | � |t|d|�||�}d|_|S )zQ @see: twisted.internet.interfaces.IReactorSSL.listenSSL F�TLS)r^ r �_type�r r rZ �contextFactoryr[ r\ r r r! � listenSSL� s �zIOCPReactor.listenSSLc C s | � ||t|d|�||�S )zR @see: twisted.internet.interfaces.IReactorSSL.connectSSL T)re r �r rb r rZ ri r5 rc r r r! � connectSSL� s �zIOCPReactor.connectSSLc C � t d��)z� Non-implementation of L{IReactorSSL.listenSSL}. Some dependency is not satisfied. This implementation always raises L{NotImplementedError}. �|pyOpenSSL 0.10 or newer is required for SSL support in iocpreactor. It is missing, so the reactor does not support SSL APIs.��NotImplementedErrorrh r r r! rj � s �c C rm )z� Non-implementation of L{IReactorSSL.connectSSL}. Some dependency is not satisfied. This implementation always raises L{NotImplementedError}. rn ro rk r r r! rl � s �� c C rW )z� Connects a given L{DatagramProtocol} to the given numeric UDP port. @returns: object conforming to L{IListeningPort}. )r rX rY )r r �protocolr\ � maxPacketSizer] r r r! � listenUDP� s zIOCPReactor.listenUDPFc C s t �||||| |�}|�� |S )z� Connects a given DatagramProtocol to the given numeric UDP port. EXPERIMENTAL. @returns: object conforming to IListeningPort. )r � MulticastPortrY )r r rr r\ rs �listenMultipler] r r r! �listenMulticast� s �zIOCPReactor.listenMulticastr r c C sN |durt d��|durt d��|rt d��| dur t d��t| |||||�S )z" Spawn a process. Nz,Setting UID is unsupported on this platform.z,Setting GID is unsupported on this platform.z&PTYs are unsupported on this platform.zGCustom child file descriptor mappings are unsupported on this platform.)� ValueErrorr ) r �processProtocol� executable�args�env�path�uid�gid�usePTY�childFDsr r r! � spawnProcess� s �zIOCPReactor.spawnProcessc C s t | j�}| j�� |S r )�listr �clear)r �resr r r! � removeAll s zIOCPReactor.removeAll)rU rV )r_ N)rV rq )rV rq F)�__name__� __module__� __qualname__r r r& r) r; r3 rJ rL rN rT r^ re r rj rl rt rw r� r� r r r r! r 2 s@ , � � � � r c C s t � } t�| � d S r )r r �installReactor)�rr r r! �install s r� )5�__doc__rO rC �warnings�typingr r �zope.interfacer �twisted.internetr r r r � twisted.internet._dumbwin32procr � twisted.internet.iocpreactorr r r r �"twisted.internet.iocpreactor.constr �"twisted.internet.win32eventreactorr �twisted.pythonr r �twisted.protocols.tlsr �_TLSMemoryBIOFactory� ImportErrorr � IReactorSSL�__annotations__�warnr+ r4 �ranger0 r/ �ConnectionFdescWentAwayr@ rA � IReactorTCP� IReactorUDP�IReactorMulticast�IReactorProcess�_SignalReactorMixinr r r� �__all__r r r r! �<module> sT �� �� � c __pycache__/abstract.cpython-310.pyc 0000644 00000024301 15030441410 0013250 0 ustar 00 o �b3 � @ s� d Z ddlZddlmZ ddlmZmZmZ ddlm Z m Z m Z ddl m Z ddlmZmZ ddlmZ eejejejej�G d d � d e e ��Zd gZdS ) z Abstract file handle class � N)� implementer)�error� interfaces�main)�_ConsumerMixin�_dataMustBeBytes� _LogOwner)� iocpsupport)�ERROR_HANDLE_EOF�ERROR_IO_PENDING)�failurec @ sn e Zd ZdZdZdZdZdZdZdZ dZ dZ dd � Z d d � Z d d � Zdd� Zdd� Zdd� Zdd� Zdd� Zdd� Zdd� Zd ZdZdZdZdZdZd Z d d � Z d d!� Z d"d#� Z d$d%� Z d&d'� Z!d(d)� Z"d*d+� Z#d,d-� Z$d.d/� Z%d0d1� Z&d2d3� Z'd4d5� Z(dZ)dZ*dZ+d6Z,d7Z-dLd8d9�Z.d:d;� Z/d<d=� Z0e1�2e3j4�fd>d?�Z5d@dA� Z6dBdC� Z7dDdE� Z8dFdG� Z9dHdI� Z:dJdK� Z;dS )M� FileHandlez< File handle that can read and write asynchronously � i FTr Nc C �< | j �| � | js| js d| _| j �d| j�| _d S d S d S �NTr )�reactor�addActiveHandle�_readScheduled�reading� callLater�_resumeReading��self� r �G/usr/lib/python3/dist-packages/twisted/internet/iocpreactor/abstract.py� startReading( s �zFileHandle.startReadingc C � | j r | j �� d | _ d| _d S �NF)r �cancelr r r r r � stopReading. � zFileHandle.stopReadingc C s( d | _ | �� r| js| �� d S d S d S �N)r � _dispatchData�_readScheduledInOS�doReadr r r r r 4 s �zFileHandle._resumeReadingc C s� | j s| jS | j }|| j }| j|k r-| �| j| j � | jd7 _| js(dS | j|k s|| j }|r@| �| j| d|� � | jrl| jt| j� }||| j k rW| jd= n||krlt| j�| jk rl| j� t | j�� d| _d| _ | jS )zt Dispatch previously read data. Return True if self.reading and we don't have any more data � Fr ���) � _readSizer �readBufferSize�_readNextBuffer� dataReceived� _readBuffers�dynamicReadBuffers�len�maxReadBuffers�append� bytearray)r �size� full_buffers� remainder�total_buffer_sizer r r r" 9 s. � zFileHandle._dispatchDatac C s$ d| _ | �|||�r| �� d S d S r )r# � _handleReadr$ �r �rc�data�evtr r r �_cbReadY s �zFileHandle._cbReadc C s� | j rdS |s |r|tjtfv r!| j�| � | �t�t j �� dS |r9| � t�t � dtj�|d�|f ��� dS | jdks@J �| jdksGJ �|| _| �� S )zA Returns False if we should stop reading for now Fzread error -- %s (%s)�unknownr )� disconnected�errno� WSAEDISCONr r �removeActiveHandle�readConnectionLostr �Failurer �CONNECTION_DONE�connectionLostr �ConnectionLost� errorcode�getr' r) r" r6 r r r r5 ^ s* ����zFileHandle._handleReadc C sR t �| j| �}| j |_}| �||�\}}|r|tkr d| _d S | �|||� d S �NT) �_iocp�Eventr: r+ �buff�readFromHandler r# r5 )r r9 rJ r7 � numBytesReadr r r r$ { s zFileHandle.doReadc C � t � �r! ��NotImplementedError)r �bufflistr9 r r r rK � � zFileHandle.readFromHandlec C s t �r! rN �r r8 r r r r* � � zFileHandle.dataReceivedc C � | � |� d S r! �rC �r �reasonr r r r@ � s z FileHandle.readConnectionLost� i c C s d| _ | �� d S rG )�_writeDisconnecting� startWritingr r r r �loseWriteConnection� � z FileHandle.loseWriteConnectionc C s d S r! r r r r r �_closeWriteConnection� s z FileHandle._closeWriteConnectionc C rT r! rU rV r r r �writeConnectionLost� s z FileHandle.writeConnectionLostc C r r )r r �_writeScheduled�writingr �_resumeWritingr r r r rZ � s �zFileHandle.startWritingc C r r )r_ r r` r r r r � stopWriting� r zFileHandle.stopWritingc C s d | _ | �� d S r! )r_ �doWriter r r r ra � r\ zFileHandle._resumeWritingc C s | � |||�r | �� d S d S r! )� _handleWriterc �r r7 �numBytesWrittenr9 r r r �_cbWrite� s �zFileHandle._cbWritec C s� | j s| jrdS |r | �t�t�dtj� |d�|f ��� dS | j |7 _ | j t | j �krk| j skd| _ d| _ | �� | jdurQ| jrG| jrQd| _| j�� dS | jr_| �t�tj�� dS | jrid| _| �� dS dS )zA Returns false if we should stop writing for now Fzwrite error -- %s (%s)r; rX r NT)r< �_writeDisconnectedrC r rA r rD r= rE rF �offsetr- � dataBuffer� _tempDataLenrb �producer�streamingProducer�producerPaused�resumeProducing� disconnectingr rB rY r] re r r r rd � sD ���� �� ��zFileHandle._handleWritec C s� t | j�| j | jk r#| j| jd � d�| j� | _d| _g | _d| _t�| j | �}| jr=t | j�}|| jd � |_ }n| j |_ }| � ||�\}}|rZ|t kr\| �|||� d S d S d S )NrX r )r- rj ri � SEND_LIMIT�join�_tempDataBufferrk rH rI rg � memoryviewrJ � writeToHandler rd )r r9 �sendViewrJ r7 r8 r r r rc � s � �zFileHandle.doWritec C rM r! rN )r rJ r9 r r r ru � rQ zFileHandle.writeToHandlec C s� t |� | jr | jr dS |r>| j�|� | jt|�7 _| jdur8| jr8t| j �| j | j kr8d| _ | j� � | � � dS dS )zpReliably write some data. The data is buffered until his file descriptor is ready for writing. NT)r � connectedrh rs r/ rk r- rl rm rj �writeBufferSizern �pauseProducingrZ rR r r r �write s �zFileHandle.writec C s� |D ]}t |� q| jr|r| jrd S | j�|� |D ] }| jt|�7 _q| jd urB| jrBt| j �| j | j krBd| _ | j� � | � � d S rG )r rw rh rs �extendrk r- rl rm rj rx rn ry rZ )r �iovec�ir r r � writeSequence s zFileHandle.writeSequence� Uninitializedi c C s4 |sddl m} || _g | _d| _t| j�g| _d S )Nr )r )�twisted.internetr rs rk r0 r( r+ )r r r r r �__init__' s zFileHandle.__init__c C sF d| _ d| _| jdur| j�� d| _| �� | �� | j�| � dS )a� The connection was lost. This is called when the connection on a selectable object has been lost. It will be called whether the connection was closed explicitly, an exception occurred in an event handler, or the other end of the connection closed it first. Clean up state here, but make sure to call back up to FileDescriptor. TFN)r< rw rl � stopProducingr rb r r? rV r r r rC / s zFileHandle.connectionLostc C s dS )Nr& r r r r r � getFileHandleD rS zFileHandle.getFileHandlec C sP | j r$| js&| jr| �� | j | �|� dS | �� | �� d| _dS dS dS )ad Close the connection at the next available opportunity. Call this to cause this FileDescriptor to lose its connection. It will first write any data that it has buffered. If there is data buffered yet to be written, this method will cause the transport to lose its connection as soon as it's done flushing its write buffer. If you have a producer registered, the connection won't be closed until the producer is finished. Therefore, make sure you unregister your producer when it's finished, or the connection will never close. r% N)rw rp rh r rb rC rZ )r � _connDoner r r �loseConnectionG s �zFileHandle.loseConnectionc C s | � � | �� dS )z� Stop consuming data. This is called when a producer has lost its connection, to tell the consumer to go lose its connection (and break potential circular references). N)�unregisterProducerr� r r r r � stopConsumingc s zFileHandle.stopConsumingc C s | j r | js| �� d S d S d S r! )rw rp r r r r r ro p s �zFileHandle.resumeProducingc C � | � � d S r! )r r r r r ry t � zFileHandle.pauseProducingc C r� r! )r� r r r r r� w r� zFileHandle.stopProducingc C rM r! rN r r r r �getHostz � zFileHandle.getHostc C rM r! rN r r r r �getPeer~ r� zFileHandle.getPeerr! )<�__name__� __module__� __qualname__�__doc__r. r( r r, r) r' r r# r r r r" r: r5 r$ rK r* r@ rj ri r` r_ rY rh rx r[ r] r^ rZ rb ra rg rd rc ru rz r~ rw r< rp �logstrrq r� rC r� r rA r rB r� r� ro ry r� r� r� r r r r r sl - r )r� r= �zope.interfacer r� r r r �twisted.internet.abstractr r r � twisted.internet.iocpreactorr rH �"twisted.internet.iocpreactor.constr r �twisted.pythonr � IPushProducer� IConsumer� ITransport�IHalfCloseableDescriptorr �__all__r r r r �<module> s$ � l __pycache__/__init__.cpython-310.pyc 0000644 00000000455 15030441410 0013210 0 ustar 00 o �b� � @ s d Z ddlmZ dgZdS )z I/O Completion Ports reactor � )�installr N)�__doc__�$twisted.internet.iocpreactor.reactorr �__all__� r r �G/usr/lib/python3/dist-packages/twisted/internet/iocpreactor/__init__.py�<module> s __pycache__/udp.cpython-310.pyc 0000644 00000031353 15030441410 0012242 0 ustar 00 o �b\6 � @ s� d Z ddlZddlZddlZddlZddlmZ ddlmZ ddl m Z m Z m Z m Z ddlmZmZ ddlmZmZ ddlmZmZmZ dd lmZ dd lm Z m Z eee j e j e j �G d d � d ej!��Z"G d d� d�Z#ee j$�G dd� de#e"��Z%dS )z UDP support for IOCP reactor � N)�Optional)� implementer)�address�defer�error� interfaces)� isIPAddress� isIPv6Address)�abstract� iocpsupport)�ERROR_CONNECTION_REFUSED�ERROR_IO_PENDING�ERROR_PORT_UNREACHABLE)�IReadWriteHandle)�failure�logc @ s� e Zd ZU dZejZejZdZ dZ e e e d< d6dd�Zd d � Zd efd d �Zdd� Zdd� Zdd� Zdd� Zdd� Zdd� Zdd� Zd d � Zd7d d �Zd d!� Zd"d#� Z d$d%� Z d&d'� Z d(d)� Z d7d*d+�Z d,d-� Z!d.d/� Z"d0d1� Z#d2d3� Z$d4d5� Z%dS )8�Portz� UDP port, listening for packets. @ivar addressFamily: L{socket.AF_INET} or L{socket.AF_INET6}, depending on whether this port is listening on an IPv4 address or an IPv6 address. FN�_realPortNumber� � c C sx || _ || _|| _|| _| �� d| _| �� tj� | |� t � | j | j �}t �|�� �}t|�| _tt�d��| _dS )z> Initialize with a numeric port to listen on. N�i)�port�protocol�readBufferSize� interface� setLogStr�_connectedAddr�_setAddressFamilyr � FileHandle�__init__�socket� addressFamily� socketType�_iocp� maxAddrLen�fileno� bytearray� addressBuffer�struct�calcsize�addressLengthBuffer)�selfr �protor � maxPacketSize�reactor�skt�addrLen� r1 �B/usr/lib/python3/dist-packages/twisted/internet/iocpreactor/udp.pyr 2 s z Port.__init__c C sD t | j�r tj| _dS t| j�rtj| _dS | jr t�| jd��dS )z8 Resolve address family for the socket. znot an IPv4 or IPv6 addressN) r r r �AF_INET6r! r �AF_INETr �InvalidAddressError�r+ r1 r1 r2 r F s ��zPort._setAddressFamily�returnc C s2 | j d urd| jj� d| j � d�S d| jj� d�S )N�<z on �>z not connected>)r r � __class__r6 r1 r1 r2 �__repr__S s z Port.__repr__c C � | j S )z) Return a socket object. )r r6 r1 r1 r2 � getHandleY � zPort.getHandlec C s | � � | �� dS )z� Create and bind my socket, and begin listening on it. This is called on unserialization, and must be called after creating a server to begin listening on the specified port. N)� _bindSocket�_connectToProtocolr6 r1 r1 r2 �startListening_ s zPort.startListeningc C s | j �| j| j�S �N)r. � createSocketr! r" r6 r1 r1 r2 rC i � zPort.createSocketc C s� z| � � }|�| j| jf� W n ty$ } z t�| j| j|��d }~ww |�� d | _t � d| � | j �| jf � d| _ || _| jj| _d S )N� z%s starting on %sT)rC �bindr r �OSErrorr �CannotListenError� getsocknamer r �msg� _getLogPrefixr � connectedr r% � getFileHandle)r+ r/ �ler1 r1 r2 r? l s ����zPort._bindSocketc C s$ | j �| � | �� | j�| � d S rB )r �makeConnection� startReadingr. �addActiveHandler6 r1 r1 r2 r@ � s zPort._connectToProtocolc C s$ | j r| �|||� | �� d S d S rB )�reading� handleRead�doRead�r+ �rc�data�evtr1 r1 r2 �cbRead� s �z Port.cbReadc C s� |t jt jttfv r| jr| j�� d S d S |r(t� dt j � |d�|f � d S z| j� t |jd |� �t�|j�� W d S tyK t�� Y d S w )Nz error in recvfrom -- %s (%s)z unknown error)�errno�WSAECONNREFUSED� WSAECONNRESETr r r r �connectionRefusedr rJ � errorcode�get�datagramReceived�bytes�buffr# � makesockaddr� addr_buff� BaseException�errrU r1 r1 r2 rS � s, ���� � �zPort.handleReadc C sv t �| j| �}| jd |_}| j |_}| j |_}t � | � � ||||�\}}|r7|t kr9| � |||� d S d S d S �Nr ) r# �EventrY � _readBuffersrb r' rd r* � addr_len_buff�recvfromrM r rS )r+ rX rb rd rj rV rW r1 r1 r2 rT � s � �z Port.doReadc C s� | j rW|d| j fv s J �z| j�|�W S tyV } z7|jd }|tjkr0| �|�W Y d}~S |tjkr:t � d��|tj tj t tfv rJ| j�� n� W Y d}~dS d}~ww |dks]J �t|d �swt|d �sw|d dkrwt �|d d��t|d �r�| jtjkr�t �|d d��t|d �r�| jtjkr�t �|d d��z| j�||�W S ty� } z2|jd }|tjkr�| �||�W Y d}~S |tjkr�t � d��|tj tj t tfv r�W Y d}~dS � d}~ww )z~ Write a datagram. @param addr: should be a tuple (ip, port), can be None in connected mode. Nr zmessage too longz <broadcast>z0write() only accepts IP addresses, not hostnamesz*IPv6 port write() called with IPv4 addressz*IPv4 port write() called with IPv6 address)r r �sendrG �argsrZ �WSAEINTR�write� WSAEMSGSIZEr �MessageLengthErrorr[ r\ r r r r] r r r5 r! r3 r4 �sendto)r+ �datagram�addr�se�nor1 r1 r2 ro � sp � ��� � � ��� � ��z Port.writec C s | � d�|�|� d S )N� )ro �join)r+ �seqrt r1 r1 r2 � writeSequence� s zPort.writeSequencec C sH | j rtd��t|�st|�st�|d��||f| _ | j�||f� dS )z- 'Connect' to remote server. z\already connected, reconnecting is not currently supported (talk to itamar if you want this)z not an IPv4 or IPv6 address.N)r � RuntimeErrorr r r r5 r �connect)r+ �hostr r1 r1 r2 r| � s � z Port.connectc C s2 | � � | j�| � | jr| j�d| j� d S d S rg )� stopReadingr. �removeActiveHandlerL � callLater�connectionLostr6 r1 r1 r2 �_loseConnection� s �zPort._loseConnectionc C s& | j r t�� }| _nd }| �� |S rB )rL r �Deferred�dr� )r+ �resultr1 r1 r2 � stopListening s zPort.stopListeningc C s t jdtdd� | �� d S )Nz-Please use stopListening() to disconnect port� )� stacklevel)�warnings�warn�DeprecationWarningr� r6 r1 r1 r2 �loseConnection s � zPort.loseConnectionc C sb t �d| j � d| _tj�| |� | j�� | j� � | `| ` t | d�r/| j � d� | ` dS dS )z& Cleans up my socket. z(UDP Port %s Closed)Nr� )r rJ r r r r� r �doStopr �closerM �hasattrr� �callback)r+ �reasonr1 r1 r2 r� s �zPort.connectionLostc C s | � | j�}d| | _dS )zP Initialize the C{logstr} attribute to be used by C{logPrefix}. z%s (UDP)N)rK r �logstr)r+ � logPrefixr1 r1 r2 r $ s zPort.setLogStrc C r< )zK Returns the name of my class, to prefix log entries with. )r� r6 r1 r1 r2 r� + r> zPort.logPrefixc C sR | j �� }| jt jkrtjdg|�R � S | jt jkr'tjdg|dd� �R � S dS )z� Return the local address of the UDP connection @returns: the local address of the UDP connection @rtype: L{IPv4Address} or L{IPv6Address} �UDPNr� )r rI r! r4 r � IPv4Addressr3 � IPv6Address�r+ rt r1 r1 r2 �getHost1 s �z Port.getHostc C s | j �t jt j|� dS )z� Set whether this port may broadcast. This is disabled by default. @param enabled: Whether the port may broadcast. @type enabled: L{bool} N)r � setsockopt� SOL_SOCKET� SO_BROADCAST)r+ �enabledr1 r1 r2 �setBroadcastAllowed> s zPort.setBroadcastAllowedc C s t | j�tjtj��S )z� Checks if broadcast is currently allowed on this port. @return: Whether this port may broadcast. @rtype: L{bool} )�boolr � getsockoptr� r� r6 r1 r1 r2 �getBroadcastAllowedG s zPort.getBroadcastAllowed)r r NrB )&�__name__� __module__� __qualname__�__doc__r r4 r! � SOCK_DGRAMr" �dynamicReadBuffersr r �int�__annotations__r r �strr; r= rA rC r? r@ rY rS rT ro rz r| r� r� r� r� r r� r� r� r� r1 r1 r1 r2 r s: A r c @ sl e Zd ZdZdd� Zdd� Zdd� Zdd � Zd d � Zd d � Z dd� Z ddd�Z dd� Z dd� Z ddd�ZdS )�MulticastMixinz, Implement multicast functionality. c C s$ | j �t jt j�}t �t�d|��S )Nz@i)r r� � IPPROTO_IP�IP_MULTICAST_IF� inet_ntoar( �pack)r+ r r1 r1 r2 �getOutgoingInterfaceV s z#MulticastMixin.getOutgoingInterfacec C s | j �|��| j�S )z. Returns Deferred of success. )r. �resolve� addCallback� _setInterfacer� r1 r1 r2 �setOutgoingInterfaceZ s z#MulticastMixin.setOutgoingInterfacec C s" t �|�}| j �t jt j|� dS )NrE )r � inet_atonr� r� r� )r+ rt r r1 r1 r2 r� ` s z MulticastMixin._setInterfacec C � | j �t jt j�S rB )r r� r� �IP_MULTICAST_LOOPr6 r1 r1 r2 �getLoopbackModee rD z MulticastMixin.getLoopbackModec C s( t �dt|��}| j�tjtj|� d S )N�b)r( r� r� r r� r� r� )r+ �moder1 r1 r2 �setLoopbackModeh s z MulticastMixin.setLoopbackModec C r� rB )r r� r� �IP_MULTICAST_TTLr6 r1 r1 r2 �getTTLl rD zMulticastMixin.getTTLc C s$ t �d|�}| j�tjtj|� d S )N�B)r( r� r r� r� r� )r+ �ttlr1 r1 r2 �setTTLo s zMulticastMixin.setTTLr c C � | j �|��| j|d�S )zF Join a multicast group. Returns Deferred of success. rE �r. r� r� � _joinAddr1�r+ rt r r1 r1 r2 � joinGroups � zMulticastMixin.joinGroupc C s | j �|��| j||�S rB )r. r� r� � _joinAddr2)r+ rt r rx r1 r1 r2 r� y s zMulticastMixin._joinAddr1c C s� t �|�}t �|�}|rt j}nt j}z| j �t j||| � W d S tyA } zt�t j ||g|j �R � �W Y d }~S d }~ww rB ) r r� �IP_ADD_MEMBERSHIP�IP_DROP_MEMBERSHIPr� r� rG r �Failurer �MulticastJoinErrorrm )r+ r rt rx �cmd�er1 r1 r2 r� | s (��zMulticastMixin._joinAddr2c C r� )zD Leave multicast group, return Deferred of success. r r� r� r1 r1 r2 � leaveGroup� r� zMulticastMixin.leaveGroupN)r )r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� r1 r1 r1 r2 r� Q s r� c @ s* e Zd ZdZ d dd�Zdd � ZdS ) � MulticastPortz. UDP Port that supports multicasting. r r NFc C s t �| |||||� || _d S rB )r r �listenMultiple)r+ r r, r r- r. r� r1 r1 r2 r � s zMulticastPort.__init__c C sB t �| �}| jr |�tjtjd� ttd�r |�tjtjd� |S )NrE � SO_REUSEPORT) r rC r� r� r r� � SO_REUSEADDRr� r� )r+ r/ r1 r1 r2 rC � s zMulticastPort.createSocket)r r NF)r� r� r� r� r rC r1 r1 r1 r2 r� � s � r� )&r� rZ r r( r� �typingr �zope.interfacer �twisted.internetr r r r �twisted.internet.abstractr r � twisted.internet.iocpreactorr r r# �"twisted.internet.iocpreactor.constr r r �'twisted.internet.iocpreactor.interfacesr �twisted.pythonr r �IListeningPort� IUDPTransport� ISystemHandler r r� �IMulticastTransportr� r1 r1 r1 r2 �<module> s2 � 1> __pycache__/iocpsupport.cpython-310.pyc 0000644 00000000704 15030441410 0014035 0 ustar 00 o �b� � @ s@ g d �Z dd lmZmZmZmZmZmZmZm Z m Z m Z m Z dS )) �CompletionPort�Event�accept�connect�get_accept_addrs�have_connectex� makesockaddr� maxAddrLen�recv�recvfrom�send� N) �__all__� twisted_iocpsupport.iocpsupportr r r r r r r r r r r � r r �J/usr/lib/python3/dist-packages/twisted/internet/iocpreactor/iocpsupport.py�<module> s 8 __pycache__/const.cpython-310.pyc 0000644 00000001060 15030441410 0012570 0 ustar 00 o �b � @ s4 d Z dZdZdZdZdZdZdZdZd Z d Z d Z d S ) z Windows constants for IOCP i� i� i� i� i� i �@ �& ���ip i p N) �__doc__�ERROR_PORT_UNREACHABLE�ERROR_NETWORK_UNREACHABLE�ERROR_CONNECTION_REFUSED�ERROR_IO_PENDING�ERROR_OPERATION_ABORTED� WAIT_TIMEOUT�ERROR_NETNAME_DELETED�ERROR_HANDLE_EOF�INFINITE�SO_UPDATE_CONNECT_CONTEXT�SO_UPDATE_ACCEPT_CONTEXT� r r �D/usr/lib/python3/dist-packages/twisted/internet/iocpreactor/const.py�<module> s __pycache__/interfaces.cpython-310.pyc 0000644 00000003001 15030441410 0013562 0 ustar 00 o �b� � @ sF d Z ddlmZ G dd� de�ZG dd� de�ZG dd� dee�Zd S ) z Interfaces for iocpreactor � )� Interfacec @ � e Zd Zdd� ZdS )� IReadHandlec C � dS )a, Read into the given buffers from this handle. @param bufflist: the buffers to read into @type bufflist: list of objects implementing the read/write buffer protocol @param evt: an IOCP Event object @return: tuple (return code, number of bytes read) N� )�bufflist�evtr r �I/usr/lib/python3/dist-packages/twisted/internet/iocpreactor/interfaces.py�readFromHandle � zIReadHandle.readFromHandleN)�__name__� __module__� __qualname__r r r r r r � r c @ r )� IWriteHandlec C r )a Write the given buffer to this handle. @param buff: the buffer to write @type buff: any object implementing the buffer protocol @param evt: an IOCP Event object @return: tuple (return code, number of bytes written) Nr )�buffr r r r � writeToHandle r zIWriteHandle.writeToHandleN)r r r r r r r r r r r c @ s e Zd ZdS )�IReadWriteHandleN)r r r r r r r r ) s r N)�__doc__�zope.interfacer r r r r r r r �<module> s tcp.py 0000644 00000046616 15030441410 0005711 0 ustar 00 # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ TCP support for IOCP reactor """ import errno import socket import struct from typing import Optional from zope.interface import classImplements, implementer from twisted.internet import address, defer, error, interfaces, main from twisted.internet.abstract import _LogOwner, isIPv6Address from twisted.internet.iocpreactor import abstract, iocpsupport as _iocp from twisted.internet.iocpreactor.const import ( ERROR_CONNECTION_REFUSED, ERROR_IO_PENDING, ERROR_NETWORK_UNREACHABLE, SO_UPDATE_ACCEPT_CONTEXT, SO_UPDATE_CONNECT_CONTEXT, ) from twisted.internet.iocpreactor.interfaces import IReadWriteHandle from twisted.internet.protocol import Protocol from twisted.internet.tcp import ( Connector as TCPConnector, _AbortingMixin, _BaseBaseClient, _BaseTCPClient, _getsockname, _resolveIPv6, _SocketCloser, ) from twisted.python import failure, log, reflect try: from twisted.internet._newtls import startTLS as __startTLS except ImportError: _startTLS = None else: _startTLS = __startTLS # ConnectEx returns these. XXX: find out what it does for timeout connectExErrors = { ERROR_CONNECTION_REFUSED: errno.WSAECONNREFUSED, # type: ignore[attr-defined] ERROR_NETWORK_UNREACHABLE: errno.WSAENETUNREACH, # type: ignore[attr-defined] } @implementer(IReadWriteHandle, interfaces.ITCPTransport, interfaces.ISystemHandle) class Connection(abstract.FileHandle, _SocketCloser, _AbortingMixin): """ @ivar TLS: C{False} to indicate the connection is in normal TCP mode, C{True} to indicate that TLS has been started and that operations must be routed through the L{TLSMemoryBIOProtocol} instance. """ TLS = False def __init__(self, sock, proto, reactor=None): abstract.FileHandle.__init__(self, reactor) self.socket = sock self.getFileHandle = sock.fileno self.protocol = proto def getHandle(self): return self.socket def dataReceived(self, rbuffer): """ @param rbuffer: Data received. @type rbuffer: L{bytes} or L{bytearray} """ if isinstance(rbuffer, bytes): pass elif isinstance(rbuffer, bytearray): # XXX: some day, we'll have protocols that can handle raw buffers rbuffer = bytes(rbuffer) else: raise TypeError("data must be bytes or bytearray, not " + type(rbuffer)) self.protocol.dataReceived(rbuffer) def readFromHandle(self, bufflist, evt): return _iocp.recv(self.getFileHandle(), bufflist, evt) def writeToHandle(self, buff, evt): """ Send C{buff} to current file handle using C{_iocp.send}. The buffer sent is limited to a size of C{self.SEND_LIMIT}. """ writeView = memoryview(buff) return _iocp.send( self.getFileHandle(), writeView[0 : self.SEND_LIMIT].tobytes(), evt ) def _closeWriteConnection(self): try: self.socket.shutdown(1) except OSError: pass p = interfaces.IHalfCloseableProtocol(self.protocol, None) if p: try: p.writeConnectionLost() except BaseException: f = failure.Failure() log.err() self.connectionLost(f) def readConnectionLost(self, reason): p = interfaces.IHalfCloseableProtocol(self.protocol, None) if p: try: p.readConnectionLost() except BaseException: log.err() self.connectionLost(failure.Failure()) else: self.connectionLost(reason) def connectionLost(self, reason): if self.disconnected: return abstract.FileHandle.connectionLost(self, reason) isClean = reason is None or not reason.check(error.ConnectionAborted) self._closeSocket(isClean) protocol = self.protocol del self.protocol del self.socket del self.getFileHandle protocol.connectionLost(reason) def logPrefix(self): """ Return the prefix to log with when I own the logging thread. """ return self.logstr def getTcpNoDelay(self): return bool(self.socket.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY)) def setTcpNoDelay(self, enabled): self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, enabled) def getTcpKeepAlive(self): return bool(self.socket.getsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE)) def setTcpKeepAlive(self, enabled): self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, enabled) if _startTLS is not None: def startTLS(self, contextFactory, normal=True): """ @see: L{ITLSTransport.startTLS} """ _startTLS(self, contextFactory, normal, abstract.FileHandle) def write(self, data): """ Write some data, either directly to the underlying handle or, if TLS has been started, to the L{TLSMemoryBIOProtocol} for it to encrypt and send. @see: L{twisted.internet.interfaces.ITransport.write} """ if self.disconnected: return if self.TLS: self.protocol.write(data) else: abstract.FileHandle.write(self, data) def writeSequence(self, iovec): """ Write some data, either directly to the underlying handle or, if TLS has been started, to the L{TLSMemoryBIOProtocol} for it to encrypt and send. @see: L{twisted.internet.interfaces.ITransport.writeSequence} """ if self.disconnected: return if self.TLS: self.protocol.writeSequence(iovec) else: abstract.FileHandle.writeSequence(self, iovec) def loseConnection(self, reason=None): """ Close the underlying handle or, if TLS has been started, first shut it down. @see: L{twisted.internet.interfaces.ITransport.loseConnection} """ if self.TLS: if self.connected and not self.disconnecting: self.protocol.loseConnection() else: abstract.FileHandle.loseConnection(self, reason) def registerProducer(self, producer, streaming): """ Register a producer. If TLS is enabled, the TLS connection handles this. """ if self.TLS: # Registering a producer before we're connected shouldn't be a # problem. If we end up with a write(), that's already handled in # the write() code above, and there are no other potential # side-effects. self.protocol.registerProducer(producer, streaming) else: abstract.FileHandle.registerProducer(self, producer, streaming) def unregisterProducer(self): """ Unregister a producer. If TLS is enabled, the TLS connection handles this. """ if self.TLS: self.protocol.unregisterProducer() else: abstract.FileHandle.unregisterProducer(self) def getHost(self): # ITCPTransport.getHost pass def getPeer(self): # ITCPTransport.getPeer pass if _startTLS is not None: classImplements(Connection, interfaces.ITLSTransport) class Client(_BaseBaseClient, _BaseTCPClient, Connection): """ @ivar _tlsClientDefault: Always C{True}, indicating that this is a client connection, and by default when TLS is negotiated this class will act as a TLS client. """ addressFamily = socket.AF_INET socketType = socket.SOCK_STREAM _tlsClientDefault = True _commonConnection = Connection def __init__(self, host, port, bindAddress, connector, reactor): # ConnectEx documentation says socket _has_ to be bound if bindAddress is None: bindAddress = ("", 0) self.reactor = reactor # createInternetSocket needs this _BaseTCPClient.__init__(self, host, port, bindAddress, connector, reactor) def createInternetSocket(self): """ Create a socket registered with the IOCP reactor. @see: L{_BaseTCPClient} """ return self.reactor.createSocket(self.addressFamily, self.socketType) def _collectSocketDetails(self): """ Clean up potentially circular references to the socket and to its C{getFileHandle} method. @see: L{_BaseBaseClient} """ del self.socket, self.getFileHandle def _stopReadingAndWriting(self): """ Remove the active handle from the reactor. @see: L{_BaseBaseClient} """ self.reactor.removeActiveHandle(self) def cbConnect(self, rc, data, evt): if rc: rc = connectExErrors.get(rc, rc) self.failIfNotConnected( error.getConnectError((rc, errno.errorcode.get(rc, "Unknown error"))) ) else: self.socket.setsockopt( socket.SOL_SOCKET, SO_UPDATE_CONNECT_CONTEXT, struct.pack("P", self.socket.fileno()), ) self.protocol = self.connector.buildProtocol(self.getPeer()) self.connected = True logPrefix = self._getLogPrefix(self.protocol) self.logstr = logPrefix + ",client" if self.protocol is None: # Factory.buildProtocol is allowed to return None. In that # case, make up a protocol to satisfy the rest of the # implementation; connectionLost is going to be called on # something, for example. This is easier than adding special # case support for a None protocol throughout the rest of the # transport implementation. self.protocol = Protocol() # But dispose of the connection quickly. self.loseConnection() else: self.protocol.makeConnection(self) self.startReading() def doConnect(self): if not hasattr(self, "connector"): # this happens if we connector.stopConnecting in # factory.startedConnecting return assert _iocp.have_connectex self.reactor.addActiveHandle(self) evt = _iocp.Event(self.cbConnect, self) rc = _iocp.connect(self.socket.fileno(), self.realAddress, evt) if rc and rc != ERROR_IO_PENDING: self.cbConnect(rc, 0, evt) class Server(Connection): """ Serverside socket-stream connection class. I am a serverside network connection transport; a socket which came from an accept() on a server. @ivar _tlsClientDefault: Always C{False}, indicating that this is a server connection, and by default when TLS is negotiated this class will act as a TLS server. """ _tlsClientDefault = False def __init__(self, sock, protocol, clientAddr, serverAddr, sessionno, reactor): """ Server(sock, protocol, client, server, sessionno) Initialize me with a socket, a protocol, a descriptor for my peer (a tuple of host, port describing the other end of the connection), an instance of Port, and a session number. """ Connection.__init__(self, sock, protocol, reactor) self.serverAddr = serverAddr self.clientAddr = clientAddr self.sessionno = sessionno logPrefix = self._getLogPrefix(self.protocol) self.logstr = f"{logPrefix},{sessionno},{self.clientAddr.host}" self.repstr = "<{} #{} on {}>".format( self.protocol.__class__.__name__, self.sessionno, self.serverAddr.port, ) self.connected = True self.startReading() def __repr__(self) -> str: """ A string representation of this connection. """ return self.repstr def getHost(self): """ Returns an IPv4Address. This indicates the server's address. """ return self.serverAddr def getPeer(self): """ Returns an IPv4Address. This indicates the client's address. """ return self.clientAddr class Connector(TCPConnector): def _makeTransport(self): return Client(self.host, self.port, self.bindAddress, self, self.reactor) @implementer(interfaces.IListeningPort) class Port(_SocketCloser, _LogOwner): connected = False disconnected = False disconnecting = False addressFamily = socket.AF_INET socketType = socket.SOCK_STREAM _addressType = address.IPv4Address sessionno = 0 # Actual port number being listened on, only set to a non-None # value when we are actually listening. _realPortNumber: Optional[int] = None # A string describing the connections which will be created by this port. # Normally this is C{"TCP"}, since this is a TCP port, but when the TLS # implementation re-uses this class it overrides the value with C{"TLS"}. # Only used for logging. _type = "TCP" def __init__(self, port, factory, backlog=50, interface="", reactor=None): self.port = port self.factory = factory self.backlog = backlog self.interface = interface self.reactor = reactor if isIPv6Address(interface): self.addressFamily = socket.AF_INET6 self._addressType = address.IPv6Address def __repr__(self) -> str: if self._realPortNumber is not None: return "<{} of {} on {}>".format( self.__class__, self.factory.__class__, self._realPortNumber, ) else: return "<{} of {} (not listening)>".format( self.__class__, self.factory.__class__, ) def startListening(self): try: skt = self.reactor.createSocket(self.addressFamily, self.socketType) # TODO: resolve self.interface if necessary if self.addressFamily == socket.AF_INET6: addr = _resolveIPv6(self.interface, self.port) else: addr = (self.interface, self.port) skt.bind(addr) except OSError as le: raise error.CannotListenError(self.interface, self.port, le) self.addrLen = _iocp.maxAddrLen(skt.fileno()) # Make sure that if we listened on port 0, we update that to # reflect what the OS actually assigned us. self._realPortNumber = skt.getsockname()[1] log.msg( "%s starting on %s" % (self._getLogPrefix(self.factory), self._realPortNumber) ) self.factory.doStart() skt.listen(self.backlog) self.connected = True self.disconnected = False self.reactor.addActiveHandle(self) self.socket = skt self.getFileHandle = self.socket.fileno self.doAccept() def loseConnection(self, connDone=failure.Failure(main.CONNECTION_DONE)): """ Stop accepting connections on this port. This will shut down my socket and call self.connectionLost(). It returns a deferred which will fire successfully when the port is actually closed. """ self.disconnecting = True if self.connected: self.deferred = defer.Deferred() self.reactor.callLater(0, self.connectionLost, connDone) return self.deferred stopListening = loseConnection def _logConnectionLostMsg(self): """ Log message for closing port """ log.msg(f"({self._type} Port {self._realPortNumber} Closed)") def connectionLost(self, reason): """ Cleans up the socket. """ self._logConnectionLostMsg() self._realPortNumber = None d = None if hasattr(self, "deferred"): d = self.deferred del self.deferred self.disconnected = True self.reactor.removeActiveHandle(self) self.connected = False self._closeSocket(True) del self.socket del self.getFileHandle try: self.factory.doStop() except BaseException: self.disconnecting = False if d is not None: d.errback(failure.Failure()) else: raise else: self.disconnecting = False if d is not None: d.callback(None) def logPrefix(self): """ Returns the name of my class, to prefix log entries with. """ return reflect.qual(self.factory.__class__) def getHost(self): """ Returns an IPv4Address or IPv6Address. This indicates the server's address. """ return self._addressType("TCP", *_getsockname(self.socket)) def cbAccept(self, rc, data, evt): self.handleAccept(rc, evt) if not (self.disconnecting or self.disconnected): self.doAccept() def handleAccept(self, rc, evt): if self.disconnecting or self.disconnected: return False # possible errors: # (WSAEMFILE, WSAENOBUFS, WSAENFILE, WSAENOMEM, WSAECONNABORTED) if rc: log.msg( "Could not accept new connection -- %s (%s)" % (errno.errorcode.get(rc, "unknown error"), rc) ) return False else: # Inherit the properties from the listening port socket as # documented in the `Remarks` section of AcceptEx. # https://docs.microsoft.com/en-us/windows/win32/api/mswsock/nf-mswsock-acceptex # In this way we can call getsockname and getpeername on the # accepted socket. evt.newskt.setsockopt( socket.SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT, struct.pack("P", self.socket.fileno()), ) family, lAddr, rAddr = _iocp.get_accept_addrs(evt.newskt.fileno(), evt.buff) assert family == self.addressFamily # Build an IPv6 address that includes the scopeID, if necessary if "%" in lAddr[0]: scope = int(lAddr[0].split("%")[1]) lAddr = (lAddr[0], lAddr[1], 0, scope) if "%" in rAddr[0]: scope = int(rAddr[0].split("%")[1]) rAddr = (rAddr[0], rAddr[1], 0, scope) protocol = self.factory.buildProtocol(self._addressType("TCP", *rAddr)) if protocol is None: evt.newskt.close() else: s = self.sessionno self.sessionno = s + 1 transport = Server( evt.newskt, protocol, self._addressType("TCP", *rAddr), self._addressType("TCP", *lAddr), s, self.reactor, ) protocol.makeConnection(transport) return True def doAccept(self): evt = _iocp.Event(self.cbAccept, self) # see AcceptEx documentation evt.buff = buff = bytearray(2 * (self.addrLen + 16)) evt.newskt = newskt = self.reactor.createSocket( self.addressFamily, self.socketType ) rc = _iocp.accept(self.socket.fileno(), newskt.fileno(), buff, evt) if rc and rc != ERROR_IO_PENDING: self.handleAccept(rc, evt) interfaces.py 0000644 00000001661 15030441410 0007235 0 ustar 00 # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Interfaces for iocpreactor """ from zope.interface import Interface class IReadHandle(Interface): def readFromHandle(bufflist, evt): """ Read into the given buffers from this handle. @param bufflist: the buffers to read into @type bufflist: list of objects implementing the read/write buffer protocol @param evt: an IOCP Event object @return: tuple (return code, number of bytes read) """ class IWriteHandle(Interface): def writeToHandle(buff, evt): """ Write the given buffer to this handle. @param buff: the buffer to write @type buff: any object implementing the buffer protocol @param evt: an IOCP Event object @return: tuple (return code, number of bytes written) """ class IReadWriteHandle(IReadHandle, IWriteHandle): pass abstract.py 0000644 00000031420 15030441410 0006711 0 ustar 00 # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Abstract file handle class """ import errno from zope.interface import implementer from twisted.internet import error, interfaces, main from twisted.internet.abstract import _ConsumerMixin, _dataMustBeBytes, _LogOwner from twisted.internet.iocpreactor import iocpsupport as _iocp from twisted.internet.iocpreactor.const import ERROR_HANDLE_EOF, ERROR_IO_PENDING from twisted.python import failure @implementer( interfaces.IPushProducer, interfaces.IConsumer, interfaces.ITransport, interfaces.IHalfCloseableDescriptor, ) class FileHandle(_ConsumerMixin, _LogOwner): """ File handle that can read and write asynchronously """ # read stuff maxReadBuffers = 16 readBufferSize = 4096 reading = False dynamicReadBuffers = True # set this to false if subclass doesn't do iovecs _readNextBuffer = 0 _readSize = 0 # how much data we have in the read buffer _readScheduled = None _readScheduledInOS = False def startReading(self): self.reactor.addActiveHandle(self) if not self._readScheduled and not self.reading: self.reading = True self._readScheduled = self.reactor.callLater(0, self._resumeReading) def stopReading(self): if self._readScheduled: self._readScheduled.cancel() self._readScheduled = None self.reading = False def _resumeReading(self): self._readScheduled = None if self._dispatchData() and not self._readScheduledInOS: self.doRead() def _dispatchData(self): """ Dispatch previously read data. Return True if self.reading and we don't have any more data """ if not self._readSize: return self.reading size = self._readSize full_buffers = size // self.readBufferSize while self._readNextBuffer < full_buffers: self.dataReceived(self._readBuffers[self._readNextBuffer]) self._readNextBuffer += 1 if not self.reading: return False remainder = size % self.readBufferSize if remainder: self.dataReceived(self._readBuffers[full_buffers][0:remainder]) if self.dynamicReadBuffers: total_buffer_size = self.readBufferSize * len(self._readBuffers) # we have one buffer too many if size < total_buffer_size - self.readBufferSize: del self._readBuffers[-1] # we filled all buffers, so allocate one more elif ( size == total_buffer_size and len(self._readBuffers) < self.maxReadBuffers ): self._readBuffers.append(bytearray(self.readBufferSize)) self._readNextBuffer = 0 self._readSize = 0 return self.reading def _cbRead(self, rc, data, evt): self._readScheduledInOS = False if self._handleRead(rc, data, evt): self.doRead() def _handleRead(self, rc, data, evt): """ Returns False if we should stop reading for now """ if self.disconnected: return False # graceful disconnection if (not (rc or data)) or rc in (errno.WSAEDISCON, ERROR_HANDLE_EOF): self.reactor.removeActiveHandle(self) self.readConnectionLost(failure.Failure(main.CONNECTION_DONE)) return False # XXX: not handling WSAEWOULDBLOCK # ("too many outstanding overlapped I/O requests") elif rc: self.connectionLost( failure.Failure( error.ConnectionLost( "read error -- %s (%s)" % (errno.errorcode.get(rc, "unknown"), rc) ) ) ) return False else: assert self._readSize == 0 assert self._readNextBuffer == 0 self._readSize = data return self._dispatchData() def doRead(self): evt = _iocp.Event(self._cbRead, self) evt.buff = buff = self._readBuffers rc, numBytesRead = self.readFromHandle(buff, evt) if not rc or rc == ERROR_IO_PENDING: self._readScheduledInOS = True else: self._handleRead(rc, numBytesRead, evt) def readFromHandle(self, bufflist, evt): raise NotImplementedError() # TODO: this should default to ReadFile def dataReceived(self, data): raise NotImplementedError def readConnectionLost(self, reason): self.connectionLost(reason) # write stuff dataBuffer = b"" offset = 0 writing = False _writeScheduled = None _writeDisconnecting = False _writeDisconnected = False writeBufferSize = 2 ** 2 ** 2 ** 2 def loseWriteConnection(self): self._writeDisconnecting = True self.startWriting() def _closeWriteConnection(self): # override in subclasses pass def writeConnectionLost(self, reason): # in current code should never be called self.connectionLost(reason) def startWriting(self): self.reactor.addActiveHandle(self) if not self._writeScheduled and not self.writing: self.writing = True self._writeScheduled = self.reactor.callLater(0, self._resumeWriting) def stopWriting(self): if self._writeScheduled: self._writeScheduled.cancel() self._writeScheduled = None self.writing = False def _resumeWriting(self): self._writeScheduled = None self.doWrite() def _cbWrite(self, rc, numBytesWritten, evt): if self._handleWrite(rc, numBytesWritten, evt): self.doWrite() def _handleWrite(self, rc, numBytesWritten, evt): """ Returns false if we should stop writing for now """ if self.disconnected or self._writeDisconnected: return False # XXX: not handling WSAEWOULDBLOCK # ("too many outstanding overlapped I/O requests") if rc: self.connectionLost( failure.Failure( error.ConnectionLost( "write error -- %s (%s)" % (errno.errorcode.get(rc, "unknown"), rc) ) ) ) return False else: self.offset += numBytesWritten # If there is nothing left to send, if self.offset == len(self.dataBuffer) and not self._tempDataLen: self.dataBuffer = b"" self.offset = 0 # stop writing self.stopWriting() # If I've got a producer who is supposed to supply me with data if self.producer is not None and ( (not self.streamingProducer) or self.producerPaused ): # tell them to supply some more. self.producerPaused = True self.producer.resumeProducing() elif self.disconnecting: # But if I was previously asked to let the connection die, # do so. self.connectionLost(failure.Failure(main.CONNECTION_DONE)) elif self._writeDisconnecting: # I was previously asked to half-close the connection. self._writeDisconnected = True self._closeWriteConnection() return False else: return True def doWrite(self): if len(self.dataBuffer) - self.offset < self.SEND_LIMIT: # If there is currently less than SEND_LIMIT bytes left to send # in the string, extend it with the array data. self.dataBuffer = self.dataBuffer[self.offset :] + b"".join( self._tempDataBuffer ) self.offset = 0 self._tempDataBuffer = [] self._tempDataLen = 0 evt = _iocp.Event(self._cbWrite, self) # Send as much data as you can. if self.offset: sendView = memoryview(self.dataBuffer) evt.buff = buff = sendView[self.offset :] else: evt.buff = buff = self.dataBuffer rc, data = self.writeToHandle(buff, evt) if rc and rc != ERROR_IO_PENDING: self._handleWrite(rc, data, evt) def writeToHandle(self, buff, evt): raise NotImplementedError() # TODO: this should default to WriteFile def write(self, data): """Reliably write some data. The data is buffered until his file descriptor is ready for writing. """ _dataMustBeBytes(data) if not self.connected or self._writeDisconnected: return if data: self._tempDataBuffer.append(data) self._tempDataLen += len(data) if self.producer is not None and self.streamingProducer: if len(self.dataBuffer) + self._tempDataLen > self.writeBufferSize: self.producerPaused = True self.producer.pauseProducing() self.startWriting() def writeSequence(self, iovec): for i in iovec: _dataMustBeBytes(i) if not self.connected or not iovec or self._writeDisconnected: return self._tempDataBuffer.extend(iovec) for i in iovec: self._tempDataLen += len(i) if self.producer is not None and self.streamingProducer: if len(self.dataBuffer) + self._tempDataLen > self.writeBufferSize: self.producerPaused = True self.producer.pauseProducing() self.startWriting() # general stuff connected = False disconnected = False disconnecting = False logstr = "Uninitialized" SEND_LIMIT = 128 * 1024 def __init__(self, reactor=None): if not reactor: from twisted.internet import reactor self.reactor = reactor self._tempDataBuffer = [] # will be added to dataBuffer in doWrite self._tempDataLen = 0 self._readBuffers = [bytearray(self.readBufferSize)] def connectionLost(self, reason): """ The connection was lost. This is called when the connection on a selectable object has been lost. It will be called whether the connection was closed explicitly, an exception occurred in an event handler, or the other end of the connection closed it first. Clean up state here, but make sure to call back up to FileDescriptor. """ self.disconnected = True self.connected = False if self.producer is not None: self.producer.stopProducing() self.producer = None self.stopReading() self.stopWriting() self.reactor.removeActiveHandle(self) def getFileHandle(self): return -1 def loseConnection(self, _connDone=failure.Failure(main.CONNECTION_DONE)): """ Close the connection at the next available opportunity. Call this to cause this FileDescriptor to lose its connection. It will first write any data that it has buffered. If there is data buffered yet to be written, this method will cause the transport to lose its connection as soon as it's done flushing its write buffer. If you have a producer registered, the connection won't be closed until the producer is finished. Therefore, make sure you unregister your producer when it's finished, or the connection will never close. """ if self.connected and not self.disconnecting: if self._writeDisconnected: # doWrite won't trigger the connection close anymore self.stopReading() self.stopWriting self.connectionLost(_connDone) else: self.stopReading() self.startWriting() self.disconnecting = 1 # Producer/consumer implementation def stopConsuming(self): """ Stop consuming data. This is called when a producer has lost its connection, to tell the consumer to go lose its connection (and break potential circular references). """ self.unregisterProducer() self.loseConnection() # producer interface implementation def resumeProducing(self): if self.connected and not self.disconnecting: self.startReading() def pauseProducing(self): self.stopReading() def stopProducing(self): self.loseConnection() def getHost(self): # ITransport.getHost raise NotImplementedError() def getPeer(self): # ITransport.getPeer raise NotImplementedError() __all__ = ["FileHandle"] udp.py 0000644 00000033134 15030441410 0005702 0 ustar 00 # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ UDP support for IOCP reactor """ import errno import socket import struct import warnings from typing import Optional from zope.interface import implementer from twisted.internet import address, defer, error, interfaces from twisted.internet.abstract import isIPAddress, isIPv6Address from twisted.internet.iocpreactor import abstract, iocpsupport as _iocp from twisted.internet.iocpreactor.const import ( ERROR_CONNECTION_REFUSED, ERROR_IO_PENDING, ERROR_PORT_UNREACHABLE, ) from twisted.internet.iocpreactor.interfaces import IReadWriteHandle from twisted.python import failure, log @implementer( IReadWriteHandle, interfaces.IListeningPort, interfaces.IUDPTransport, interfaces.ISystemHandle, ) class Port(abstract.FileHandle): """ UDP port, listening for packets. @ivar addressFamily: L{socket.AF_INET} or L{socket.AF_INET6}, depending on whether this port is listening on an IPv4 address or an IPv6 address. """ addressFamily = socket.AF_INET socketType = socket.SOCK_DGRAM dynamicReadBuffers = False # Actual port number being listened on, only set to a non-None # value when we are actually listening. _realPortNumber: Optional[int] = None def __init__(self, port, proto, interface="", maxPacketSize=8192, reactor=None): """ Initialize with a numeric port to listen on. """ self.port = port self.protocol = proto self.readBufferSize = maxPacketSize self.interface = interface self.setLogStr() self._connectedAddr = None self._setAddressFamily() abstract.FileHandle.__init__(self, reactor) skt = socket.socket(self.addressFamily, self.socketType) addrLen = _iocp.maxAddrLen(skt.fileno()) self.addressBuffer = bytearray(addrLen) # WSARecvFrom takes an int self.addressLengthBuffer = bytearray(struct.calcsize("i")) def _setAddressFamily(self): """ Resolve address family for the socket. """ if isIPv6Address(self.interface): self.addressFamily = socket.AF_INET6 elif isIPAddress(self.interface): self.addressFamily = socket.AF_INET elif self.interface: raise error.InvalidAddressError( self.interface, "not an IPv4 or IPv6 address" ) def __repr__(self) -> str: if self._realPortNumber is not None: return f"<{self.protocol.__class__} on {self._realPortNumber}>" else: return f"<{self.protocol.__class__} not connected>" def getHandle(self): """ Return a socket object. """ return self.socket def startListening(self): """ Create and bind my socket, and begin listening on it. This is called on unserialization, and must be called after creating a server to begin listening on the specified port. """ self._bindSocket() self._connectToProtocol() def createSocket(self): return self.reactor.createSocket(self.addressFamily, self.socketType) def _bindSocket(self): try: skt = self.createSocket() skt.bind((self.interface, self.port)) except OSError as le: raise error.CannotListenError(self.interface, self.port, le) # Make sure that if we listened on port 0, we update that to # reflect what the OS actually assigned us. self._realPortNumber = skt.getsockname()[1] log.msg( "%s starting on %s" % (self._getLogPrefix(self.protocol), self._realPortNumber) ) self.connected = True self.socket = skt self.getFileHandle = self.socket.fileno def _connectToProtocol(self): self.protocol.makeConnection(self) self.startReading() self.reactor.addActiveHandle(self) def cbRead(self, rc, data, evt): if self.reading: self.handleRead(rc, data, evt) self.doRead() def handleRead(self, rc, data, evt): if rc in ( errno.WSAECONNREFUSED, errno.WSAECONNRESET, ERROR_CONNECTION_REFUSED, ERROR_PORT_UNREACHABLE, ): if self._connectedAddr: self.protocol.connectionRefused() elif rc: log.msg( "error in recvfrom -- %s (%s)" % (errno.errorcode.get(rc, "unknown error"), rc) ) else: try: self.protocol.datagramReceived( bytes(evt.buff[:data]), _iocp.makesockaddr(evt.addr_buff) ) except BaseException: log.err() def doRead(self): evt = _iocp.Event(self.cbRead, self) evt.buff = buff = self._readBuffers[0] evt.addr_buff = addr_buff = self.addressBuffer evt.addr_len_buff = addr_len_buff = self.addressLengthBuffer rc, data = _iocp.recvfrom( self.getFileHandle(), buff, addr_buff, addr_len_buff, evt ) if rc and rc != ERROR_IO_PENDING: self.handleRead(rc, data, evt) def write(self, datagram, addr=None): """ Write a datagram. @param addr: should be a tuple (ip, port), can be None in connected mode. """ if self._connectedAddr: assert addr in (None, self._connectedAddr) try: return self.socket.send(datagram) except OSError as se: no = se.args[0] if no == errno.WSAEINTR: return self.write(datagram) elif no == errno.WSAEMSGSIZE: raise error.MessageLengthError("message too long") elif no in ( errno.WSAECONNREFUSED, errno.WSAECONNRESET, ERROR_CONNECTION_REFUSED, ERROR_PORT_UNREACHABLE, ): self.protocol.connectionRefused() else: raise else: assert addr != None if ( not isIPAddress(addr[0]) and not isIPv6Address(addr[0]) and addr[0] != "<broadcast>" ): raise error.InvalidAddressError( addr[0], "write() only accepts IP addresses, not hostnames" ) if isIPAddress(addr[0]) and self.addressFamily == socket.AF_INET6: raise error.InvalidAddressError( addr[0], "IPv6 port write() called with IPv4 address" ) if isIPv6Address(addr[0]) and self.addressFamily == socket.AF_INET: raise error.InvalidAddressError( addr[0], "IPv4 port write() called with IPv6 address" ) try: return self.socket.sendto(datagram, addr) except OSError as se: no = se.args[0] if no == errno.WSAEINTR: return self.write(datagram, addr) elif no == errno.WSAEMSGSIZE: raise error.MessageLengthError("message too long") elif no in ( errno.WSAECONNREFUSED, errno.WSAECONNRESET, ERROR_CONNECTION_REFUSED, ERROR_PORT_UNREACHABLE, ): # in non-connected UDP ECONNREFUSED is platform dependent, # I think and the info is not necessarily useful. # Nevertheless maybe we should call connectionRefused? XXX return else: raise def writeSequence(self, seq, addr): self.write(b"".join(seq), addr) def connect(self, host, port): """ 'Connect' to remote server. """ if self._connectedAddr: raise RuntimeError( "already connected, reconnecting is not currently supported " "(talk to itamar if you want this)" ) if not isIPAddress(host) and not isIPv6Address(host): raise error.InvalidAddressError(host, "not an IPv4 or IPv6 address.") self._connectedAddr = (host, port) self.socket.connect((host, port)) def _loseConnection(self): self.stopReading() self.reactor.removeActiveHandle(self) if self.connected: # actually means if we are *listening* self.reactor.callLater(0, self.connectionLost) def stopListening(self): if self.connected: result = self.d = defer.Deferred() else: result = None self._loseConnection() return result def loseConnection(self): warnings.warn( "Please use stopListening() to disconnect port", DeprecationWarning, stacklevel=2, ) self.stopListening() def connectionLost(self, reason=None): """ Cleans up my socket. """ log.msg("(UDP Port %s Closed)" % self._realPortNumber) self._realPortNumber = None abstract.FileHandle.connectionLost(self, reason) self.protocol.doStop() self.socket.close() del self.socket del self.getFileHandle if hasattr(self, "d"): self.d.callback(None) del self.d def setLogStr(self): """ Initialize the C{logstr} attribute to be used by C{logPrefix}. """ logPrefix = self._getLogPrefix(self.protocol) self.logstr = "%s (UDP)" % logPrefix def logPrefix(self): """ Returns the name of my class, to prefix log entries with. """ return self.logstr def getHost(self): """ Return the local address of the UDP connection @returns: the local address of the UDP connection @rtype: L{IPv4Address} or L{IPv6Address} """ addr = self.socket.getsockname() if self.addressFamily == socket.AF_INET: return address.IPv4Address("UDP", *addr) elif self.addressFamily == socket.AF_INET6: return address.IPv6Address("UDP", *(addr[:2])) def setBroadcastAllowed(self, enabled): """ Set whether this port may broadcast. This is disabled by default. @param enabled: Whether the port may broadcast. @type enabled: L{bool} """ self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, enabled) def getBroadcastAllowed(self): """ Checks if broadcast is currently allowed on this port. @return: Whether this port may broadcast. @rtype: L{bool} """ return bool(self.socket.getsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST)) class MulticastMixin: """ Implement multicast functionality. """ def getOutgoingInterface(self): i = self.socket.getsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_IF) return socket.inet_ntoa(struct.pack("@i", i)) def setOutgoingInterface(self, addr): """ Returns Deferred of success. """ return self.reactor.resolve(addr).addCallback(self._setInterface) def _setInterface(self, addr): i = socket.inet_aton(addr) self.socket.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_IF, i) return 1 def getLoopbackMode(self): return self.socket.getsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_LOOP) def setLoopbackMode(self, mode): mode = struct.pack("b", bool(mode)) self.socket.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_LOOP, mode) def getTTL(self): return self.socket.getsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL) def setTTL(self, ttl): ttl = struct.pack("B", ttl) self.socket.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, ttl) def joinGroup(self, addr, interface=""): """ Join a multicast group. Returns Deferred of success. """ return self.reactor.resolve(addr).addCallback(self._joinAddr1, interface, 1) def _joinAddr1(self, addr, interface, join): return self.reactor.resolve(interface).addCallback(self._joinAddr2, addr, join) def _joinAddr2(self, interface, addr, join): addr = socket.inet_aton(addr) interface = socket.inet_aton(interface) if join: cmd = socket.IP_ADD_MEMBERSHIP else: cmd = socket.IP_DROP_MEMBERSHIP try: self.socket.setsockopt(socket.IPPROTO_IP, cmd, addr + interface) except OSError as e: return failure.Failure(error.MulticastJoinError(addr, interface, *e.args)) def leaveGroup(self, addr, interface=""): """ Leave multicast group, return Deferred of success. """ return self.reactor.resolve(addr).addCallback(self._joinAddr1, interface, 0) @implementer(interfaces.IMulticastTransport) class MulticastPort(MulticastMixin, Port): """ UDP Port that supports multicasting. """ def __init__( self, port, proto, interface="", maxPacketSize=8192, reactor=None, listenMultiple=False, ): Port.__init__(self, port, proto, interface, maxPacketSize, reactor) self.listenMultiple = listenMultiple def createSocket(self): skt = Port.createSocket(self) if self.listenMultiple: skt.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) if hasattr(socket, "SO_REUSEPORT"): skt.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) return skt reactor.py 0000644 00000022331 15030441410 0006546 0 ustar 00 # -*- test-case-name: twisted.internet.test.test_iocp -*- # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Reactor that uses IO completion ports """ import socket import sys import warnings from typing import Tuple, Type from zope.interface import implementer from twisted.internet import base, error, interfaces, main from twisted.internet._dumbwin32proc import Process from twisted.internet.iocpreactor import iocpsupport as _iocp, tcp, udp from twisted.internet.iocpreactor.const import WAIT_TIMEOUT from twisted.internet.win32eventreactor import _ThreadedWin32EventsMixin from twisted.python import failure, log try: from twisted.protocols.tls import TLSMemoryBIOFactory as _TLSMemoryBIOFactory except ImportError: TLSMemoryBIOFactory = None # Either pyOpenSSL isn't installed, or it is too old for this code to work. # The reactor won't provide IReactorSSL. _extraInterfaces: Tuple[Type[interfaces.IReactorSSL], ...] = () warnings.warn( "pyOpenSSL 0.10 or newer is required for SSL support in iocpreactor. " "It is missing, so the reactor will not support SSL APIs." ) else: TLSMemoryBIOFactory = _TLSMemoryBIOFactory _extraInterfaces = (interfaces.IReactorSSL,) MAX_TIMEOUT = 2000 # 2 seconds, see doIteration for explanation EVENTS_PER_LOOP = 1000 # XXX: what's a good value here? # keys to associate with normal and waker events KEY_NORMAL, KEY_WAKEUP = range(2) _NO_GETHANDLE = error.ConnectionFdescWentAway("Handler has no getFileHandle method") _NO_FILEDESC = error.ConnectionFdescWentAway("Filedescriptor went away") @implementer( interfaces.IReactorTCP, interfaces.IReactorUDP, interfaces.IReactorMulticast, interfaces.IReactorProcess, *_extraInterfaces, ) class IOCPReactor( base._SignalReactorMixin, base.ReactorBase, _ThreadedWin32EventsMixin ): port = None def __init__(self): base.ReactorBase.__init__(self) self.port = _iocp.CompletionPort() self.handles = set() def addActiveHandle(self, handle): self.handles.add(handle) def removeActiveHandle(self, handle): self.handles.discard(handle) def doIteration(self, timeout): """ Poll the IO completion port for new events. """ # This function sits and waits for an IO completion event. # # There are two requirements: process IO events as soon as they arrive # and process ctrl-break from the user in a reasonable amount of time. # # There are three kinds of waiting. # 1) GetQueuedCompletionStatus (self.port.getEvent) to wait for IO # events only. # 2) Msg* family of wait functions that can stop waiting when # ctrl-break is detected (then, I think, Python converts it into a # KeyboardInterrupt) # 3) *Ex family of wait functions that put the thread into an # "alertable" wait state which is supposedly triggered by IO completion # # 2) and 3) can be combined. Trouble is, my IO completion is not # causing 3) to trigger, possibly because I do not use an IO completion # callback. Windows is weird. # There are two ways to handle this. I could use MsgWaitForSingleObject # here and GetQueuedCompletionStatus in a thread. Or I could poll with # a reasonable interval. Guess what! Threads are hard. processed_events = 0 if timeout is None: timeout = MAX_TIMEOUT else: timeout = min(MAX_TIMEOUT, int(1000 * timeout)) rc, numBytes, key, evt = self.port.getEvent(timeout) while 1: if rc == WAIT_TIMEOUT: break if key != KEY_WAKEUP: assert key == KEY_NORMAL log.callWithLogger( evt.owner, self._callEventCallback, rc, numBytes, evt ) processed_events += 1 if processed_events >= EVENTS_PER_LOOP: break rc, numBytes, key, evt = self.port.getEvent(0) def _callEventCallback(self, rc, numBytes, evt): owner = evt.owner why = None try: evt.callback(rc, numBytes, evt) handfn = getattr(owner, "getFileHandle", None) if not handfn: why = _NO_GETHANDLE elif handfn() == -1: why = _NO_FILEDESC if why: return # ignore handles that were closed except BaseException: why = sys.exc_info()[1] log.err() if why: owner.loseConnection(failure.Failure(why)) def installWaker(self): pass def wakeUp(self): self.port.postEvent(0, KEY_WAKEUP, None) def registerHandle(self, handle): self.port.addHandle(handle, KEY_NORMAL) def createSocket(self, af, stype): skt = socket.socket(af, stype) self.registerHandle(skt.fileno()) return skt def listenTCP(self, port, factory, backlog=50, interface=""): """ @see: twisted.internet.interfaces.IReactorTCP.listenTCP """ p = tcp.Port(port, factory, backlog, interface, self) p.startListening() return p def connectTCP(self, host, port, factory, timeout=30, bindAddress=None): """ @see: twisted.internet.interfaces.IReactorTCP.connectTCP """ c = tcp.Connector(host, port, factory, timeout, bindAddress, self) c.connect() return c if TLSMemoryBIOFactory is not None: def listenSSL(self, port, factory, contextFactory, backlog=50, interface=""): """ @see: twisted.internet.interfaces.IReactorSSL.listenSSL """ port = self.listenTCP( port, TLSMemoryBIOFactory(contextFactory, False, factory), backlog, interface, ) port._type = "TLS" return port def connectSSL( self, host, port, factory, contextFactory, timeout=30, bindAddress=None ): """ @see: twisted.internet.interfaces.IReactorSSL.connectSSL """ return self.connectTCP( host, port, TLSMemoryBIOFactory(contextFactory, True, factory), timeout, bindAddress, ) else: def listenSSL(self, port, factory, contextFactory, backlog=50, interface=""): """ Non-implementation of L{IReactorSSL.listenSSL}. Some dependency is not satisfied. This implementation always raises L{NotImplementedError}. """ raise NotImplementedError( "pyOpenSSL 0.10 or newer is required for SSL support in " "iocpreactor. It is missing, so the reactor does not support " "SSL APIs." ) def connectSSL( self, host, port, factory, contextFactory, timeout=30, bindAddress=None ): """ Non-implementation of L{IReactorSSL.connectSSL}. Some dependency is not satisfied. This implementation always raises L{NotImplementedError}. """ raise NotImplementedError( "pyOpenSSL 0.10 or newer is required for SSL support in " "iocpreactor. It is missing, so the reactor does not support " "SSL APIs." ) def listenUDP(self, port, protocol, interface="", maxPacketSize=8192): """ Connects a given L{DatagramProtocol} to the given numeric UDP port. @returns: object conforming to L{IListeningPort}. """ p = udp.Port(port, protocol, interface, maxPacketSize, self) p.startListening() return p def listenMulticast( self, port, protocol, interface="", maxPacketSize=8192, listenMultiple=False ): """ Connects a given DatagramProtocol to the given numeric UDP port. EXPERIMENTAL. @returns: object conforming to IListeningPort. """ p = udp.MulticastPort( port, protocol, interface, maxPacketSize, self, listenMultiple ) p.startListening() return p def spawnProcess( self, processProtocol, executable, args=(), env={}, path=None, uid=None, gid=None, usePTY=0, childFDs=None, ): """ Spawn a process. """ if uid is not None: raise ValueError("Setting UID is unsupported on this platform.") if gid is not None: raise ValueError("Setting GID is unsupported on this platform.") if usePTY: raise ValueError("PTYs are unsupported on this platform.") if childFDs is not None: raise ValueError( "Custom child file descriptor mappings are unsupported on " "this platform." ) return Process(self, processProtocol, executable, args, env, path) def removeAll(self): res = list(self.handles) self.handles.clear() return res def install(): r = IOCPReactor() main.installReactor(r) __all__ = ["IOCPReactor", "install"] __init__.py 0000644 00000000277 15030441410 0006653 0 ustar 00 # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ I/O Completion Ports reactor """ from twisted.internet.iocpreactor.reactor import install __all__ = ["install"]
| ver. 1.4 |
Github
|
.
| PHP 8.2.28 | Generation time: 0.02 |
proxy
|
phpinfo
|
Settings