o
    ci(                     @   s  d Z ddlZddlZddlmZ ddlZddlZddlm  m	Z	 ddl
mZ ddlmZmZ ddlmZ zddlZW n eyF   edw ejejjejejjejejjejejjiZejejjejejjejejjejejj ej!ejj"ej#ejj$ej%ejjej&ejj'e(ejj)ej*ejj'ej+ejj,ej-ejj)ej.ejj)iZ/e rddl0Z0e0j1ejj$e0jejje0j!ejj"e0j#ejj$e0j%ejje0j2ejje0j&ejj'e0j(ejj,e0j*ejj'e0j+ejj,e0j-ejj)e0j.ejj)iZ3e0j1ej#e0jeje0j!ej!e0j#ej#e0j%ej%e0j2ej%e0j&ej&e0j(ej+e0j*ej*e0j+ej+e0j-ej-iZ4dd	 Z5d
d Z6dd Z7dd Z8dd Z9dd Z:dd Z;dd Z<dd Z=dd Z>G dd deZ?ej@ddG dd  d ZAG d!d" d"ZBdS )#z!Code to wrap some GLOO API calls.    N)List)	GcsClient)ReduceOptorch_available)_QueueActorzICan not import pygloo. Please run 'pip install pygloo' to install pygloo.c                 C   s   t j| |}|S )zCreate a GLOO context using GLOO APIs.

    Args:
        rank: the rank of this process.
        world_size: the number of processes of this collective group.

    Returns:
        context (pygloo.Context): a GLOO context.
    )pygloo
rendezvousContext)rank
world_sizecontext r   b/home/ubuntu/.local/lib/python3.10/site-packages/ray/util/collective/collective_group/gloo_util.pycreate_gloo_contextS   s   
r   c                 C   s   | t vrtd| t |  S )zMap the reduce op to GLOO reduce op type.

    Args:
        reduce_op: ReduceOp Enum (SUM/PRODUCT/MIN/MAX).

    Returns:
        (pygloo.ReduceOp): the mapped GLOO reduce op.
    z&Gloo does not support reduce op: '{}'.)GLOO_REDUCE_OP_MAPRuntimeErrorformat)	reduce_opr   r   r   get_gloo_reduce_opa   s   	r   c                 C   s\   t | tjrt| jj S t r%t | tjr%| j	st
| j S td| jtdt| )z2Return the corresponded GLOO dtype given a tensor.z Expect torch CPU tensor. Got {}.!Unsupported tensor type. Got: {}.)
isinstancenumpyndarrayNUMPY_GLOO_DTYPE_MAPdtypetyper   torchTensoris_cudaTORCH_GLOO_DTYPE_MAP
ValueErrorr   devicetensorr   r   r   get_gloo_tensor_dtypeo   s   

r$   c                 C   sB   t | tjr
| jjS t rt | tjrt| j S t	d
t| )z2Return the corresponded Cupy dtype given a tensor.^Unsupported tensor type. Got: {}. Supported CPU tensor types are: torch.Tensor, numpy.ndarray.)r   r   r   r   r   r   r   r   TORCH_NUMPY_DTYPE_MAPr    r   r"   r   r   r   get_numpy_tensor_dtype~   s   

r'   c                 C   sN   t | tjr
| jjS t rt | tjr| jrt	d| 
 S tdt| )z@Return the pointer to the underlying memory storage of a tensor.z8Torch tensor must be on CPU when using GLOO collectives.r%   )r   r   r   ctypesdatar   r   r   r   r   data_ptrr    r   r   r"   r   r   r   get_tensor_ptr   s   
r+   c                 C   s@   t | tjr	| jS t rt | tjrt| S td	t
| )z*Return the number of elements in a tensor.r   )r   r   r   sizer   r   r   numelr    r   r   r"   r   r   r   get_tensor_n_elements   s   
r.   c                 C   s    ddl m} |  d|  }|S )Nr   )get_ray_temp_dirz_collective/gloo/)ray._common.utilsr/   )
store_namer/   
store_pathr   r   r   get_gloo_store_path   s   r3   c                 C   sB   t | tjrdS t rt | tjr| jsdS dS tdt	| )NcpucudazUnrecognized tensor type: '{}'.)
r   r   r   r   r   r   r   r   r   r   r"   r   r   r   get_tensor_device   s   r6   c                 C   sF   t | tjrt| jS t rt | tjrt|  S t	d
t| )z)Return the shape of the tensor as a list.r%   )r   r   r   listshaper   r   r   r,   r    r   r   r"   r   r   r   get_tensor_shape   s   

r9   c                 C   s   d}t | tjrt |tjrt| | nHt r[t | tjr*t |tjr*| | n3t | tjrAt |tjrAt|}| | nt | tjrXt |tjrX| }t| | nd}nd}|sktd	t
| t
|dS )zCopy the content from src_tensor to dst_tensor.

    Args:
        dst_tensor: the tensor to copy from.
        src_tensor: the tensor to copy to.

    Returns:
        None
    TFzeUnsupported tensor type. Got: {} and {}. Supported CPU tensor types are: torch.Tensor, numpy.ndarray.N)r   r   r   copytor   r   r   copy_r    r   r   )
dst_tensor
src_tensorcopiedtr   r   r   copy_tensor   s8   

r@   c                   @   s   e Zd Zdd ZdS )	glooQueuec                 C   s&   z| j j|W S  ty   Y dS w )N)queue_queueindexr    )self
group_namer   r   r   rE      s
   zglooQueue.indexN)__name__
__module____qualname__rE   r   r   r   r   rA      s    rA   )num_cpusc                   @   s(   e Zd Zdd Zd
ddZdddZd	S )SignalActorc                 C   s   dd t |D | _|| _d S )Nc                 S   s   g | ]}t  qS r   )asyncioEvent).0_r   r   r   
<listcomp>   s    z(SignalActor.__init__.<locals>.<listcomp>)rangeready_eventsr   )rF   r   r   r   r   __init__   s   
zSignalActor.__init__Fc                 C   s(   | j |   |r| j |   d S d S N)rS   setclear)rF   r
   rW   r   r   r   send  s   zSignalActor.sendTc                    s2   |rt | jD ]}| j|  I d H  qd S d S rU   )rR   r   rS   wait)rF   should_waitir   r   r   rY     s   zSignalActor.waitN)F)T)rH   rI   rJ   rT   rX   rY   r   r   r   r   rL      s    
rL   c                   @   s   e Zd ZdefddZdededefddZdedefd	d
Zdede	fddZ
dee dee	 fddZdee fddZdd ZdS )RayInternalKvStorerG   c                 C   s<   || _ t  | _tjjjj}t	|d| _
t| j
 d S )N)address)_group_namerayget_runtime_context
get_job_id_job_id_privateworker_global_nodegcs_addressr   _gcs_clientinternal_kv_initialize_internal_kv)rF   rG   rf   r   r   r   rT     s
   zRayInternalKvStore.__init__keyr)   returnc                 C   s   |  |}t||}|S rU   )-_RayInternalKvStore__concat_key_with_prefixesrh   _internal_kv_put)rF   rj   r)   retr   r   r   rV     s   
zRayInternalKvStore.setc                 C      |  |}t|}|S rU   )rl   rh   _internal_kv_getrF   rj   rn   r   r   r   get     

zRayInternalKvStore.getc                 C   ro   rU   )rl   rh   _internal_kv_delrq   r   r   r   delete!  rs   zRayInternalKvStore.deletekeysc                 C   s"   g }|D ]
}| | | q|S rU   )appendru   )rF   rv   resultsrj   r   r   r   del_keys&  s   zRayInternalKvStore.del_keysc                 C   sD   	 d}|D ]}|  |}t|}|sd} nq|rdS td q)NTF   )rl   rh   _internal_kv_existstimesleep)rF   rv   	all_existrj   resultr   r   r   rY   ,  s   


zRayInternalKvStore.waitc                 C   s   | j  d| j d| S )zlConcat the necessary prefixes and key for isolation purpose for
        different jobs and different groups.-)rb   r^   )rF   original_keyr   r   r   __concat_key_with_prefixes9  s   z-RayInternalKvStore.__concat_key_with_prefixesN)rH   rI   rJ   strrT   bytesboolrV   rr   intru   r   ry   rY   rl   r   r   r   r   r\     s    r\   )C__doc__rM   r|   typingr   r   r_   ray.experimental.internal_kvexperimentalrh   ray._rayletr   ray.util.collective.typesr   r   ray.util.queuer   r   ImportErrorSUMPRODUCTMINMAXr   int_glooDataType_t	glooInt64uint8	glooUint8uint32
glooUint32uint64
glooUint64int8glooInt8int32	glooInt32int64halfglooFloat16floatglooFloat64float16float32glooFloat32float64doubler   r   r   longr   r&   r   r   r$   r'   r+   r.   r3   r6   r9   r@   rA   remoterL   r\   r   r   r   r   <module>   s    




	























+
