import copy import heapq import threading import time import uuid from collections import defaultdict, deque from typing import Any, Dict, List, Optional, Tuple, Union import numpy as np from ray._common.deprecation import Deprecated from ray.rllib.utils import force_list from ray.rllib.utils.framework import try_import_torch from ray.rllib.utils.numpy import convert_to_numpy from ray.util.annotations import DeveloperAPI torch, _ = try_import_torch() @Deprecated(new="rllib.utils.metrics.stats", error=False) @DeveloperAPI class Stats: """A container class holding a number of values and executing reductions over them. The individual values in a Stats object may be of any type, for example python int or float, numpy arrays, or more complex structured (tuple, dict) and are stored in a list under `self.values`. This class is not meant to be interfaced with directly from application code. Instead, use `MetricsLogger` to log and manipulate Stats. Stats can be used to store metrics of the same type over time, for example a loss or a learning rate, and to reduce all stored values applying a certain reduction mechanism (for example "mean" or "sum"). Available reduction mechanisms are: - "mean" using EMA with a configurable EMA coefficient. - "mean" using a sliding window (over the last n stored values). - "max/min" with an optional sliding window (over the last n stored values). - "sum" with an optional sliding window (over the last n stored values). - None: Simply store all logged values to an ever-growing list. Through the `reduce()` API, one of the above-mentioned reduction mechanisms will be executed on `self.values`. """ def __init__( self, init_values: Optional[Any] = None, reduce: Optional[str] = "mean", percentiles: Union[List[int], bool] = False, reduce_per_index_on_aggregate: bool = False, window: Optional[Union[int, float]] = None, ema_coeff: Optional[float] = None, clear_on_reduce: bool = False, throughput: Union[bool, float] = False, throughput_ema_coeff: Optional[float] = None, ): """Initializes a Stats instance. Args: init_values: Optional initial values to be placed into `self.values`. If None, `self.values` will start empty. If percentiles is True, values must be ordered if provided. reduce: The name of the reduce method to be used. Allowed are "mean", "min", "max", and "sum". Use None to apply no reduction method (leave `self.values` as-is when reducing, except for shortening it to `window`). Note that if both `reduce` and `window` are None, the user of this Stats object needs to apply some caution over the values list not growing infinitely. percentiles: If reduce is `None`, we can compute the percentiles of the values list given by `percentiles`. Defaults to [0, 50, 75, 90, 95, 99, 100] if set to True. When using percentiles, a window must be provided. This window should be chosen carfully. RLlib computes exact percentiles and the computational complexity is O(m*n*log(n/m)) where n is the window size and m is the number of parallel metrics loggers invovled (for example, m EnvRunners). To be safe, choose a window < 1M and less than 1000 Stats objects to aggregate. See #52963 for more details. window: An optional window size to reduce over. If `window` is not None, then the reduction operation is only applied to the most recent `windows` items, and - after reduction - the values list is shortened to hold at most `window` items (the most recent ones). Must be None if `ema_coeff` is not None. If `window` is None (and `ema_coeff` is None), reduction must not be "mean". reduce_per_index_on_aggregate: If True, when merging Stats objects, we reduce incoming values per index such that the new value at index `n` will be the reduced value of all incoming values at index `n`. If False, when reducing `n` Stats, the first `n` merged values will be the reduced value of all incoming values at index `0`, the next `n` merged values will be the reduced values of all incoming values at index `1`, etc. ema_coeff: An optional EMA coefficient to use if reduce is "mean" and no `window` is provided. Note that if both `window` and `ema_coeff` are provided, an error is thrown. Also, if `ema_coeff` is provided, `reduce` must be "mean". The reduction formula for EMA performed by Stats is: EMA(t1) = (1.0 - ema_coeff) * EMA(t0) + ema_coeff * new_value clear_on_reduce: If True, the Stats object will reset its entire values list to an empty one after `self.reduce()` is called. However, it will then return from the `self.reduce()` call a new Stats object with the properly reduced (not completely emptied) new values. Setting this to True is useful for cases, in which the internal values list would otherwise grow indefinitely, for example if reduce is None and there is no `window` provided. throughput: If True, track a throughput estimate together with this Stats. This is only supported for `reduce=sum` and `clear_on_reduce=False` metrics (aka. "lifetime counts"). The `Stats` then keeps track of the time passed between two consecutive calls to `reduce()` and update its throughput estimate. The current throughput estimate can be obtained through: `throughput_per_sec = Stats.peek(throughput=True)`. If a float, track throughput and also set current throughput estimate to the given value. throughput_ema_coeff: An optional EMA coefficient to use for throughput tracking. Only used if throughput=True. """ # Thus far, we only support mean, max, min, and sum. if reduce not in [None, "mean", "min", "max", "sum", "percentiles"]: raise ValueError( "`reduce` must be one of `mean|min|max|sum|percentiles` or None!" ) # One or both window and ema_coeff must be None. if window is not None and ema_coeff is not None: raise ValueError("Only one of `window` or `ema_coeff` can be specified!") # If `ema_coeff` is provided, `reduce` must be "mean". if ema_coeff is not None and reduce != "mean": raise ValueError( "`ema_coeff` arg only allowed (not None) when `reduce=mean`!" ) if percentiles is not False: if reduce is not None: raise ValueError( "`reduce` must be `None` when `percentiles` is not `False`!" ) if window in (None, float("inf")): raise ValueError( "A window must be specified when reduce is 'percentiles'!" ) if reduce_per_index_on_aggregate is not False: raise ValueError( f"`reduce_per_index_on_aggregate` ({reduce_per_index_on_aggregate})" f" must be `False` when `percentiles` is not `False`!" ) if percentiles is True: percentiles = [0, 50, 75, 90, 95, 99, 100] else: if type(percentiles) not in (bool, list): raise ValueError("`percentiles` must be a list or bool!") if isinstance(percentiles, list): if not all(isinstance(p, (int, float)) for p in percentiles): raise ValueError( "`percentiles` must contain only ints or floats!" ) if not all(0 <= p <= 100 for p in percentiles): raise ValueError( "`percentiles` must contain only values between 0 and 100!" ) self._percentiles = percentiles # If `window` is explicitly set to inf, `clear_on_reduce` must be True. self._inf_window = window in [None, float("inf")] # If `window` is set to inf, `clear_on_reduce` must be True. # Otherwise, we risk a memory leak. if self._inf_window and not clear_on_reduce and reduce is None: raise ValueError( "When using an infinite window without reduction, `clear_on_reduce` must " "be set to True!" ) # If reduce=mean AND window=ema_coeff=None, we use EMA by default with a coeff # of 0.01 (we do NOT support infinite window sizes for mean as that would mean # to keep data in the cache forever). if reduce == "mean" and self._inf_window and ema_coeff is None: ema_coeff = 0.01 self._reduce_method = reduce self._window = window self._ema_coeff = ema_coeff if ( self._reduce_method not in ["mean", "sum", "min", "max"] and reduce_per_index_on_aggregate ): raise ValueError( "reduce_per_index_on_aggregate is only supported for mean, sum, min, and max reduction!" ) self._reduce_per_index_on_aggregate = reduce_per_index_on_aggregate # Timing functionality (keep start times per thread). self._start_times = defaultdict(lambda: None) # Simply store ths flag for the user of this class. self._clear_on_reduce = clear_on_reduce self._has_returned_zero = False # On each `.reduce()` call, we store the result of this call in # self._last_reduce. self._last_reduced = [np.nan] # The ID of this Stats instance. self.id_ = str(uuid.uuid4()) self._prev_merge_values = defaultdict(int) self._throughput_ema_coeff = throughput_ema_coeff self._throughput_stats = None if throughput is not False: self._throughput_stats = Stats( # We have to check for bool here because in Python, bool is a subclass # of int. init_values=[throughput] if ( isinstance(throughput, (int, float)) and not isinstance(throughput, bool) ) else None, reduce="mean", ema_coeff=throughput_ema_coeff, window=None, clear_on_reduce=False, throughput=False, throughput_ema_coeff=None, ) if init_values is not None: self._last_throughput_measure_time = time.perf_counter() else: self._last_throughput_measure_time = ( -1 ) # Track last push time for throughput calculation # The actual, underlying data in this Stats object. self.values: Union[List, deque.Deque] = None self._set_values(force_list(init_values)) self._is_tensor = False # Track if new values were pushed since last reduce if init_values is not None: self._has_new_values = True else: self._has_new_values = False def check_value(self, value: Any) -> None: # If we have a reduce method, value should always be a scalar # If we don't reduce, we can keep track of value as it is if self._reduce_method is not None: if isinstance(value, np.ndarray) and value.shape == (): return elif torch and torch.is_tensor(value): self._is_tensor = True if tuple(value.shape) == (): return elif type(value) not in (list, tuple, deque): return raise ValueError( f"Value ({value}) is required to be a scalar when using a reduce " "method!" ) def push(self, value: Any) -> None: """Pushes a value into this Stats object. Args: value: The value to be pushed. Can be of any type. """ self.check_value(value) # If throughput tracking is enabled, calculate it based on time between pushes if self.has_throughput: self._recompute_throughput(value) # Handle different reduction methods if self._window is not None: # For windowed operations, append to values and trim if needed self.values.append(value) if len(self.values) > self._window: self.values.popleft() else: # For non-windowed operations, use _reduced_values if len(self.values) == 0: self._set_values([value]) else: self.values.append(value) _, values = self._reduced_values() self._set_values(values) # Mark that we have new values self._has_new_values = True def __enter__(self) -> "Stats": """Called when entering a context (with which users can measure a time delta). Returns: This Stats instance (self), unless another thread has already entered (and not exited yet), in which case a copy of `self` is returned. This way, the second thread(s) cannot mess with the original Stat's (self) time-measuring. This also means that only the first thread to __enter__ actually logs into `self` and the following threads' measurements are discarded (logged into a non-referenced shim-Stats object, which will simply be garbage collected). """ # In case another thread already is measuring this Stats (timing), simply ignore # the "enter request" and return a clone of `self`. thread_id = threading.get_ident() self._start_times[thread_id] = time.perf_counter() return self def __exit__(self, exc_type, exc_value, tb) -> None: """Called when exiting a context (with which users can measure a time delta).""" thread_id = threading.get_ident() assert self._start_times[thread_id] is not None time_delta_s = time.perf_counter() - self._start_times[thread_id] self.push(time_delta_s) del self._start_times[thread_id] def peek(self, compile: bool = True) -> Union[Any, List[Any]]: """Returns the result of reducing the internal values list. Note that this method does NOT alter the internal values list in this process. Thus, users can call this method to get an accurate look at the reduced value(s) given the current internal values list. Args: compile: If True, the result is compiled into a single value if possible. Returns: The result of reducing the internal values list. """ if self._has_new_values or (not compile and not self._inf_window): reduced_value, reduced_values = self._reduced_values() if not compile and not self._inf_window: return reduced_values if compile and self._reduce_method: return reduced_value[0] if compile and self._percentiles is not False: return compute_percentiles(reduced_values, self._percentiles) return reduced_value else: return_value = self._last_reduced if compile: # We don't need to check for self._reduce_method or percentiles here # because we only store the reduced value if there is a reduce method. return_value = return_value[0] return return_value @property def throughput(self) -> float: """Returns the current throughput estimate per second. Raises: ValueError: If throughput tracking is not enabled for this Stats object. Returns: The current throughput estimate per second. """ if not self.has_throughput: raise ValueError("Throughput tracking is not enabled for this Stats object") # We can always return the first value here because throughput is a single value return self._throughput_stats.peek() @property def has_throughput(self) -> bool: """Returns whether this Stats object tracks throughput. Returns: True if this Stats object has throughput tracking enabled, False otherwise. """ return self._throughput_stats is not None def reduce(self, compile: bool = True) -> Union[Any, List[Any]]: """Reduces the internal values list according to the constructor settings. Thereby, the internal values list is changed (note that this is different from `peek()`, where the internal list is NOT changed). See the docstring of this class for details on the reduction logic applied to the values list, based on the constructor settings, such as `window`, `reduce`, etc.. Args: compile: If True, the result is compiled into a single value if possible. If it is not possible, the result is a list of values. If False, the result is a list of one or more values. Returns: The reduced value (can be of any type, depending on the input values and reduction method). """ len_before_reduce = len(self) if self._has_new_values: # Only calculate and update history if there were new values pushed since # last reduce reduced, reduced_internal_values_list = self._reduced_values() # `clear_on_reduce` -> Clear the values list. if self._clear_on_reduce: self._set_values([]) else: self._set_values(reduced_internal_values_list) else: reduced_internal_values_list = None reduced = self._last_reduced reduced = self._numpy_if_necessary(reduced) # Shift historic reduced valued by one in our reduce_history. if self._reduce_method is not None: # It only makes sense to extend the history if we are reducing to a single # value. We need to make a copy here because the new_values_list is a # reference to the internal values list self._last_reduced = force_list(reduced.copy()) else: # If there is a window and no reduce method, we don't want to use the reduce # history to return reduced values in other methods self._has_new_values = True if compile and self._reduce_method is not None: assert ( len(reduced) == 1 ), f"Reduced values list must contain exactly one value, found {reduced}" reduced = reduced[0] if not compile and not self._inf_window: if reduced_internal_values_list is None: _, reduced_internal_values_list = self._reduced_values() return_values = self._numpy_if_necessary( reduced_internal_values_list ).copy() elif compile and self._percentiles is not False: if reduced_internal_values_list is None: _, reduced_internal_values_list = self._reduced_values() return_values = compute_percentiles( reduced_internal_values_list, self._percentiles ) else: return_values = reduced if compile: return return_values else: if len_before_reduce == 0: # return_values will be be 0 if we reduce a sum over zero elements # But we don't want to create such a zero out of nothing for our new # Stats object that we return here return Stats.similar_to(self) return Stats.similar_to(self, init_values=return_values) def merge_on_time_axis(self, other: "Stats") -> None: """Merges another Stats object's values into this one along the time axis. Args: other: The other Stats object to merge values from. """ self.values.extend(other.values) # Mark that we have new values since we modified the values list self._has_new_values = True def merge_in_parallel(self, *others: "Stats") -> None: """Merges all internal values of `others` into `self`'s internal values list. Thereby, the newly incoming values of `others` are treated equally with respect to each other as well as with respect to the internal values of self. Use this method to merge other `Stats` objects, which resulted from some parallelly executed components, into this one. For example: n Learner workers all returning a loss value in the form of `{"total_loss": [some value]}`. The following examples demonstrate the parallel merging logic for different reduce- and window settings: Args: others: One or more other Stats objects that need to be parallely merged into `self, meaning with equal weighting as the existing values in `self`. """ win = self._window or float("inf") # If any of the value lists have a length of 0 or if there is only one value and # it is nan, we skip stats_to_merge = [ s for s in [self, *others] if not ( len(s) == 0 or ( len(s) == 1 and np.all(np.isnan(self._numpy_if_necessary(s.values))) ) ) ] # If there is only one stat to merge, and it is the same as self, return. if len(stats_to_merge) == 0: # If none of the stats have values, return. return elif len(stats_to_merge) == 1: if stats_to_merge[0] == self: # If no incoming stats have values, return. return else: # If there is only one stat with values, and it's incoming, copy its # values. self.values = stats_to_merge[0].values return # Take turns stepping through `self` and `*others` values, thereby moving # backwards from last index to beginning and will up the resulting values list. # Stop as soon as we reach the window size. new_values = [] tmp_values = [] if self._percentiles is not False: # Use heapq to sort values (assumes that the values are already sorted) # and then pick the correct percentiles lists_to_merge = [list(self.values), *[list(o.values) for o in others]] merged = list(heapq.merge(*lists_to_merge)) self._set_values(merged) else: # Loop from index=-1 backward to index=start until our new_values list has # at least a len of `win`. for i in range(1, max(map(len, stats_to_merge)) + 1): # Per index, loop through all involved stats, including `self` and add # to `tmp_values`. for stats in stats_to_merge: if len(stats) < i: continue tmp_values.append(stats.values[-i]) # Now reduce across `tmp_values` based on the reduce-settings of this # Stats. if self._reduce_per_index_on_aggregate: n_values = 1 else: n_values = len(tmp_values) if self._ema_coeff is not None: new_values.extend([np.nanmean(tmp_values)] * n_values) elif self._reduce_method is None: new_values.extend(tmp_values) elif self._reduce_method == "sum": # We add [sum(tmp_values) / n_values] * n_values to the new values # list instead of tmp_values, because every incoming element should # have the same weight. added_sum = self._reduced_values(values=tmp_values)[0][0] new_values.extend([added_sum / n_values] * n_values) if self.has_throughput: self._recompute_throughput(added_sum) else: new_values.extend( self._reduced_values(values=tmp_values)[0] * n_values ) tmp_values.clear() if len(new_values) >= win: new_values = new_values[:win] break self._set_values(list(reversed(new_values))) # Mark that we have new values since we modified the values list self._has_new_values = True def clear_throughput(self) -> None: """Clears the throughput Stats, if applicable and `self` has throughput. Also resets `self._last_throughput_measure_time` to -1 such that the Stats object has to create a new timestamp first, before measuring any new throughput values. """ if self.has_throughput: self._throughput_stats._set_values([]) self._last_throughput_measure_time = -1 def _recompute_throughput(self, value) -> None: """Recomputes the current throughput value of this Stats instance.""" # Make sure this Stats object does measure throughput. assert self.has_throughput # Take the current time stamp. current_time = time.perf_counter() # Check, whether we have a previous timestamp (non -1). if self._last_throughput_measure_time >= 0: # Compute the time delta. time_diff = current_time - self._last_throughput_measure_time # Avoid divisions by zero. if time_diff > 0: # Push new throughput value into our throughput stats object. self._throughput_stats.push(value / time_diff) # Update the time stamp of the most recent throughput computation (this one). self._last_throughput_measure_time = current_time @staticmethod def _numpy_if_necessary(values): # Torch tensor handling. Convert to CPU/numpy first. if torch and len(values) > 0 and torch.is_tensor(values[0]): # Convert all tensors to numpy values. values = [v.cpu().numpy() for v in values] return values def __len__(self) -> int: """Returns the length of the internal values list.""" return len(self.values) def __repr__(self) -> str: win_or_ema = ( f"; win={self._window}" if self._window else f"; ema={self._ema_coeff}" if self._ema_coeff else "" ) return ( f"Stats({self.peek()}; len={len(self)}; " f"reduce={self._reduce_method}{win_or_ema})" ) def __int__(self): if self._reduce_method is None: raise ValueError( "Cannot convert Stats object with reduce method `None` to int because " "it can not be reduced to a single value." ) else: return int(self.peek()) def __float__(self): if self._reduce_method is None: raise ValueError( "Cannot convert Stats object with reduce method `None` to float " "because it can not be reduced to a single value." ) else: return float(self.peek()) def __eq__(self, other): if self._reduce_method is None: self._comp_error("__eq__") else: return float(self) == float(other) def __le__(self, other): if self._reduce_method is None: self._comp_error("__le__") else: return float(self) <= float(other) def __ge__(self, other): if self._reduce_method is None: self._comp_error("__ge__") else: return float(self) >= float(other) def __lt__(self, other): if self._reduce_method is None: self._comp_error("__lt__") else: return float(self) < float(other) def __gt__(self, other): if self._reduce_method is None: self._comp_error("__gt__") else: return float(self) > float(other) def __add__(self, other): if self._reduce_method is None: self._comp_error("__add__") else: return float(self) + float(other) def __sub__(self, other): if self._reduce_method is None: self._comp_error("__sub__") else: return float(self) - float(other) def __mul__(self, other): if self._reduce_method is None: self._comp_error("__mul__") else: return float(self) * float(other) def __format__(self, fmt): if self._reduce_method is None: raise ValueError( "Cannot format Stats object with reduce method `None` because it can " "not be reduced to a single value." ) else: return f"{float(self):{fmt}}" def _comp_error(self, comp): raise ValueError( f"Cannot {comp} Stats object with reduce method `None` to other " "because it can not be reduced to a single value." ) def get_state(self) -> Dict[str, Any]: state = { # Make sure we don't return any tensors here. "values": convert_to_numpy(self.values), "reduce": self._reduce_method, "percentiles": self._percentiles, "reduce_per_index_on_aggregate": self._reduce_per_index_on_aggregate, "window": self._window, "ema_coeff": self._ema_coeff, "clear_on_reduce": self._clear_on_reduce, "_last_reduced": self._last_reduced, "_is_tensor": self._is_tensor, } if self._throughput_stats is not None: state["throughput_stats"] = self._throughput_stats.get_state() return state @staticmethod def from_state(state: Dict[str, Any]) -> "Stats": # If `values` could contain tensors, don't reinstate them (b/c we don't know # whether we are on a supported device). values = state["values"] if "_is_tensor" in state and state["_is_tensor"]: values = [] if "throughput_stats" in state: throughput_stats = Stats.from_state(state["throughput_stats"]) stats = Stats( values, reduce=state["reduce"], percentiles=state.get("percentiles", False), reduce_per_index_on_aggregate=state.get( "reduce_per_index_on_aggregate", False ), window=state["window"], ema_coeff=state["ema_coeff"], clear_on_reduce=state["clear_on_reduce"], throughput=throughput_stats.peek(), throughput_ema_coeff=throughput_stats._ema_coeff, ) elif state.get("_throughput", False): # Older checkpoints have a _throughput key that is boolean or # a float (throughput value). They don't have a throughput_ema_coeff # so we use a default of 0.05. # TODO(Artur): Remove this after a few Ray releases. stats = Stats( values, reduce=state["reduce"], percentiles=state.get("percentiles", False), window=state["window"], ema_coeff=state["ema_coeff"], clear_on_reduce=state["clear_on_reduce"], throughput=state["_throughput"], throughput_ema_coeff=0.05, ) else: stats = Stats( values, reduce=state["reduce"], percentiles=state.get("percentiles", False), window=state["window"], ema_coeff=state["ema_coeff"], clear_on_reduce=state["clear_on_reduce"], throughput=False, throughput_ema_coeff=None, ) # Compatibility to old checkpoints where a reduce sometimes resulted in a single # values instead of a list such that the history would be a list of integers # instead of a list of lists. if "_hist" in state: # TODO(Artur): Remove this after a few Ray releases. if not isinstance(state["_hist"][0], list): state["_hist"] = list(map(lambda x: [x], state["_hist"])) stats._last_reduced = state["_hist"][-1] else: stats._last_reduced = state.get("_last_reduced", [np.nan]) return stats @staticmethod def similar_to( other: "Stats", init_values: Optional[Any] = None, ) -> "Stats": """Returns a new Stats object that's similar to `other`. "Similar" here means it has the exact same settings (reduce, window, ema_coeff, etc..). The initial values of the returned `Stats` are empty by default, but can be set as well. Args: other: The other Stats object to return a similar new Stats equivalent for. init_value: The initial value to already push into the returned Stats. Returns: A new Stats object similar to `other`, with the exact same settings and maybe a custom initial value (if provided; otherwise empty). """ stats = Stats( init_values=init_values, reduce=other._reduce_method, percentiles=other._percentiles, reduce_per_index_on_aggregate=other._reduce_per_index_on_aggregate, window=other._window, ema_coeff=other._ema_coeff, clear_on_reduce=other._clear_on_reduce, throughput=other._throughput_stats.peek() if other.has_throughput else False, throughput_ema_coeff=other._throughput_ema_coeff, ) stats.id_ = other.id_ stats._last_reduced = other._last_reduced return stats def _set_values(self, new_values): # For stats with window, use a deque with maxlen=window. # This way, we never store more values than absolutely necessary. if not self._inf_window: self.values = deque(new_values, maxlen=self._window) # For infinite windows, use `new_values` as-is (a list). else: self.values = new_values self._has_new_values = True def _reduced_values(self, values=None) -> Tuple[Any, Any]: """Runs a non-committed reduction procedure on given values (or `self.values`). Note that this method does NOT alter any state of `self` or the possibly provided list of `values`. It only returns new values as they should be adopted after a possible, actual reduction step. Args: values: The list of values to reduce. If not None, use `self.values` Returns: A tuple containing 1) the reduced values and 2) the new internal values list to be used. If there is no reduciton method, the reduced values will be the same as the values. """ values = values if values is not None else self.values # No reduction method. Return list as-is OR reduce list to len=window. if self._reduce_method is None: if self._percentiles is not False: # Sort values values = list(values) # (Artur): Numpy can sort faster than Python's built-in sort for large lists. Howoever, if we convert to an array here # and then sort, this only slightly (<2x) improved the runtime of this method, even for an internal values list of 1M values. values.sort() return values, values # Special case: Internal values list is empty -> return NaN or 0.0 for sum. elif len(values) == 0: if self._reduce_method in ["min", "max", "mean"] or self._has_returned_zero: # We also return np.nan if we have returned zero before. # This helps with cases where stats are cleared on reduce, but we don't want to log 0's, except for the first time. return [np.nan], [] else: return [0], [] # Do EMA (always a "mean" reduction; possibly using a window). elif self._ema_coeff is not None: # Perform EMA reduction over all values in internal values list. mean_value = values[0] for v in values[1:]: mean_value = self._ema_coeff * v + (1.0 - self._ema_coeff) * mean_value if self._inf_window: return [mean_value], [mean_value] else: return [mean_value], values # Non-EMA reduction (possibly using a window). else: # Use the numpy/torch "nan"-prefix to ignore NaN's in our value lists. if torch and torch.is_tensor(values[0]): self._is_tensor = True # Only one item in the if len(values[0].shape) == 0: reduced = values[0] else: reduce_meth = getattr(torch, "nan" + self._reduce_method) reduce_in = torch.stack(list(values)) if self._reduce_method == "mean": reduce_in = reduce_in.float() reduced = reduce_meth(reduce_in) else: reduce_meth = getattr(np, "nan" + self._reduce_method) if np.all(np.isnan(values)): # This avoids warnings for taking a mean of an empty array. reduced = np.nan else: reduced = reduce_meth(values) def safe_isnan(value): if torch and isinstance(value, torch.Tensor): return torch.isnan(value) return np.isnan(value) # Convert from numpy to primitive python types, if original `values` are # python types. if ( not safe_isnan(reduced) and reduced.shape == () and isinstance(values[0], (int, float)) ): if reduced.dtype in [np.int32, np.int64, np.int8, np.int16]: reduced = int(reduced) else: reduced = float(reduced) # For window=None|inf (infinite window) and reduce != mean, we don't have to # keep any values, except the last (reduced) one. if self._inf_window and self._reduce_method != "mean": # TODO (sven): What if values are torch tensors? In this case, we # would have to do reduction using `torch` above (not numpy) and only # then return the python primitive AND put the reduced new torch # tensor in the new `self.values`. return [reduced], [reduced] else: # In all other cases, keep the values that were also used for the reduce # operation. return [reduced], values @DeveloperAPI def compute_percentiles(sorted_list, percentiles): """Compute percentiles from an already sorted list. Note that this will not raise an error if the list is not sorted to avoid overhead. Args: sorted_list: A list of numbers sorted in ascending order percentiles: A list of percentile values (0-100) Returns: A dictionary mapping percentile values to their corresponding data values """ n = len(sorted_list) if n == 0: return {p: None for p in percentiles} results = {} for p in percentiles: index = (p / 100) * (n - 1) if index.is_integer(): results[p] = sorted_list[int(index)] else: lower_index = int(index) upper_index = lower_index + 1 weight = index - lower_index results[p] = ( sorted_list[lower_index] * (1 - weight) + sorted_list[upper_index] * weight ) return results @DeveloperAPI def merge_stats(base_stats: Optional[Stats], incoming_stats: List[Stats]) -> Stats: """Merges Stats objects. If `base_stats` is None, we use the first incoming Stats object as the new base Stats object. If `base_stats` is not None, we merge all incoming Stats objects into the base Stats object. Args: base_stats: The base Stats object to merge into. incoming_stats: The list of Stats objects to merge. Returns: The merged Stats object. """ if base_stats is None: new_root_stats = True else: new_root_stats = False # Nothing to be merged if len(incoming_stats) == 0: return base_stats if new_root_stats: # We need to deepcopy here first because stats from incoming_stats may be altered in the future base_stats = copy.deepcopy(incoming_stats[0]) base_stats.clear_throughput() # Note that we may take a mean of means here, which is not the same as a # mean of all values. In the future, we could implement a weighted mean # of means here by introducing a new Stats object that counts samples # for each mean Stats object. if len(incoming_stats) > 1: base_stats.merge_in_parallel(*incoming_stats[1:]) if ( base_stats._reduce_method == "sum" and base_stats._inf_window and base_stats._clear_on_reduce is False ): for stat in incoming_stats: base_stats._prev_merge_values[stat.id_] = stat.peek() elif len(incoming_stats) > 0: # Special case: `base_stats` is a lifetime sum (reduce=sum, # clear_on_reduce=False) -> We subtract the previous value (from 2 # `reduce()` calls ago) from all to-be-merged stats, so we don't count # twice the older sum from before. # Also, for the merged, new throughput value, we need to find out what the # actual value-delta is between before the last reduce and the current one. added_sum = 0.0 # Used in `base_stats._recompute_throughput` if applicable. if ( base_stats._reduce_method == "sum" and base_stats._inf_window and base_stats._clear_on_reduce is False ): for stat in incoming_stats: # Subtract "lifetime counts" from the Stat's values to not count # older "lifetime counts" more than once. prev_reduction = base_stats._prev_merge_values[stat.id_] new_reduction = stat.peek(compile=True) base_stats.values[-1] -= prev_reduction # Keep track of how many counts we actually gained (for throughput # recomputation). added_sum += new_reduction - prev_reduction base_stats._prev_merge_values[stat.id_] = new_reduction parallel_merged_stat = copy.deepcopy(incoming_stats[0]) if len(incoming_stats) > 1: # There are more than one incoming parallel others -> Merge all of # them in parallel (equal importance). parallel_merged_stat.merge_in_parallel(*incoming_stats[1:]) # Merge incoming Stats object into base Stats object on time axis # (giving incoming ones priority). if base_stats._reduce_method == "mean" and not base_stats._clear_on_reduce: # If we don't clear values, values that are not cleared would contribute # to the mean multiple times. base_stats._set_values(parallel_merged_stat.values.copy()) else: base_stats.merge_on_time_axis(parallel_merged_stat) # Keep track of throughput through the sum of added counts. if base_stats.has_throughput: base_stats._recompute_throughput(added_sum) return base_stats