o `۷ik$@sdZddlZddlZddlmZddlmZmZddlZ ddl m Z ddl m Z ddlmZddlmZmZdd lmZmZd Zd ZGd d d Zede fddZdS)z [1] IMPACT: Importance Weighted Asynchronous Architectures with Clipped Target Networks. Luo et al. 2020 https://arxiv.org/pdf/1912.00167 N)deque)AnyOptional) ModelCatalog)ModelV2) OldAPIStack))DEFAULT_HISTOGRAM_BOUNDARIES_SHORT_EVENTSTimerAndPrometheusLogger)Counter Histogramfunc target_funcc @seZdZdZdedefddZ dded ed ee d efd d Z ded efddZ dd ed ee d efddZ d efddZ ed efddZd efddZd efddZddZd efddZdS)CircularBufferatA circular batch-wise buffer with Queue-like interface. The buffer holds at most N batches, which are sampled at random (uniformly). If full and a new batch is added, the oldest batch is discarded. Each batch can be sampled at most K times (after which it is also discarded). This version implements Queue-like put/get methods with blocking support. num_batchesiterations_per_batchcCs||_||_|j|j|_d|_tddt|jD|jd|_t|_|j|_ t |_ t d|_tj|_d|_d|_d|_tddtdd|_|jd |jjitd d dd |_|jd |jjitd dtdd|_|jd |jjidS)a Args: num_batches: N from the paper (queue buffer size). iterations_per_batch: K ("replay coefficient") from the paper. Defines how often a single batch can sampled before being discarded. If a new batch is added when the buffer is full, the oldest batch is discarded entirely (regardless of how often it has been sampled). rcSsg|]}dqSN).0_rrU/home/ubuntu/vllm_env/lib/python3.10/site-packages/ray/rllib/algorithms/appo/utils.py 3sz+CircularBuffer.__init__..)maxlen$rllib_utils_circular_buffer_put_timez"Time spent in CircularBuffer.put())rllib)name description boundariestag_keysr2rllib_utils_circular_buffer_put_ts_dropped_counterz8Total number of env steps dropped by the CircularBuffer.)rrr$rllib_utils_circular_buffer_get_timez"Time spent in CircularBuffer.get()N)rr_NxK _num_addedrrange_bufferset_indices_offset threadingLock_lock Semaphore_items_availablenprandom default_rng_rng _total_puts _total_gets_total_droppedr r!_metrics_circular_buffer_put_timeset_default_tags __class____name__r '_metrics_circular_buffer_put_ts_dropped!_metrics_circular_buffer_get_time)selfrrrrr__init__$sN        zCircularBuffer.__init__TNitemblocktimeoutreturnc Cs"t|j|jf|jd7_|jd}t|jD]%}|j||j |j |j |j |j |j d7_ |j q|jd7_d}|durht|trY|dn|}|dkrh|jj|dWdn1srwYWd|SWd|S1swY|S)aAdd a new batch to the buffer. The batch is added K times (iterations_per_batch) to allow for K samples. If full, the oldest batch entries are dropped. Args: item: The batch to add block: Not used (always non-blocking for puts) timeout: Not used Returns: Number of dropped entries (0 or iterations_per_batch) rN)value)r r3r)r0r#r"rappendr%addr&discardr r+releaser! isinstancetuple env_stepsr7inc)r9r;r<r= dropped_entryr dropped_tsrrrput`s<         zCircularBuffer.putcCs|j|ddS)z$Equivalent to self.put(block=False).Fr<)rK)r9r;rrr put_nowaitszCircularBuffer.put_nowaitc Cst|jkt|dkrtdt|dks |j>|jt|j }||j |j }|j |}|dusFJ|||j |j dd|j Dfd|j |<|j |Wdn1s[wYWd|SWd|S1sswY|S)aSample a random batch from the buffer. The sampled entry is removed and won't be sampled again. Blocks if the buffer is empty (when block=True). Args: block: If True, block until an item is available timeout: Maximum time to wait (only used when block=True) Returns: A randomly sampled batch Raises: TimeoutError: If timeout expires while blocking IndexError: If buffer is empty and block=False rg-C6?NcSsg|]}|duqSrr)rbrrrrsz&CircularBuffer.get..)r r8lentimesleepr)r/choicelistr%r&r r#rC)r9r<r=idxactual_buffer_idxbatchrrrgets2         zCircularBuffer.getcCs |jddS)z$Equivalent to self.get(block=False).FrL)rWr9rrr get_nowaits zCircularBuffer.get_nowaitcCs6|j|j|jkWdS1swYdS)zIWhether the buffer has been filled once with at least `self.num_batches`.N)r)r!rrXrrrfilleds $zCircularBuffer.filledcCs4|j t|jWdS1swYdS)zIReturns the number of actually valid (non-expired) batches in the buffer.N)r)rOr%rXrrrqsizes$zCircularBuffer.qsizecCs|Sr)r[rXrrr__len__szCircularBuffer.__len__cCsdS)zNo-op for Queue compatibility.NrrXrrr task_doneszCircularBuffer.task_donec CsZ|j t|j|j|j|j|j|j|j|j |jkdWdS1s&wYdS)z%Get buffer statistics for monitoring.)sizecapacityrr total_puts total_gets total_droppedrZN) r)rOr%r rrr0r1r2r!rXrrr get_statss $zCircularBuffer.get_stats)TN)r6 __module__ __qualname____doc__intr:rboolrfloatrKrMrWrYpropertyrZr[r\r]dictrcrrrrrs, = 2(rr>cCs~t|j|jd\}}tj|j|j||jdt|jd|_|j |_ tj|j|j||jdt |jd|_ |j |_ |jS)zBuilds model and target model for APPO. Returns: ModelV2: The Model for the Policy to use. Note: The target model will not be returned, just assigned to `policy.target_model`. model)r framework)rget_action_dist action_spaceconfig get_model_v2observation_space POLICY_SCOPErmrl variablesmodel_variablesTARGET_POLICY_SCOPE target_modeltarget_model_variables)policyr logit_dimrrrmake_appo_modelss,   r{)rfr'rP collectionsrtypingrrnumpyr,ray.rllib.models.catalogrray.rllib.models.modelv2rray.rllib.utils.annotationsr#ray.rllib.utils.metrics.ray_metricsrr ray.util.metricsr r rsrvrr{rrrrs"    J