U
    nËjc%  ã                   @  sœ  U d dl mZ d dlZd dlZd dlZd dlZd dlmZ d dlm	Z	 d dl
mZmZ d dlmZmZ ddlmZmZmZ dd	lmZ dd
lmZ ddlmZ ddlmZmZ ddlmZmZm Z  ddl!m"Z"m#Z# ddl$m%Z% ej&dkrúd dlm'Z'm(Z( nd dl)m'Z'm(Z( dZ*edƒZ+e'dƒZ,e"dƒZ-de.d< e"dƒZ/de.d< e"dƒZ0de.d< dddœdddd dd!œd"d#„Z1d$d%œd&d'„Z2d(d%œd)d*„Z3e4d+k�r˜e3ƒ  dS ),é    )ÚannotationsN)Údeque)ÚCallable)Úmodule_from_specÚspec_from_file_location)ÚTypeVarÚcasté   )Úcurrent_timeÚget_async_backendÚget_cancelled_exc_class)ÚBrokenWorkerProcess)Úopen_process)ÚCapacityLimiter)ÚCancelScopeÚ
fail_after)ÚByteReceiveStreamÚByteSendStreamÚProcess)ÚRunVarÚcheckpoint_if_cancelled)ÚBufferedByteReceiveStream)é   é   )ÚTypeVarTupleÚUnpacki,  ÚT_RetvalÚPosArgsTÚ_process_pool_workerszRunVar[set[Process]]Ú_process_pool_idle_workersz$RunVar[deque[tuple[Process, float]]]Ú_default_process_limiterzRunVar[CapacityLimiter]F)ÚcancellableÚlimiterz&Callable[[Unpack[PosArgsT]], T_Retval]zUnpack[PosArgsT]ÚboolzCapacityLimiter | None)ÚfuncÚargsr!   r"   Úreturnc                ‡  s  dddœ‡ ‡‡‡fdd„}t ƒ I dH  tjd| |ftjd�}zt ¡ ‰t ¡ }W n@ tk
r�   tƒ ‰t	ƒ }t ˆ¡ t |¡ t
ƒ  ˆ¡ Y nX |pštƒ 4 I dH �šX |�rv| ¡ \‰}ˆjdk�rjttˆjƒ‰tttˆjƒƒ‰ tƒ }g }	|�r6||d	 d
  tk �r�q6| ¡ \}
}|
 ¡  ˆ |
¡ |	 |
¡ qìtdd��  |	D ]}| ¡ I dH  �qFW 5 Q R X �qŽˆ ˆ¡ q¨tjddtg}t |t!j"t!j"d�I dH ‰z”ttˆjƒ‰tttˆjƒƒ‰ t#dƒ� ˆ  $d¡I dH }W 5 Q R X |dk�röt%d|›�ƒ‚t&tj'd ddƒ}tjdtj(|ftjd�}||ƒI dH  W nR t%t)ƒ fk
�rN   ‚ Y n6 t*k
�r‚ } zˆ ¡  t%dƒ|‚W 5 d}~X Y nX ˆ +ˆ¡ t| d��Z z6tt,||ƒI dH ƒW ¢ W  5 Q R £ W  5 Q I dH R £ S ˆˆk�rî| ˆtƒ f¡ X W 5 Q R X W 5 Q I dH R X dS )a¸  
    Call the given function with the given arguments in a worker process.

    If the ``cancellable`` option is enabled and the task waiting for its completion is
    cancelled, the worker process running it will be abruptly terminated using SIGKILL
    (or ``terminateProcess()`` on Windows).

    :param func: a callable
    :param args: positional arguments for the callable
    :param cancellable: ``True`` to allow cancellation of the operation while it's
        running
    :param limiter: capacity limiter to use to limit the total amount of processes
        running (if omitted, the default limiter is used)
    :return: an awaitable that yields the return value of the function.

    ÚbytesÚobject)Úpickled_cmdr&   c                 “  s  z^ˆ  | ¡I d H  ˆ  dd¡I d H }| d¡\}}|dkrHtd|›�ƒ‚ˆ  t|ƒ¡I d H }W nˆ tk
ræ } zjˆ ˆ¡ z0ˆ ¡  t	dd�� ˆ 
¡ I d H  W 5 Q R X W n tk
r¾   Y nX t|tƒ ƒrÐ‚ nt|‚W 5 d }~X Y nX t |¡}|dk�rt|tƒ�st‚|‚n|S d S )	Nó   
é2   ó    )ó   RETURNó	   EXCEPTIONú-Worker process returned unexpected response: T©Zshieldr.   )ÚsendZreceive_untilÚsplitÚRuntimeErrorZreceive_exactlyÚintÚBaseExceptionÚdiscardÚkillr   ÚacloseÚProcessLookupErrorÚ
isinstancer   r   ÚpickleÚloadsÚAssertionError)r)   ÚresponseÚstatusÚlengthZpickled_responseÚexcÚretval©ZbufferedÚprocessÚstdinÚworkers© ú4/tmp/pip-unpacked-wheel-k8mofgf9/anyio/to_process.pyÚsend_raw_command=   s2    ÿ


z"run_sync.<locals>.send_raw_commandNÚrun)Úprotocolr   r	   Tr0   z-uz-m)rE   Ústdouté   é   ó   READY
r/   Ú__main__Ú__file__Úinitz*Error during worker process initialization)-r   r;   ÚdumpsÚHIGHEST_PROTOCOLr   Úgetr   ÚLookupErrorÚsetr   r   Z#setup_process_pool_exit_at_shutdownÚcurrent_default_process_limiterÚpopÚ
returncoder   r   rE   r   r   rL   r
   ÚWORKER_MAX_IDLE_TIMEÚpopleftr7   ÚremoveÚappendr   r8   ÚsysÚ
executableÚ__name__r   Ú
subprocessÚPIPEr   Zreceiver   ÚgetattrÚmodulesÚpathr   r5   Úaddr   )r$   r!   r"   r%   rI   ÚrequestZidle_workersZ
idle_sinceÚnowZkilled_processesZprocess_to_killZkilled_processÚcommandÚmessageÚmain_module_pathÚpickledrA   rG   rC   rH   Úrun_sync&   s�    !


ÿ
  ÿ
ÿ

ÿ
þÿþ
6
rn   r   )r&   c                  C  sB   z
t  ¡ W S  tk
r<   tt ¡ p&dƒ} t  | ¡ |  Y S X dS )z“
    Return the capacity limiter that is used by default to limit the number of worker
    processes.

    :return: a capacity limiter object

    é   N)r    rU   rV   r   ÚosÚ	cpu_countrW   )r"   rG   rG   rH   rX   ¶   s    

rX   ÚNonec               
   C  s  t j} t j}ttjƒt _ttjdƒt _|j d¡ d  }}zt 	| j¡^}}W n: t
k
rd   Y d S  tk
rˆ } z|}W 5 d }~X Y nÜX |dkrÐ|\}}z||Ž }W n& tk
rÌ } z|}W 5 d }~X Y nX n”|dk�rd|\t _}t jd= |�rdtj |¡�rdz:td|ƒ}	|	�r8|	j�r8t|	ƒ}
|	j |
¡ |
t jd< W n( tk
�rb } z|}W 5 d }~X Y nX z4|d k	�r„d}t |tj¡}nd}t |tj¡}W n: tk
�rÒ } z|}d}t |tj¡}W 5 d }~X Y nX |j d	|t|ƒf ¡ |j |¡ t|tƒr2|‚q2d S )
NÚwrO   rJ   rR   rP   Ú__mp_main__r.   r-   s   %s %d
)r_   rE   rL   Úopenrp   ÚdevnullÚbufferÚwriter;   ÚloadÚEOFErrorr5   rf   re   Úisfiler   Úloaderr   Úexec_modulerS   rT   Úlenr:   Ú
SystemExit)rE   rL   rB   Ú	exceptionrj   r%   rA   r$   rl   ÚspecÚmainr?   rm   rG   rG   rH   Úprocess_workerÆ   sX    



 
rƒ   rP   )5Ú
__future__r   rp   r;   rb   r_   Úcollectionsr   Úcollections.abcr   Úimportlib.utilr   r   Útypingr   r   Z_core._eventloopr
   r   r   Z_core._exceptionsr   Z_core._subprocessesr   Z_core._synchronizationr   Z_core._tasksr   r   Úabcr   r   r   Zlowlevelr   r   Zstreams.bufferedr   Úversion_infor   r   Ztyping_extensionsr[   r   r   r   Ú__annotations__r   r    rn   rX   rƒ   ra   rG   rG   rG   rH   Ú<module>   sH    
ÿü ;
