U
    *-e%                     @   s   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mZ d dl	m
Z
mZmZmZmZ ddddd	d
gZeeZG dd dZG dd de jZG dd de jZG dd de jZdaeddd	Zedeee ee ddd
ZdS )    N)contextmanager)getframeinfostack)AnyDictListOptionalSetTimerRequestTimerClientRequestQueueTimerServer	configureexpiresc                   @   s4   e Zd ZdZdddgZeeedddZdd	 Z	d
S )r
   a  
    Data object representing a countdown timer acquisition and release
    that is used between the ``TimerClient`` and ``TimerServer``.
    A negative ``expiration_time`` should be interpreted as a "release"
    request.

    .. note:: the type of ``worker_id`` is implementation specific.
              It is whatever the TimerServer and TimerClient implementations
              have on to uniquely identify a worker.
    	worker_idscope_idexpiration_timer   r   r   c                 C   s   || _ || _|| _d S )Nr   )selfr   r   r    r   d/var/www/html/Darija-Ai-Train/env/lib/python3.8/site-packages/torch/distributed/elastic/timer/api.py__init__    s    zTimerRequest.__init__c                 C   s2   t |tr.| j|jko,| j|jko,| j|jkS dS )NF)
isinstancer
   r   r   r   )r   otherr   r   r   __eq__%   s    


zTimerRequest.__eq__N)
__name__
__module____qualname____doc__	__slots__r   strfloatr   r   r   r   r   r   r
      s   
c                   @   s<   e Zd ZdZejeeddddZejedddZ	dS )	r   zk
    Client library to acquire and release countdown timers by communicating
    with the TimerServer.
    N)r   r   returnc                 C   s   dS )z
        Acquires a timer for the worker that holds this client object
        given the scope_id and expiration_time. Typically registers
        the timer with the TimerServer.
        Nr   )r   r   r   r   r   r   acquire5   s    zTimerClient.acquirer   c                 C   s   dS )z
        Releases the timer for the ``scope_id`` on the worker this
        client represents. After this method is
        called, the countdown timer on the scope is no longer in effect.
        Nr   )r   r   r   r   r   release>   s    zTimerClient.release)
r   r   r   r   abcabstractmethodr    r!   r#   r%   r   r   r   r   r   /   s
   c                   @   s@   e Zd ZdZejedddZejeee	e
 dddZdS )	r   zC
    Consumer queue holding timer acquisition/release requests
    r"   c                 C   s   dS )a  
        Returns the size of the queue at the time this method is called.
        Note that by the time ``get`` is called the size of the queue
        may have increased. The size of the queue should not decrease
        until the ``get`` method is called. That is, the following assertion
        should hold:

        size = q.size()
        res = q.get(size, timeout=0)
        assert size == len(res)

        -- or --

        size = q.size()
        res = q.get(size * 2, timeout=1)
        assert size <= len(res) <= size * 2
        Nr   r   r   r   r   sizeM   s    zRequestQueue.size)r*   timeoutr"   c                 C   s   dS )z
        Gets up to ``size`` number of timer requests in a blocking fashion
        (no more than ``timeout`` seconds).
        Nr   )r   r*   r+   r   r   r   getb   s    zRequestQueue.getN)r   r   r   r   r&   r'   intr*   r!   r   r
   r,   r   r   r   r   r   H   s
   c                       s   e Zd ZdZd eeed fddZej	e
e dddd	Zej	ee dd
ddZej	eeee
e f dddZej	eedddZeedddZdd Zdd Zdd ZddddZddddZ  ZS )!r   z
    Entity that monitors active timers and expires them
    in a timely fashion. This server is responsible for
    reaping workers that have expired timers.
    T)request_queuemax_intervaldaemonc                    s,   t    || _|| _|| _d| _d| _dS )a   
        :param request_queue: Consumer ``RequestQueue``
        :param max_interval: max time (in seconds) to wait
                             for an item in the request_queue
        :param daemon: whether to run the watchdog thread as a daemon
        NF)superr   _request_queue_max_interval_daemon_watchdog_thread_stop_signaled)r   r.   r/   r0   	__class__r   r   r   r   s    	
zTimerServer.__init__N)timer_requestsr"   c                 C   s   dS )a  
        Processes the incoming timer requests and registers them with the server.
        The timer request can either be a acquire-timer or release-timer request.
        Timer requests with a negative expiration_time should be interpreted
        as a release-timer request.
        Nr   r   r9   r   r   r   register_timers   s    zTimerServer.register_timers)
worker_idsr"   c                 C   s   dS )zA
        Clears all timers for the given ``worker_ids``.
        Nr   )r   r<   r   r   r   clear_timers   s    zTimerServer.clear_timers)deadliner"   c                 C   s   dS )z
        Returns all expired timers for each worker_id. An expired timer
        is a timer for which the expiration_time is less than or equal to
        the provided deadline.
        Nr   )r   r>   r   r   r   get_expired_timers   s    zTimerServer.get_expired_timers)r   r"   c                 C   s   dS )a  
        Reaps the given worker. Returns True if the worker has been
        successfully reaped, False otherwise. If any uncaught exception
        is thrown from this method, the worker is considered reaped
        and all associated timers will be removed.
        Nr   )r   r   r   r   r   _reap_worker   s    zTimerServer._reap_workerc              
   C   sH   z|  |W S  tk
rB } ztjd|d W Y dS d}~X Y nX dS )z
        Wraps ``_reap_worker(worker_id)``, if an uncaught exception is
        thrown, then it considers the worker as reaped.
        ziUncaught exception thrown from _reap_worker(), check that the implementation correctly catches exceptionsexc_infoTN)r@   	Exceptionlogerror)r   r   er   r   r   _reap_worker_no_throw   s    z!TimerServer._reap_worker_no_throwc              
   C   sJ   | j sFz|   W q  tk
rB } ztjd|d W 5 d }~X Y q X q d S )NzError running watchdogrA   )r6   _run_watchdogrC   rD   rE   )r   rF   r   r   r   _watchdog_loop   s
    zTimerServer._watchdog_loopc                 C   s   t d| j }| j|| j}| | t }t }| |	 D ]J\}}t
d|| | | |rt
d| || qFt
d| qF| | d S )N   z*Reaping worker_id=[%s]. Expired timers: %szSuccessfully reaped worker=[%s]z7Error reaping worker=[%s]. Will retry on next watchdog.)maxr2   r*   r,   r3   r;   timesetr?   itemsrD   info_get_scopesrG   addrE   r=   )r   Z
batch_sizer9   nowZreaped_worker_idsr   Zexpired_timersr   r   r   rH      s&    
 
 zTimerServer._run_watchdogc                 C   s   dd |D S )Nc                 S   s   g | ]
}|j qS r   r$   ).0rr   r   r   
<listcomp>   s     z+TimerServer._get_scopes.<locals>.<listcomp>r   r:   r   r   r   rP      s    zTimerServer._get_scopesr(   c                 C   sF   t dt| j| j| j tj| j| jd| _	t d | j	
  d S )Nz)Starting %s... max_interval=%s, daemon=%s)targetr0   zStarting watchdog thread...)rD   rO   typer   r3   r4   	threadingThreadrI   r5   startr)   r   r   r   rZ      s       
zTimerServer.startc                 C   sL   t dt| j d| _| jr>t d | j| j d | _n
t d d S )NzStopping %sTzStopping watchdog thread...z)No watchdog thread running, doing nothing)rD   rO   rW   r   r6   r5   joinr3   r)   r   r   r   stop   s    
zTimerServer.stop)T)r   r   r   r   r   r!   boolr   r&   r'   r   r
   r;   r	   r   r=   r   r    r?   r@   rG   rI   rH   rP   rZ   r\   __classcell__r   r   r7   r   r   k   s*      	Ztimer_clientc                 C   s   | a tdtt j dS )zM
    Configures a timer client. Must be called before using ``expires``.
    zTimer client configured to: %sN)_timer_clientrD   rO   rW   r   r_   r   r   r   r      s    )afterscopeclientc              	   c   s|   |dkrt dkrtdt }|dkrHtt d d }|j d|j }t |  }||| z
dV  W 5 || X dS )a-  
    Acquires a countdown timer that expires in ``after`` seconds from now,
    unless the code-block that it wraps is finished within the timeframe.
    When the timer expires, this worker is eligible to be reaped. The
    exact meaning of "reaped" depends on the client implementation. In
    most cases, reaping means to terminate the worker process.
    Note that the worker is NOT guaranteed to be reaped at exactly
    ``time.now() + after``, but rather the worker is "eligible" for being
    reaped and the ``TimerServer`` that the client talks to will ultimately
    make the decision when and how to reap the workers with expired timers.

    Usage::

        torch.distributed.elastic.timer.configure(LocalTimerClient())
        with expires(after=10):
            torch.distributed.all_reduce(...)
    Nz5Configure timer client before using countdown timers.rJ   r   #)	r`   RuntimeErrorr   r   filenamelinenorL   r#   r%   )ra   rb   rc   ZcallerZ
expirationr   r   r   r      s    
)NN)r&   loggingrX   rL   
contextlibr   inspectr   r   typingr   r   r   r   r	   __all__	getLoggerr   rD   r
   ABCr   r   r   r`   r   r!   r    r   r   r   r   r   <module>   s.   
# 	     