o $iP?@sddlZddlmZmZddlmZddlmZmZm Z ddl m Z ddl m Z ddlmZmZe Gdd d Ze Gd d d eZGd d d eZe GdddZe GdddZe GdddZe GdddZdS)N)ListOptional) DataIterator)MultiAgentBatch SampleBatchconcat_samples)unflatten_dict) DeveloperAPI) DeviceType 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)selfrrrrrrr\/home/ubuntu/veenaModal/venv/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 TrrrrrrrrNcshtj||||d||_||_||_||_dd|jD|_dd|jD|_ d|_ ||_ dS)z/Initializes a MiniBatchCyclicIterator instance.)rrrcSi|]}|dqSrr.0midrrr Oz4MiniBatchCyclicIterator.__init__..cSr r!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|Sr4r5r7rrrr9szLength of a sample must be > 0!r )r1minr/valuesr*r0r(r,itemsr5 ValueErrorr.r)_slice_seq_lens_in_Bgetrr6rappendr+shufflerr) r minibatch module_id module_batchsn_stepssamples_to_concatr9sample len_sampleerrr__iter__Vsf              z MiniBatchCyclicIterator.__iter__) rrrrrrrrrL __classcell__rrr2rr.s& rcs*eZdZdeffdd ZddZZS)MiniBatchDummyIteratorrc stj|fi|||_dSr4)r'rr()rrkwargsr2rrrs zMiniBatchDummyIterator.__init__ccs|jVdSr4)r()rrrrrLs zMiniBatchDummyIterator.__iter__)rrrrrrLrMrrr2rrNsrNc @s8eZdZdedededeefddZdefdd Z d S) MiniBatchRayDataIteratoriteratordevicer num_iterscKsL||_dd|D|_|jjd||d|j|_t|j|_||_dS)NcSsi|] \}}|dkr||qS) return_stater)r#kvrrrr%sz5MiniBatchRayDataIterator.__init__..) batch_sizerRr) _iteratorr=_kwargsiter_torch_batches_batched_iterableiter_epoch_iterator _num_iters)rrQrRrrSrOrrrrs   z!MiniBatchRayDataIterator.__init__rccsd}|jdus ||jkrU|jD]-}|d7}t|}tdd|Dtdd|Dd}|V|jr=||jkr=n qt|j|_|jsIdS|jdus ||jks dSdS)Nrr cSsi|] \}}|t|qSr)r)r#rD module_datarrrr%sz5MiniBatchRayDataIterator.__iter__..css$|] }ttt|VqdSr4)r5nextr\r<)r#r_rrr s  z4MiniBatchRayDataIterator.__iter__..) env_steps) r^r]rrr=sumr<r\r[)r iterationrrrrrLs,   z!MiniBatchRayDataIterator.__iter__N) rrrrr rrrrrLrrrrrPs rPc@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||_||_dSr4)r( _num_shards)rrrfrrrr zShardBatchIterator.__init__c cst|jD]<}i}|jjD]'\}}tt||j}||}t||t|}|t |t |||<qt |t |}|VqdSr4) rangerhr(r,r=mathceilr5r;rr) ri batch_to_sendpid sub_batchrWstartend new_batchrrrrL szShardBatchIterator.__iter__N)rrrrrrrrLrrrrres rec@sDeZdZdZ d deededeefddZdeefd d Z dS) ShardEpisodesIteratorzMIterator for sharding a list of Episodes into `num_shards` lists of Episodes.Nepisodesrflen_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|VqdSr4r:)r#rKrrrra2sz1ShardEpisodesIterator.__init__..cSg|]}dqSr!rr#_rrr 3z2ShardEpisodesIterator.__init__..N) sortedr5 _episodesrh_len_lookback_bufferrc _total_lengthrj_target_lengths)rrurfrvremaining_lengthrFlen_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|]}gqSrrrzrrrr|Br}z2ShardEpisodesIterator.__iter__..cSryr!rrzrrrr|Cr}rr N)rv) rjrhr5rindexr;rrAslicer) rsublistslengths episode_indexepisode min_indexr slice_partremaining_partsublistrrrrL:s:       zShardEpisodesIterator.__iter__r4) rrrrrr rrrrLrrrrrts rtc@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. rfcCrgr4) _object_refsrh)r object_refsrfrrrrvrizShardObjectRefIterator.__init__ccsdt|j}||j}||j}d}t|jD]}||||kr"dnd}|j||V|}qdS)Nrr )r5rrhrj)rn sublist_sizeremaining_elementsrqrmrrrrrrLzs   zShardObjectRefIterator.__iter__N)rrrrrrrLrrrrrjs r)rktypingrrray.datarray.rllib.policy.sample_batchrrrray.rllib.utilsrray.rllib.utils.annotationsr ray.rllib.utils.typingr r r rrNrPrertrrrrrs*   " : N