o ci(>@sddlZddlmZmZmZddlmZddlmZm Z ddlm Z ddl m Z ddl mZe Gdd d Ze Gd d d eZGd d d eZe GdddZe GdddZe GdddZe GdddZdS)N)CallableListOptional) DataIterator)MultiAgentBatchconcat_samples) SampleBatch) DeveloperAPI) EpisodeTypec @s<eZdZdZdddddededed ed ed d f d dZd S)MiniBatchIteratorBasez+The base class for all minibatch iterators.Tr num_epochsshuffle_batch_per_epochnum_total_minibatchesbatchrrminibatch_sizerreturnNcCsdS)aInitializes a MiniBatchIteratorBase instance. Args: batch: The input multi-agent batch. num_epochs: The number of complete passes over the entire train batch. Each pass might be further split into n minibatches (if `minibatch_size` provided). The train batch is generated from the given `episodes` through the Learner connector pipeline. minibatch_size: The size of minibatches to use to further split the train batch into per epoch. The train batch is generated from the given `episodes` through the Learner connector pipeline. num_total_minibatches: The total number of minibatches to loop through (over all `num_epochs` epochs). It's only required to set this to != 0 in multi-agent + multi-GPU situations, in which the MultiAgentEpisodes themselves are roughly sharded equally, however, they might contain SingleAgentEpisodes with very lopsided length distributions. Thus, without this fixed, pre-computed value, one Learner might go through a different number of minibatche passes than others causing a deadlock. N)selfrrrrrrrS/home/ubuntu/.local/lib/python3.10/site-packages/ray/rllib/utils/minibatch_utils.py__init__szMiniBatchIteratorBase.__init__)__name__ __module__ __qualname____doc__rintboolrrrrrr s$r c sLeZdZdZdddddededed ed ed d f fd dZddZZ S)MiniBatchCyclicIteratorawThis implements a simple multi-agent minibatch iterator. This iterator will split the input multi-agent batch into minibatches where the size of batch for each module_id (aka policy_id) is equal to minibatch_size. If the input batch is smaller than minibatch_size, then the iterator will cycle through the batch until it has covered `num_epochs` epochs. r Trr rrrrrrNcshtj||||d||_||_||_||_dd|jD|_dd|jD|_ d|_ ||_ dS)z/Initializes a MiniBatchCyclicIterator instance.)rrrcSi|]}|dqSrr.0midrrr Oz4MiniBatchCyclicIterator.__init__..cSrr rr!rrrr$Qr%rN) superr_batch_minibatch_size _num_epochs_shuffle_batch_per_epochpolicy_batcheskeys_start_num_covered_epochs_minibatch_count_num_total_minibatches)rrrrrr __class__rrr8s  z MiniBatchCyclicIterator.__init__c cs|jdkrt|j|jks|jdkr|j|jkri}|jjD]\}}t |dkr5t d|d|j |}|j }g}|j r`|tjdusNJddd}t|||j t |}ndd}||||kr||d}||||} | dksJd|| 8}d}|j|d 7<|jr|||||ksl||} | |kr|||| t|||<| |j |<q#t|t |j}|V|jd 7_|jdkrt|j|jks|jdkr|j|jksdSdSdSdS) NrzThe batch for module_id zt is empty! This will create an infinite loop because we need to cover the same number of samples for each module_id.z}MiniBatchCyclicIterator requires SampleBatch.SEQ_LENSto be present in the batch for slicing a batch in the batch dimension B.cSst|tjSN)lenrSEQ_LENSbrrrget_lensz1MiniBatchCyclicIterator.__iter__..get_lencSst|Sr3r4r6rrrr8szLength of a sample must be > 0!r )r0minr.valuesr)r/r'r+itemsr4 ValueErrorr-r(_slice_seq_lens_in_Bgetrr5rappendr*shufflerr) r minibatch module_id module_batchsn_stepssamples_to_concatr8sample len_sampleerrr__iter__Vsf              z MiniBatchCyclicIterator.__iter__) rrrrrrrrrK __classcell__rrr1rr.s& rcs*eZdZdeffdd ZddZZS)MiniBatchDummyIteratorrc stj|fi|||_dSr3)r&rr')rrkwargsr1rrrs zMiniBatchDummyIterator.__init__ccs|jVdSr3)r')rrrrrKs zMiniBatchDummyIterator.__iter__)rrrrrrKrLrrr1rrMsrMc @s<eZdZdededededeef ddZdefd d Z d S) MiniBatchRayDataIteratoriterator collate_fn finalize_fnr num_iterscKs^||_||_||_dd|D|_|jjd||j|jd|j|_t|j|_||_ dS)NcSsi|] \}}|dkr||qS) return_stater)r"kvrrrr$sz5MiniBatchRayDataIterator.__init__..) batch_size _collate_fn _finalize_fnr) _iteratorrXrYr<_kwargs _iter_batches_batched_iterableiter_epoch_iterator _num_iters)rrPrQrRrrSrNrrrrs   z!MiniBatchRayDataIterator.__init__rccszd}|jdus ||jkr;|jD]}|d7}|V|jr#||jkr#n qt|j|_|js/dS|jdus ||jks dSdSNrr )r`r_r^r])r iterationrrrrrKs  z!MiniBatchRayDataIterator.__iter__N) rrrrrrrrrrKrrrrrOs rOc@s*eZdZdZdedefddZddZdS) ShardBatchIteratorzIterator for sharding batch into num_shards batches. Args: batch: The input multi-agent batch. num_shards: The number of shards to split the batch into. Yields: A MultiAgentBatch of size len(batch) / num_shards. r num_shardscC||_||_dSr3)r' _num_shards)rrrdrrrr zShardBatchIterator.__init__c cst|jD]<}i}|jjD]'\}}tt||j}||}t||t|}|t |t |||<qt |t |}|VqdSr3) rangerfr'r+r<mathceilr4r:rr) ri batch_to_sendpid sub_batchrWstartend new_batchrrrrKszShardBatchIterator.__iter__N)rrrrrrrrKrrrrrcs rcc@sDeZdZdZ d deededeefddZdeefd d Z dS) ShardEpisodesIteratorzMIterator for sharding a list of Episodes into `num_shards` lists of Episodes.Nepisodesrdlen_lookback_buffercCs~t|tdd|_||_||_tdd|D|_ddt|jD|_|j}t|jD]}|||}||j|<||8}q+dS)aInitializes a ShardEpisodesIterator instance. Args: episodes: The input list of Episodes. num_shards: The number of shards to split the episodes into. len_lookback_buffer: An optional length of a lookback buffer to enforce on the returned shards. When spitting an episode, the second piece might need a lookback buffer (into the first piece) depending on the user's settings. T)keyreversecss|]}t|VqdSr3r9)r"rJrrr *sz1ShardEpisodesIterator.__init__..cSg|]}dqSr rr"_rrr +z2ShardEpisodesIterator.__init__..N) sortedr4 _episodesrf_len_lookback_buffersum _total_lengthrh_target_lengths)rrsrdrtremaining_lengthrElen_rrrrs   zShardEpisodesIterator.__init__rc csJddt|jD}ddt|jD}d}|t|jkr|j|}|t|}||t||j|krM|||||t|7<|d7}nG|j|||}|dkr|d||jt|d|j d}}|||||t|7<||j|<n|dksJ||||d7}|t|jks|D]} | VqdS)a2Runs one iteration through this sharder. Yields: A sub-list of Episodes of size roughly `len(episodes) / num_shards`. The yielded sublists might have slightly different total sums of episode lengths, in order to not have to drop even a single timestep. cSsg|]}gqSrrryrrrr{:r|z2ShardEpisodesIterator.__iter__..cSrxr rryrrrr{;r|rr N)rt) rhrfr4r~indexr:rr@slicer) rsublistslengths episode_indexepisode min_indexr slice_partremaining_partsublistrrrrK2s:       zShardEpisodesIterator.__iter__r3) rrrrrr rrrrKrrrrrrs rrc@s&eZdZdZdefddZddZdS)ShardObjectRefIteratora,Iterator for sharding a list of ray ObjectRefs into num_shards sub-lists. Args: object_refs: The input list of ray ObjectRefs. num_shards: The number of shards to split the references into. Yields: A sub-list of ray ObjectRefs with lengths as equal as possible. rdcCrer3) _object_refsrf)r object_refsrdrrrrnrgzShardObjectRefIterator.__init__ccsdt|j}||j}||j}d}t|jD]}||||kr"dnd}|j||V|}qdSra)r4rrfrh)rn sublist_sizeremaining_elementsrorkrprrrrKrs   zShardObjectRefIterator.__iter__N)rrrrrrrKrrrrrbs r)ritypingrrrray.datarray.rllib.policy.sample_batchrrrray.rllib.utils.annotationsr ray.rllib.utils.typingr r rrMrOrcrrrrrrrs*    " 2 N