o ٭jT'@sUdZddlZddlZddlZddlZddlmZddlmZm Z ddl m Z m Z m Z mZmZddlZddlmZddlmZmZmZmZmZmZed Ze eeeeffZee d <e eeefZ!ee d <e ee!fZ"ee d <eGd ddZ#de!ddfddZ$d2ddZ%defddZ&defddZ'de(fddZ)de ddfddZ*defddZ+d ed!edefd"d#Z,d$ej-defd%d&Z.e Gd'd(d(eZ/d ej0d)e/dej0fd*d+Z1d2d,d-Z2de efd.d/Z3Gd0d1d1Z4dS)3z-XGBoost collective communication related API.N) dataclass)IntEnumunique)AnyDictOptional TypeAliasUnion)_T)_LIB _check_call build_infoc_str make_jcargspy_strz[xgboost.collective]_Conf_ArgVals_Argsc@steZdZUdZdZeeed<dZeeed<dZ ee ed<dZ eeed<dZ eeed<de d e fd d ZdS) ConfigaUser configuration for the communicator context. This is used for easier integration with distributed frameworks. Users of the collective module can pass the parameters directly into tracker and the communicator. .. versionadded:: 3.0 Attributes ---------- retry : See `dmlc_retry` in :py:meth:`init`. timeout : See `dmlc_timeout` in :py:meth:`init`. This is only used for communicators, not the tracker. They are different parameters since the timeout for tracker limits only the time for starting and finalizing the communication group, whereas the timeout for communicators limits the time used for collective operations, like :py:meth:`allreduce`. tracker_host_ip : See :py:class:`~xgboost.tracker.RabitTracker`. tracker_port : See :py:class:`~xgboost.tracker.RabitTracker`. tracker_timeout : See :py:class:`~xgboost.tracker.RabitTracker`. Nretrytimeouttracker_host_ip tracker_porttracker_timeoutargsreturncCs,|jdur |j|d<|jdur|j|d<|S)z*Update the arguments for the communicator.N dmlc_retry dmlc_timeout)rrselfrr!O/var/www/html/CropPilot/venv/lib/python3.10/site-packages/xgboost/collective.pyget_comm_config:s    zConfig.get_comm_config)__name__ __module__ __qualname____doc__rrint__annotations__rrstrrrrr#r!r!r!r"rs rrrcKstttdi|dS)aInitialize the collective library with arguments. Parameters ---------- args : Keyword arguments representing the parameters and their values. Accepted parameters: - dmlc_communicator: The type of the communicator. * rabit: Use Rabit. This is the default if the type is unspecified. * federated: Use the gRPC interface for Federated Learning. Only applicable to the Rabit communicator: - dmlc_tracker_uri: Hostname of the tracker. - dmlc_tracker_port: Port number of the tracker. - dmlc_task_id: ID of the current task, can be used to obtain deterministic - dmlc_retry: The number of retry when handling network errors. - dmlc_timeout: Timeout in seconds. - dmlc_nccl_path: Path to load (dlopen) nccl for GPU-based communication. Only applicable to the Federated communicator: - federated_server_address: Address of the federated server. - federated_world_size: Number of federated workers. - federated_rank: Rank of the current worker. - federated_server_cert: Server certificate file path. Only needed for the SSL mode. - federated_client_key: Client key file path. Only needed for the SSL mode. - federated_client_cert: Client certificate file path. Only needed for the SSL mode. Use upper case for environment variables, use lower case for runtime configuration. Nr!)r r XGCommunicatorInitr)rr!r!r"initCs#r,cCttdS)zFinalize the communicator.N)r r XGCommunicatorFinalizer!r!r!r"finalizeir/cC t}|S)zjGet rank of current process. Returns ------- rank : int Rank of current process. )r XGCommunicatorGetRankretr!r!r"get_ranknr5cCr1)z`Get total number workers. Returns ------- n : Total number of process. )r XGCommunicatorGetWorldSizer3r!r!r"get_world_sizezr6r8cCst}t|S)z.If the collective communicator is distributed.)r XGCommunicatorIsDistributedbool)is_distr!r!r"is_distributedsr<msgcCsPt|ts t|}t}|dkrttt|dSt|dddS)zPrint message to the communicator. This function can be used to communicate the information of the progress to the communicator. Parameters ---------- msg : str The message to be printed to the communicator. rT)flushN) isinstancer*r r9r XGCommunicatorPrintrstripprint)r=r;r!r!r"communicator_prints rCcCs*t}ttt||j}t|S)zdGet the processor name. Returns ------- name : The name of processor(host) )ctypesc_char_pr r XGCommunicatorGetProcessorNamebyrefvaluer)name_strrHr!r!r"get_processor_namesrJdatarootcCst}t}||kr |dusJdtj|tjd}t||_tt t |t tj|||krStj |j}tt t|tj|j|t|j}~|Stt tt|tj|j|~|S)aSBroadcast object from one node to all other nodes. Parameters ---------- data : any type that can be pickled Input data, if current rank does not equal root, this can be None root : int Rank of the node to broadcast data from. Returns ------- object : int the result of broadcast. Nz&need to pass in data when broadcasting)protocol)r5rDc_ulongpickledumpsHIGHEST_PROTOCOLlenrHr r XGCommunicatorBroadcastrGsizeofc_charcastc_void_ploadsrawrE)rKrLranklengthsdptrr!r!r" broadcasts8  r^dtypecCstddtddtddtddtd d td d td dtddtddtddtddi }z |tddiWn tyNYnw||vr[td|d||S)Nfloat16rfloat32r float64int8int16int32int64uint8uint16 uint32 uint64 float128z data type z* is not supported on the current platform.)npr_update TypeError)r_ dtype_mapr!r!r" _map_dtypes(             rzc@s(eZdZdZdZdZdZdZdZdZ dS) Opz#Supported operations for allreduce.rr rcrurergN) r$r%r&r'MAXMINSUM BITWISE_AND BITWISE_OR BITWISE_XORr!r!r!r"r{sr{opcCsNt|tjs td|}tt|j t j |j t |jt||S)a'Perform allreduce, return the result. Parameters ---------- data : Input data. op : Reduction operator. Returns ------- result : The result of allreduce, have same shape as data Notes ----- This function is not thread-safe. z%allreduce only takes in numpy.ndarray)r?rvndarrayrxravelcopyr r XGCommunicatorAllreducerDdata_asrWsizerzr_r()rKrbufr!r!r" allreduces   rcCr-)zKill the process.N)r r XGCommunicatorSignalErrorr!r!r!r" signal_error$r0rcCsddlm}|jdurtj|j}nt|dr%t|jdkr%|jd}nd}|s+dSt |}|s4dSd}|D] }| drC|}nq8|durQtj ||}|SdS)Nr)lib__path__z libnccl.so) nvidia.ncclr__file__ospathdirnamehasattrrRrlistdir startswithjoin)rrfileslibnamenamerr!r!r" _find_nccl)s*     rc@sBeZdZdZdeddfddZdefddZdeddfd d Z dS) CommunicatorContextzNA context controlling collective communicator initialization and finalization.rrNcKsf||_d}||ddurdSt}|dsdSzt}|r&||j|<WdSWdSty2YdSw)Ndmlc_nccl_pathUSE_DLOPEN_NCCL)rgetrr ImportError)r rkeybinforr!r!r"__init__Ns zCommunicatorContext.__init__cCs*tdi|jts Jtd|jS)Nz8-------------- communicator say hello ------------------r!)r,rr<LOGGERdebug)r r!r!r" __enter__`s  zCommunicatorContext.__enter__cGsttddS)Nz7--------------- communicator say bye ------------------)r/rrrr!r!r"__exit__fszCommunicatorContext.__exit__) r$r%r&r'rrrrrrr!r!r!r"rKs r)rN)5r'rDloggingrrO dataclassesrenumrrtypingrrrrr numpyrv_typingr corer r rrrr getLoggerrr*r(rr)rrrr,r/r5r8r:r<rCrJr^r_rzr{rrrrrr!r!r!r"s@    * &  0 !"