
    yjo                        d Z ddlmZ ddlmZ ddlZddlmZ ddlm	Z	 dd	l
mZ g Z G d
 d          Z G d d          Z G d de          Z G d de          Z G d de          Z G d de          ZdS )zBThis is definition of dataset class, which is high performance IO.    )annotations)text_formatN)data_feed_pb2   )
deprecated   )corec                  "    e Zd ZdZd Zdd	dZdS )
DatasetFactoryaf  
    DatasetFactory is a factory which create dataset by its name,
    you can create "QueueDataset" or "InMemoryDataset", or "FileInstantDataset",
    the default is "QueueDataset".

    Example:
        .. code-block:: python

            >>> import paddle.base as base
            >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
    c                    dS )Init.N selfs    c/lsinfo/ai/hellotax_ai/data_center/backend/venv/lib/python3.11/site-packages/paddle/base/dataset.py__init__zDatasetFactory.__init__*   s        QueueDatasetreturnDatasetBasec                p    	  t                      |                     }|S #  t          d| d          xY w)a  
        Create "QueueDataset" or "InMemoryDataset", or "FileInstantDataset",
        the default is "QueueDataset".

        Args:
            datafeed_class(str): datafeed class name, QueueDataset or InMemoryDataset.
                                 Default is QueueDataset.

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset()
        zdatafeed class z does not exist)globals
ValueError)r   datafeed_classdatasets      r   create_datasetzDatasetFactory.create_dataset.   sG    	P/gii/11GN	PN~NNNOOOs     5N)r   )r   r   )__name__
__module____qualname____doc__r   r   r   r   r   r   r      sM        
 
  P P P P P P Pr   r   c                      e Zd ZdZd Zd Zd Zd ZddZd Z	d	 Z
d
 Zd Zd Zd Zd Zd Zd Zd Zd Zd Zd Zd Zd ZdS )r   zBase dataset class.c                    t          j                    | _        d| j        _        t	          j        d          | _        d| _        g | _        d| _	        d| _
        dS )r   catMultiSlotDatasetr   FN)r   DataFeedDesc
proto_descpipe_commandr	   Datasetr   
thread_numfilelist
use_ps_gpupsgpur   s    r   r   zDatasetBase.__init__G   sP     (466',$|$677


r   c                    || j         _        dS )a  
        Set pipe command of current dataset
        A pipe command is a UNIX pipeline command that can be used only

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset()
                >>> dataset.set_pipe_command("python my_script.py")

        Args:
            pipe_command(str): pipe command

        N)r&   r'   )r   r'   s     r   set_pipe_commandzDatasetBase.set_pipe_commandS   s      (4$$$r   c                    || j         _        dS )aY  
        Set so parser name of current dataset

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset()
                >>> dataset.set_so_parser_name("./abc.so")

        Args:
            pipe_command(str): pipe command

        N)r&   so_parser_name)r   r0   s     r   set_so_parser_namezDatasetBase.set_so_parser_namee   s     *8&&&r   c                    || j         _        dS )ap  
        Set rank_offset for merge_pv. It set the message of Pv.

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset()
                >>> dataset.set_rank_offset("rank_offset")

        Args:
            rank_offset(str): rank_offset's name

        N)r&   rank_offset)r   r3   s     r   set_rank_offsetzDatasetBase.set_rank_offsetv   s     '2###r   Tc                N    |r| j                             ||           || _        dS )a  
        set fea eval mode for slots shuffle to debug the importance level of
        slots(features), fea_eval need to be set True for slots shuffle.

        Args:
            record_candidate_size(int): size of instances candidate to shuffle
                                        one slot
            fea_eval(bool): whether enable fea eval mode to enable slots shuffle.
                            default is True.

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
                >>> dataset.set_fea_eval(1000000, True)

        N)r   set_fea_evalfea_eval)r   record_candidate_sizer7   s      r   r6   zDatasetBase.set_fea_eval   s1    &  	GL%%h0EFFF r   c                j    | j         r+t          |          }| j                            |           dS dS )a  
        Slots Shuffle
        Slots Shuffle is a shuffle method in slots level, which is usually used
        in sparse feature with large scale of instances. To compare the metric, i.e.
        auc while doing slots shuffle on one or several slots with baseline to
        evaluate the importance level of slots(features).

        Args:
            slots(list[string]): the set of slots(string) to do slots shuffle.

        Examples:
            import paddle.base as base
            dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
            dataset.set_merge_by_lineid()
            #suppose there is a slot 0
            dataset.slots_shuffle(['0'])
        N)r7   setr   slots_shuffler   slots	slots_sets      r   r;   zDatasetBase.slots_shuffle   s?    $ = 	2E

IL&&y11111	2 	2r   c                    || j         _        dS )aV  
        Set batch size. Will be effective during training

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset()
                >>> dataset.set_batch_size(128)

        Args:
            batch_size(int): batch size

        N)r&   
batch_size)r   r@   s     r   set_batch_sizezDatasetBase.set_batch_size   s     &0"""r   c                    || j         _        dS )ak  
        Set pv batch size. It will be effective during enable_pv_merge

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset()
                >>> dataset.set_pv_batch_size(128)
        Args:
            pv_batch_size(int): pv batch size

        N)r&   pv_batch_size)r   rC   s     r   set_pv_batch_sizezDatasetBase.set_pv_batch_size   s     )6%%%r   c                H    | j                             |           || _        dS )aH  
        Set thread num, it is the num of readers.

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset()
                >>> dataset.set_thread(12)

        Args:
            thread_num(int): thread num
        N)r   set_thread_numr)   r   r)   s     r   
set_threadzDatasetBase.set_thread   s%     	##J///$r   c                H    | j                             |           || _        dS )aO  
        Set file list in current worker.

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset()
                >>> dataset.set_filelist(['a.txt', 'b.txt'])

        Args:
            filelist(list): file list
        N)r   set_filelistr*   )r   r*   s     r   rJ   zDatasetBase.set_filelist   s%     	!!(+++ r   c                    || j         _        d S N)r&   
input_type)r   rM   s     r   set_input_typezDatasetBase.set_input_type   s    %/"""r   c                   | j         j        }|D ]}|j                                        }d|_        |j        |_        t          j                                        s1|j	        dk    r&d|_
        |j                            |j                   |j        t          j        k    rd|_        |j        t          j        k    rd|_        |j        t          j        k    rd|_        t%          d          dS )aC  
        Set Variables which you will use.

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> paddle.enable_static()
                >>> dataset = base.DatasetFactory().create_dataset()
                >>> data = paddle.static.data(name="data", shape=[None, 10, 10], dtype="int64")
                >>> label = paddle.static.data(name="label", shape=[None, 1], dtype="int64", lod_level=1)
                >>> dataset.set_use_var([data, label])

        Args:
            var_list(list): variable list
        Tr   floatuint64uint32zPCurrently, base.dataset only supports dtype=float32, dtype=int32 and dtype=int64N)r&   multi_slot_descr=   addis_usednamepaddle	frameworkin_pir_mode	lod_levelis_denseshapeextenddtypefloat32typeint64int32r   )r   var_list
multi_slotvarslot_vars        r   set_use_varzDatasetBase.set_use_var   s    " _4
 	 	C!'++--H#HHHM#//11 5=A%%(,H%N))#)444yFN** 'fl** (fl** ( f  	 	r   c                <    | j                             ||           dS )at  
        Set hdfs config: fs name ad ugi

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset()
                >>> dataset.set_hdfs_config("my_fs_name", "my_fs_ugi")

        Args:
            fs_name(str): fs name
            fs_ugi(str): fs ugi
        N)r   set_hdfs_config)r   fs_namefs_ugis      r   ri   zDatasetBase.set_hdfs_config  s"     	$$Wf55555r   c                :    | j                             |           dS )ap  
        Set customized download cmd: download_cmd

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset()
                >>> dataset.set_download_cmd("./read_from_afs")

        Args:
            download_cmd(str): customized download command
        N)r   set_download_cmd)r   download_cmds     r   rm   zDatasetBase.set_download_cmd0  s      	%%l33333r   c                :   | j         t          | j                  k    rt          | j                  | _         | j                            | j                    | j                            |                                            | j                                         dS )h
        Set data_feed_desc before load or shuffle,
        user no need to call this function.
        N)r)   lenr*   r   rF   set_data_feed_descdesccreate_readersr   s    r   _prepare_to_runzDatasetBase._prepare_to_run@  s|    
 ?S////!$-00DO##DO444''		444##%%%%%r   c                l    d| _         t          j                    s	d| _         dS | j         r	|| _        dS dS )zQ
        set use_ps_gpu flag

        Args:
            use_ps_gpu: bool
        TFN)r+   r	   _is_compiled_with_heterpsr,   )r   r,   s     r   _set_use_ps_gpuzDatasetBase._set_use_ps_gpuK  sF     -// 	#DOOO_ 	DJJJ	 	r   c                8    | j                                          d S rL   )r   destroy_readersr   s    r   _finish_to_runzDatasetBase._finish_to_runY  s    $$&&&&&r   c                4    t          j        | j                  S )aF  
        Returns a protobuf message for this DataFeedDesc

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset()
                >>> print(dataset.desc())

        Returns:
            A string message
        )r   MessageToStringr&   r   s    r   rs   zDatasetBase.desc\  s     *4?;;;r   c                    d S rL   r   rG   s     r   _dynamic_adjust_before_trainz(DatasetBase._dynamic_adjust_before_trainl      r   c                    d S rL   r   r   s    r   _dynamic_adjust_after_trainz'DatasetBase._dynamic_adjust_after_traino  r   r   N)T)r   r   r   r    r   r.   r1   r4   r6   r;   rA   rD   rH   rJ   rN   rg   ri   rm   ru   rx   r{   rs   r   r   r   r   r   r   r   D   sL       
 
 
4 4 4$8 8 8"2 2 2"! ! ! !.2 2 2,0 0 0"6 6 6 % % %"! ! !"0 0 0# # #J6 6 6"4 4 4 	& 	& 	&  ' ' '< < <       r   r   c                  X    e Zd ZdZ edd           fd            Z edd          d             Z edd          d	             Z edd
          d             Z edd          d             Z	 edd          d             Z
 edd          d             Z edd          d             Zd Zd Z edd          d             Zd Zd Zd Zd Z edd          dGd            Z edd          dHd!            Z edd"          dId$            Z edd%          d&             Z edd'          d(             Zd) Z edd*          dJd,            Z edd-          dKd/            Z edd0          d1             Z edd2          d3             Z edd4          dLd6            Z edd7          d8             Zd9 Z d: Z!d; Z" edd<          dKd=            Z# edd>          dKd?            Z$dJd@Z%dA Z&dB Z'dC Z(dMdEZ)dF Z* xZ+S )NInMemoryDatasetz
    InMemoryDataset, it will load data into memory
    and shuffle data before training.
    This class should be created by DatasetFactory

    Example:
        dataset = paddle.base.DatasetFactory().create_dataset("InMemoryDataset")
    2.0.0z"paddle.distributed.InMemoryDatasetsince	update_toc                   t                                                       d| j        _        d| _        d| _        d| _        d| _        d| _        d| _	        d| _
        d| _        d| _        d| _        d| _        d| _        dS )r   MultiSlotInMemoryDataFeedNFTr   )superr   r&   rV   fleet_send_batch_sizeis_user_set_queue_num	queue_numparse_ins_idparse_contentparse_logkeymerge_by_sidenable_pv_mergemerge_by_lineidfleet_send_sleep_secondstrainer_numpass_idr   	__class__s    r   r   zInMemoryDataset.__init__}  s     	:%)"%*"!"! $$(,%r   z1paddle.distributed.InMemoryDataset._set_feed_typec                t    || j         _        | j         j        dk    rt          j        d          | _        dS dS )z$
        Set data_feed_desc
        SlotRecordInMemoryDataFeedSlotRecordDatasetN)r&   rV   r	   r(   r   )r   data_feed_types     r   set_feed_typezInMemoryDataset.set_feed_type  s>      .?#???<(;<<DLLL @?r   z2paddle.distributed.InMemoryDataset._prepare_to_runc                   | j         dk    rd| _         | j                            | j                    | j        | j         | _        | j                            | j                   | j                            | j                   | j                            | j                   | j        	                    | j
                   | j                            | j                   | j                            | j                   | j                            |                                            | j                                         | j                                         dS )rp   r   r   N)r)   r   rF   r   set_queue_numset_parse_ins_idr   set_parse_contentr   set_parse_logkeyr   set_merge_by_sidr   set_enable_pv_merger   rr   rs   create_channelrt   r   s    r   ru   zInMemoryDataset._prepare_to_run  s"    ?aDO##DO444>!!_DN""4>222%%d&7888&&t'9:::%%d&7888%%d&7888(()=>>>''		444##%%%##%%%%%r   z?paddle.distributed.InMemoryDataset._dynamic_adjust_before_trainc                    | j         s>| j        r| j                            |d           n| j                            |d           | j                            |           d S NTF)r   r+   r   dynamic_adjust_channel_numdynamic_adjust_readers_numrG   s     r   r   z,InMemoryDataset._dynamic_adjust_before_train  sj    
 ) 	K K77
DIIII77
EJJJ//
;;;;;r   z>paddle.distributed.InMemoryDataset._dynamic_adjust_after_trainc                    | j         sH| j        r!| j                            | j        d           n | j                            | j        d           | j                            | j                   d S r   )r   r+   r   r   r)   r   r   s    r   r   z+InMemoryDataset._dynamic_adjust_after_train  sp    
 ) 	P P77NNNN77OOO//@@@@@r   z1paddle.distributed.InMemoryDataset._set_queue_numc                "    d| _         || _        dS )a  
        Set Dataset output queue num, training threads get data from queues

        Args:
            queue_num(int): dataset output queue num

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
                >>> dataset.set_queue_num(12)

        TN)r   r   )r   r   s     r   r   zInMemoryDataset.set_queue_num  s    & &*""r   z4paddle.distributed.InMemoryDataset._set_parse_ins_idc                    || _         dS )ak  
        Set id Dataset need to parse ins_id

        Args:
            parse_ins_id(bool): if parse ins_id or not

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
                >>> dataset.set_parse_ins_id(True)

        N)r   )r   r   s     r   r   z InMemoryDataset.set_parse_ins_id      & )r   z5paddle.distributed.InMemoryDataset._set_parse_contentc                    || _         dS )ao  
        Set if Dataset need to parse content

        Args:
            parse_content(bool): if parse content or not

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
                >>> dataset.set_parse_content(True)

        N)r   )r   r   s     r   r   z!InMemoryDataset.set_parse_content  s    & +r   c                    || _         dS )al  
        Set if Dataset need to parse logkey

        Args:
            parse_content(bool): if parse logkey or not

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
                >>> dataset.set_parse_logkey(True)

        N)r   )r   r   s     r   r   z InMemoryDataset.set_parse_logkey  s     )r   c                    || _         dS )aG  
        Set trainer num

        Args:
            trainer_num(int): trainer num

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
                >>> dataset._set_trainer_num(1)

        N)r   )r   r   s     r   _set_trainer_numz InMemoryDataset._set_trainer_num  s     'r   z4paddle.distributed.InMemoryDataset._set_merge_by_sidc                    || _         dS )a  
        Set if Dataset need to merge sid. If not, one ins means one Pv.

        Args:
            merge_by_sid(bool): if merge sid or not

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
                >>> dataset.set_merge_by_sid(True)

        N)r   )r   r   s     r   r   z InMemoryDataset.set_merge_by_sid-  r   r   c                    || _         dS )aq  
        Set if Dataset need to merge pv.

        Args:
            enable_pv_merge(bool): if enable_pv_merge or not

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
                >>> dataset.set_enable_pv_merge(True)

        N)r   )r   r   s     r   r   z#InMemoryDataset.set_enable_pv_mergeB  s      /r   c                8    | j                                          dS )a[  
        Merge pv instance and convey it from input_channel to input_pv_channel.
        It will be effective when enable_pv_merge_ is True.

        Examples:
            .. code-block:: python

                >>> # doctest: +SKIP('Depends on external files.')
                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
                >>> filelist = ["a.txt", "b.txt"]
                >>> dataset.set_filelist(filelist)
                >>> dataset.load_into_memory()
                >>> dataset.preprocess_instance()

        N)r   preprocess_instancer   s    r   r   z#InMemoryDataset.preprocess_instanceS  s    " 	((*****r   c                :    | j                             |           dS )a?  
        Set current phase in train. It is useful for unittest.
        current_phase : 1 for join, 0 for update.

        Examples:
            .. code-block:: python

                >>> # doctest: +SKIP('Depends on external files.')
                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
                >>> filelist = ["a.txt", "b.txt"]
                >>> dataset.set_filelist(filelist)
                >>> dataset.load_into_memory()
                >>> dataset.set_current_phase(1)

        N)r   set_current_phase)r   current_phases     r   r   z!InMemoryDataset.set_current_phasef  s     " 	&&}55555r   c                8    | j                                          dS )aq  
        Divide pv instance and convey it to input_channel.

        Examples:
            .. code-block:: python

                >>> # doctest: +SKIP('Depends on external files.')
                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
                >>> filelist = ["a.txt", "b.txt"]
                >>> dataset.set_filelist(filelist)
                >>> dataset.load_into_memory()
                >>> dataset.preprocess_instance()
                >>> exe.train_from_dataset(dataset)
                >>> dataset.postprocess_instance()

        N)r   postprocess_instancer   s    r   r   z$InMemoryDataset.postprocess_instancey  s    $ 	))+++++r   z=paddle.distributed.InMemoryDataset._set_fleet_send_batch_size   c                    || _         dS )a  
        Set fleet send batch size, default is 1024

        Args:
            fleet_send_batch_size(int): fleet send batch size

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
                >>> dataset.set_fleet_send_batch_size(800)

        N)r   )r   r   s     r   set_fleet_send_batch_sizez)InMemoryDataset.set_fleet_send_batch_size  s    & &;"""r   z@paddle.distributed.InMemoryDataset._set_fleet_send_sleep_secondsr   c                    || _         dS )a  
        Set fleet send sleep time, default is 0

        Args:
            fleet_send_sleep_seconds(int): fleet send sleep time

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
                >>> dataset.set_fleet_send_sleep_seconds(2)

        N)r   )r   r   s     r   set_fleet_send_sleep_secondsz,InMemoryDataset.set_fleet_send_sleep_seconds  s    & )A%%%r   z7paddle.distributed.InMemoryDataset._set_merge_by_lineidr   c                V    | j                             |           d| _        d| _        dS )a  
        Set merge by line id, instances of same line id will be merged after
        shuffle, you should parse line id in data generator.

        Args:
            merge_size(int): ins size to merge. default is 2.

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
                >>> dataset.set_merge_by_lineid()

        TN)r   set_merge_by_lineidr   r   )r   
merge_sizes     r   r   z#InMemoryDataset.set_merge_by_lineid  s0    ( 	((444# r   z@paddle.distributed.InMemoryDataset._set_generate_unique_feasignsc                V    | j                             |           || _        || _        d S rL   )r   set_generate_unique_feasignsgen_uni_feasignslocal_shard_num)r   generate_uni_feasigns	shard_nums      r   r   z,InMemoryDataset.set_generate_unique_feasigns  s1    
 	112GHHH 5(r   z@paddle.distributed.InMemoryDataset._generate_local_tables_unlockc                B    | j                             |||||           d S rL   )r   generate_local_tables_unlock)r   table_idfea_dimread_thread_numconsume_thread_numr   s         r   r   z,InMemoryDataset.generate_local_tables_unlock  s4     	11g0BI	
 	
 	
 	
 	
r   c                   t          |dd                   }t          |dd                   }t          |dd                   }| j        r1t          j                    r | j                            |||           dS dS dS )a  
        :api_attr: Static Graph

        Set training date for pull sparse parameters, saving and loading model. Only used in psgpu

        Args:
            date(str): training date(format : YYMMDD). eg.20211111

        Examples:
            .. code-block:: python

                >>> import paddle.base as base

                >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
                >>> dataset.set_date("20211111")
        N      )intr+   r	   rw   r,   set_dater   dateyearmonthdays        r   r   zInMemoryDataset.set_date  s    " 48}}D1I$qrr(mm? 	2t=?? 	2JeS11111	2 	2 	2 	2r   z3paddle.distributed.InMemoryDataset.load_into_memoryFc                   |                                   | j        s| j                                         dS t	          j                    r;| j                            | j                   | j                            |           dS dS )a  
        Load data into memory

         Args:
            is_shuffle(bool): whether to use local shuffle, default is False

        Examples:
            .. code-block:: python

                >>> # doctest: +SKIP('Depends on external files.')
                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
                >>> filelist = ["a.txt", "b.txt"]
                >>> dataset.set_filelist(filelist)
                >>> dataset.load_into_memory()
        N)ru   r+   r   load_into_memoryr	   rw   r,   set_dataset)r   
is_shuffles     r   r   z InMemoryDataset.load_into_memory  s    * 	 	4L))++++++-- 	4J""4<000J''
33333	4 	4r   z6paddle.distributed.InMemoryDataset.preload_into_memoryNc                    |                                   || j        }| j                            |           | j                                         | j                                         dS )a:  
        Load data into memory in async mode

        Args:
            thread_num(int): preload thread num

        Examples:
            .. code-block:: python

                >>> # doctest: +SKIP('Depends on external files.')
                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
                >>> filelist = ["a.txt", "b.txt"]
                >>> dataset.set_filelist(filelist)
                >>> dataset.preload_into_memory()
                >>> dataset.wait_preload_done()
        N)ru   r)   r   set_preload_thread_numcreate_preload_readerspreload_into_memoryrG   s     r   r   z#InMemoryDataset.preload_into_memory  sg    , 	J++J777++---((*****r   z4paddle.distributed.InMemoryDataset.wait_preload_donec                j    | j                                          | j                                          dS )a  
        Wait preload_into_memory done

        Examples:
            .. code-block:: python

                >>> # doctest: +SKIP('Depends on external files.')
                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
                >>> filelist = ["a.txt", "b.txt"]
                >>> dataset.set_filelist(filelist)
                >>> dataset.preload_into_memory()
                >>> dataset.wait_preload_done()
        N)r   wait_preload_donedestroy_preload_readersr   s    r   r   z!InMemoryDataset.wait_preload_done3  s2    & 	&&(((,,.....r   z0paddle.distributed.InMemoryDataset.local_shufflec                8    | j                                          dS )a  
        Local shuffle

        Examples:
            .. code-block:: python

                >>> # doctest: +SKIP('Depends on external files.')
                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
                >>> filelist = ["a.txt", "b.txt"]
                >>> dataset.set_filelist(filelist)
                >>> dataset.load_into_memory()
                >>> dataset.local_shuffle()
        N)r   local_shuffler   s    r   r   zInMemoryDataset.local_shuffleI  s    & 	""$$$$$r   z1paddle.distributed.InMemoryDataset.global_shuffle   c                
   |qt          |d          r$t          d           |                                 n|j                                         | j        dk    r|                                | _        | j        d| _        | j        d| _        | j        	                                 | j        
                    | j                   | j                            | j                   | j                            | j                   |>t          |d          r|                                 n|j                                         | j                            |           |>t          |d          r|                                 n|j                                         | j        r| j                                         |At          |d          r|                                 dS |j                                         dS dS )a  
        Global shuffle.
        Global shuffle can be used only in distributed mode. i.e. multiple
        processes on single machine or multiple machines training together.
        If you run in distributed mode, you should pass fleet instead of None.

        Examples:
            .. code-block:: python

                >>> # doctest: +SKIP('Depends on external files.')
                >>> import paddle.base as base
                >>> from paddle.incubate.distributed.fleet.parameter_server.pslib import fleet
                >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
                >>> filelist = ["a.txt", "b.txt"]
                >>> dataset.set_filelist(filelist)
                >>> dataset.load_into_memory()
                >>> dataset.global_shuffle(fleet)

        Args:
            fleet(Fleet): fleet singleton. Default None.
            thread_num(int): shuffle thread num. Default is 12.

        Nbarrier_workerzpscore fleetr   r   r   )hasattrprintr   _role_makerr   
worker_numr   r   r   "register_client2client_msg_handlerset_trainer_numr   r   global_shuffler   )r   fleetr)   s      r   r   zInMemoryDataset.global_shuffle^  s   8 u.// 3n%%%$$&&&&!002222%%#(#3#3#5#5 %-)-D&(0,-D)77999$$T%5666..t/IJJJ11$2OPPPu.// 3$$&&&&!00222##J///u.// 3$$&&&&!00222 	+L((***u.// 3$$&&&&&!0022222	 r   z1paddle.distributed.InMemoryDataset.release_memoryc                8    | j                                          dS )a  
        :api_attr: Static Graph

        Release InMemoryDataset memory data, when data will not be used again.

        Examples:
            .. code-block:: python

                >>> # doctest: +SKIP('Depends on external files.')
                >>> import paddle.base as base
                >>> from paddle.incubate.distributed.fleet.parameter_server.pslib import fleet
                >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
                >>> filelist = ["a.txt", "b.txt"]
                >>> dataset.set_filelist(filelist)
                >>> dataset.load_into_memory()
                >>> dataset.global_shuffle(fleet)
                >>> exe = base.Executor(base.CPUPlace())
                >>> exe.run(base.default_startup_program())
                >>> exe.train_from_dataset(base.default_main_program(), dataset)
                >>> dataset.release_memory()

        N)r   release_memoryr   s    r   r   zInMemoryDataset.release_memory  s    6 	##%%%%%r   c                4    | j                                         S )a  
        Get memory data size of Pv, user can call this function to know the pv num
        of ins in all workers after load into memory.

        Note:
            This function may cause bad performance, because it has barrier

        Returns:
            The size of memory pv data.

        Examples:
            .. code-block:: python

                >>> # doctest: +SKIP('Depends on external files.')
                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
                >>> filelist = ["a.txt", "b.txt"]
                >>> dataset.set_filelist(filelist)
                >>> dataset.load_into_memory()
                >>> print(dataset.get_pv_data_size())

        )r   get_pv_data_sizer   s    r   r   z InMemoryDataset.get_pv_data_size  s    . |,,...r   c                4    | j                                         S rL   )r   get_epoch_finishr   s    r   r   z InMemoryDataset.get_epoch_finish  s    |,,...r   c                8    | j                                          d S rL   )r   clear_sample_stater   s    r   r   z"InMemoryDataset.clear_sample_state  s    '')))))r   z7paddle.distributed.InMemoryDataset.get_memory_data_sizec                    ddl }| j                                        }|                    |g          }|(|dz  }|j                            ||           |d         S |d         S )a  
        Get memory data size, user can call this function to know the num
        of ins in all workers after load into memory.

        Note:
            This function may cause bad performance, because it has barrier

        Args:
            fleet(Fleet): Fleet Object.

        Returns:
            The size of memory data.

        Examples:
            .. code-block:: python

                >>> # doctest: +SKIP('Depends on external files.')
                >>> import paddle.base as base
                >>> from paddle.incubate.distributed.fleet.parameter_server.pslib import fleet
                >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
                >>> filelist = ["a.txt", "b.txt"]
                >>> dataset.set_filelist(filelist)
                >>> dataset.load_into_memory()
                >>> print(dataset.get_memory_data_size(fleet))

        r   N)numpyr   get_memory_data_sizearrayr   all_reduce_workerr   r   nplocal_data_sizeglobal_data_sizes        r   r   z$InMemoryDataset.get_memory_data_size  s    > 	,;;==((O#455.2//!1   $A&&q!!r   z8paddle.distributed.InMemoryDataset.get_shuffle_data_sizec                B   ddl }| j                                        }|                    |g          }t	          d|           |S|dz  }t          |d          r|j                            |          }n|j        	                    ||           |d         S |d         S )a  
        Get shuffle data size, user can call this function to know the num
        of ins in all workers after local/global shuffle.

        Note:
            This function may cause bad performance to local shuffle,
            because it has barrier. It does not affect global shuffle.

        Args:
            fleet(Fleet): Fleet Object.

        Returns:
            The size of shuffle data.

        Examples:
            .. code-block:: python

                >>> # doctest: +SKIP('Depends on external files.')
                >>> import paddle.base as base
                >>> from paddle.incubate.distributed.fleet.parameter_server.pslib import fleet
                >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
                >>> filelist = ["a.txt", "b.txt"]
                >>> dataset.set_filelist(filelist)
                >>> dataset.load_into_memory()
                >>> dataset.global_shuffle(fleet)
                >>> print(dataset.get_shuffle_data_size(fleet))

        r   Nz global shuffle local_data_size: util)
r   r   get_shuffle_data_sizer   r   r   r  
all_reducer   r   r   s        r   r  z%InMemoryDataset.get_shuffle_data_size  s    B 	,<<>>((O#4550/BBB.2uf%% #(:#8#8#I#I  !33#%5   $A&&q!!r   c                :    | j                             |           dS )zO
        Set heter ps mode
        user no need to call this function.
        N)r   set_heter_ps)r   enable_heter_pss     r   _set_heter_pszInMemoryDataset._set_heter_ps5  s     
 	!!/22222r   c                L   |                     dd          | j        j        _        |                     dd          | j        j        _        |                     dd          | j        j        _        |                     dd          | j        j        _        |                     d	d
          | j        j        _        |                     dd          | j        j        _        |                     dd          | j        j        _	        |                     dd          | j        j        _
        |                     dd          | j        j        _        |                     dd          | j        j        _        |                     dd          | j        j        _        |                     dd          | j        j        _        |                     dd          | j        j        _        |                     dd          | j        j        _        |                     dd          | j        j        _        |                     dd          | j        j        _        |                     dd          | j        j        _        |                     dd          | j        j        _        |                     dd          | j        j        _        |                     dd          | j        j        _        |                     dd          | j        j        _        | j                            d           d S )!a   
        Set graph config, user can set graph config in gpu graph mode.

        Args:
            config(dict): config dict.

        Returns:
            The size of shuffle data.

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> from paddle.incubate.distributed.fleet.parameter_server.pslib import fleet
                >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
                >>> graph_config = {"walk_len": 24,
                ...     "walk_degree": 10,
                ...     "once_sample_startid_len": 80000,
                ...     "sample_times_one_chunk": 5,
                ...     "window": 3,
                ...     "debug_mode": 0,
                ...     "batch_size": 800,
                ...     "meta_path": "cuid2clk-clk2cuid;cuid2conv-conv2cuid;clk2cuid-cuid2clk;clk2cuid-cuid2conv",
                ...     "gpu_graph_training": 1}
                >>> dataset.set_graph_config(graph_config)

        walk_degreer   walk_len   window   once_sample_startid_leni@  sample_times_one_chunk
   r@   
debug_moder   first_node_type 	meta_pathgpu_graph_trainingT	sage_modeFsamplestrain_table_capi 5 infer_table_capexcluded_train_pairinfer_node_type
get_degreeweighted_samplereturn_weight
pair_labelaccumulate_numN)getr&   graph_configr  r  r  r  r  r@   r  r  r  r  r  r  r  r  r  r  r   r!  r"  r#  r$  r   set_gpu_graph_mode)r   configs     r   set_graph_configz InMemoryDataset.set_graph_config<  s   8 4:::mQ3O3O$006

:r0J0J$-.4jj1.E.E$+?Ezz%t@
 @
$< ?Ejj$b?
 ?
$; 39**\12M2M$/28**\12M2M$/7=zzr8
 8
$4 28K1L1L$.:@** $;
 ;
$7 28K1O1O$./5zz)R/H/H$,7=zzv8
 8
$4 8>zzv8
 8
$4 <B::!2<
 <
$8 8>zzr8
 8
$4 39**%3
 3
$/ 8>zzu8
 8
$4 6<ZZU6
 6
$2 39**\22N2N$/6<jja7
 7
$3 	''-----r   c                H    || _         | j                            |           dS )a  
        Set pass id, user can set pass id in gpu graph mode.

        Args:
            pass_id: pass id.

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> pass_id = 0
                >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
                >>> dataset.set_pass_id(pass_id)
        N)r   r   set_pass_id)r   r   s     r   r+  zInMemoryDataset.set_pass_id  s'       )))))r   c                    | j         S )aa  
        Get pass id, user can set pass id in gpu graph mode.

        Returns:
            The pass id.

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset("InMemoryDataset")
                >>> pass_id = dataset.get_pass_id()
        )r   r   s    r   get_pass_idzInMemoryDataset.get_pass_id  s     |r     c                <    | j                             ||           dS )z 
        dump_walk_path
        N)r   dump_walk_path)r   path	dump_rates      r   r0  zInMemoryDataset.dump_walk_path  s"     	##D)44444r   c                :    | j                             |           dS )z'
        dump_sample_neighbors
        N)r   dump_sample_neighbors)r   r1  s     r   r4  z%InMemoryDataset.dump_sample_neighbors  s      	**400000r   )r   )r   )r   )FrL   )Nr   )r.  ),r   r   r   r    r   r   r   ru   r   r   r   r   r   r   r   r   r   r   r   r   r   r   r   r   r   r   r   r   r   r   r   r   r   r   r   r   r  r  r)  r+  r-  r0  r4  __classcell__r   s   @r   r   r   s  s         Zg)MNNN    ON" ZE  = =	 = ZF  & &	 &( ZS  < <	 < ZR  A A	 A ZE  # #	 #$ ZH  ) )	 )" ZI  + +	 +") ) )"' ' '" ZH  ) )	 )"/ / /"+ + +&6 6 6&, , ,( ZQ  ; ; ;	 ;" ZT  A A A	 A" ZK  ! ! !	 !( ZT  ) )	 )
 ZT  
 
	 
2 2 2. ZG  4 4 4	 40 ZJ  + + +	 +2 ZH  / /	 /$ ZD  % %	 %" ZE  93 93 93	 93v ZE  & &	 &2/ / /2/ / /* * * ZK  %" %" %"	 %"N ZL  +" +" +"	 +"Z3 3 3 3I. I. I.V* * *$   5 5 5 51 1 1 1 1 1 1r   r   c                  \     e Zd ZdZ fdZ edd          d             Zd Zd
d	Z xZ	S )r   z
    QueueDataset, it will process data streamly.

    Examples:
        .. code-block:: python

            >>> import paddle.base as base
            >>> dataset = base.DatasetFactory().create_dataset("QueueDataset")

    c                `    t                                                       d| j        _        dS )z`
        Initialize QueueDataset
        This class should be created by DatasetFactory
        MultiSlotDataFeedNr   r   r&   rV   r   s    r   r   zQueueDataset.__init__  s+    
 	2r   r   z/paddle.distributed.QueueDataset._prepare_to_runr   c                   | j         t          | j                  k    rt          | j                  | _         | j         dk    rd| _         | j                            | j                    | j                            | j                   | j                            |                                            | j                                         dS )zp
        Set data_feed_desc/thread num/filelist before run,
        user no need to call this function.
        r   r   N)	r)   rq   r*   r   rF   rJ   rr   rs   rt   r   s    r   ru   zQueueDataset._prepare_to_run  s     ?S////!$-00DO?aDO##DO444!!$-000''		444##%%%%%r   c                     t          d          )a  
        Local shuffle data.

        Local shuffle is not supported in QueueDataset
        NotImplementedError will be raised

        Examples:
            .. code-block:: python

                >>> # doctest: +SKIP('NotImplementedError will be raised.')
                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset("QueueDataset")
                >>> dataset.local_shuffle()

        Raises:
            NotImplementedError: QueueDataset does not support local shuffle

        zYQueueDataset does not support local shuffle, please use InMemoryDataset for local_shuffleNotImplementedErrorr   s    r   r   zQueueDataset.local_shuffle  s    & ";
 
 	
r   Nc                     t          d          )a}  
        Global shuffle data.

        Global shuffle is not supported in QueueDataset
        NotImplementedError will be raised

        Args:
            fleet(Fleet): fleet singleton. Default None.

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> from paddle.incubate.distributed.fleet.parameter_server.pslib import fleet
                >>> dataset = base.DatasetFactory().create_dataset("QueueDataset")
                >>> #dataset.global_shuffle(fleet)

        Raises:
            NotImplementedError: QueueDataset does not support global shuffle

        z[QueueDataset does not support global shuffle, please use InMemoryDataset for global_shuffler=  r   r   s     r   r   zQueueDataset.global_shuffle  s    , "<
 
 	
r   rL   )
r   r   r   r    r   r   ru   r   r   r5  r6  s   @r   r   r     s        	 	3 3 3 3 3 ZC  & &	 &
 
 
0
 
 
 
 
 
 
 
r   r   c                  0     e Zd ZdZ fdZd ZddZ xZS )FileInstantDatasetz
    FileInstantDataset, it will process data streamly.

    Examples:
        .. code-block:: python

            >>> import paddle.base as base
            >>> dataset = base.DatasetFactory.create_dataset("FileInstantDataset")
    c                `    t                                                       d| j        _        dS )zf
        Initialize FileInstantDataset
        This class should be created by DatasetFactory
        MultiSlotFileInstantDataFeedNr:  r   s    r   r   zFileInstantDataset.__init__  s+    
 	=r   c                     t          d          )zY
        Local shuffle
        FileInstantDataset does not support local shuffle
        z_FileInstantDataset does not support local shuffle, please use InMemoryDataset for local_shuffler=  r   s    r   r   z FileInstantDataset.local_shuffle#  s    
 ";
 
 	
r   Nc                     t          d          )z[
        Global shuffle
        FileInstantDataset does not support global shuffle
        zaFileInstantDataset does not support global shuffle, please use InMemoryDataset for global_shuffler=  r@  s     r   r   z!FileInstantDataset.global_shuffle-  s    
 "<
 
 	
r   rL   )r   r   r   r    r   r   r   r5  r6  s   @r   rB  rB    se         > > > > >
 
 

 
 
 
 
 
 
 
r   rB  c                  X     e Zd ZdZ fdZd Zd Zd Zd Zd Z	d Z
d	 Zd
 Zd Z xZS )BoxPSDatasetz
    BoxPSDataset: derived from InMemoryDataset.

    Examples:
        .. code-block:: python

            >>> import paddle.base as base
            >>> dataset = base.DatasetFactory().create_dataset("BoxPSDataset")
    c                    t                                                       t          j        | j                  | _        d| j        _        dS )z`
        Initialize BoxPSDataset
        This class should be created by DatasetFactory
        PaddleBoxDataFeedN)r   r   r	   BoxPSr   boxpsr&   rV   r   s    r   r   zBoxPSDataset.__init__C  s>    
 	Z--
2r   c                    t          |dd                   }t          |dd                   }t          |dd                   }| j                            |||           dS )z%
        Workaround for date
        Nr   r   )r   rL  r   r   s        r   r   zBoxPSDataset.set_dateL  s]     48}}D1I$qrr(mm
D%-----r   c                8    | j                                          dS )aH  
        Begin Pass
        Notify BoxPS to load sparse parameters of next pass to GPU Memory

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset("BoxPSDataset")
                >>> dataset.begin_pass()
        N)rL  
begin_passr   s    r   rO  zBoxPSDataset.begin_passU  s     	
r   c                :    | j                             |           dS )a*  
        End Pass
        Notify BoxPS that current pass ended
        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset("BoxPSDataset")
                >>> dataset.end_pass(True)
        N)rL  end_pass)r   need_save_deltas     r   rQ  zBoxPSDataset.end_passc  s      	
O,,,,,r   c                8    | j                                          dS )a  
        Wait async preload done
        Wait Until Feed Pass Done
        Examples:
            .. code-block:: python

                >>> # doctest: +SKIP('Depends on external files.')
                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset("BoxPSDataset")
                >>> filelist = ["a.txt", "b.txt"]
                >>> dataset.set_filelist(filelist)
                >>> dataset.preload_into_memory()
                >>> dataset.wait_preload_done()
        N)rL  wait_feed_pass_doner   s    r   r   zBoxPSDataset.wait_preload_donep  s     	
&&(((((r   c                `    |                                   | j                                         dS )a  
        Load next pass into memory and notify boxps to fetch its emb from SSD
        Examples:
            .. code-block:: python

                >>> # doctest: +SKIP('Depends on external files.')
                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset("BoxPSDataset")
                >>> filelist = ["a.txt", "b.txt"]
                >>> dataset.set_filelist(filelist)
                >>> dataset.load_into_memory()
        N)ru   rL  r   r   s    r   r   zBoxPSDataset.load_into_memory  s0     	
##%%%%%r   c                `    |                                   | j                                         dS )a  
        Begin async preload next pass while current pass may be training
        Examples:
            .. code-block:: python

                >>> # doctest: +SKIP('Depends on external files.')
                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset("BoxPSDataset")
                >>> filelist = ["a.txt", "b.txt"]
                >>> dataset.set_filelist(filelist)
                >>> dataset.preload_into_memory()
        N)ru   rL  r   r   s    r   r   z BoxPSDataset.preload_into_memory  s0     	
&&(((((r   c                ~    | j         s| j                            |d           | j                            |           d S )NT)r   r   r   r   rG   s     r   r   z)BoxPSDataset._dynamic_adjust_before_train  sB    ) 	FL33JEEE//
;;;;;r   c                    d S rL   r   r   s    r   r   z(BoxPSDataset._dynamic_adjust_after_train  r   r   c                X    t          |          }| j                            |           dS )a  
        Slots Shuffle
        Slots Shuffle is a shuffle method in slots level, which is usually used
        in sparse feature with large scale of instances. To compare the metric, i.e.
        auc while doing slots shuffle on one or several slots with baseline to
        evaluate the importance level of slots(features).

        Args:
            slots(list[string]): the set of slots(string) to do slots shuffle.

        Examples:
            .. code-block:: python

                >>> import paddle.base as base
                >>> dataset = base.DatasetFactory().create_dataset("BoxPSDataset")
                >>> dataset.set_merge_by_lineid()
                >>> #suppose there is a slot 0
                >>> dataset.slots_shuffle(['0'])
        N)r:   rL  r;   r<   s      r   r;   zBoxPSDataset.slots_shuffle  s+    ( JJ	
  +++++r   )r   r   r   r    r   r   rO  rQ  r   r   r   r   r   r;   r5  r6  s   @r   rH  rH  8  s         3 3 3 3 3. . .     - - -) ) )"& & & ) ) ) < < <
  , , , , , , ,r   rH  )r    
__future__r   google.protobufr   rW   paddle.base.protor   utilsr   r  r	   __all__r   r   r   r   rB  rH  r   r   r   <module>r_     s   I H " " " " " " ' ' ' ' ' '  + + + + + +            
$P $P $P $P $P $P $P $PNl l l l l l l l^	@1 @1 @1 @1 @1k @1 @1 @1FW
 W
 W
 W
 W
; W
 W
 W
t%
 %
 %
 %
 %
 %
 %
 %
PF, F, F, F, F,? F, F, F, F, F,r   