Source code for reframe.core.deferrable

# Copyright 2016-2024 Swiss National Supercomputing Centre (CSCS/ETH Zurich)
# ReFrame Project Developers. See the top-level LICENSE file for details.
#
# SPDX-License-Identifier: BSD-3-Clause

import builtins
import functools


[docs] def deferrable(func): '''Convert the decorated function to a deferred expression. See :ref:`deferrable-functions` for further information on deferrable functions. ''' @functools.wraps(func) def _deferred(*args, **kwargs): return _DeferredExpression(func, *args, **kwargs) return _deferred
class _DeferredExpression: '''Represents an expression whose evaluation has been deferred. This class simply stores a callable and its arguments and will evaluate it as soon as the `evaluate()` method is called. This class implements the basic unary and binary operators of Python so that it makes it possible to defer also arbitrary expressions. Note the `not`, `and` and `or` operators cannot be overloaded. If you want to defer an expression containing such operators, you should use the provided `and_`, `or_` or `not_` deferred functions. Deferred expressions may by chained together. Chaining happens automatically if an argument to a function or an operand of an operator are deferred expressions, too. When you later evaluate the outermost expression, the evaluation will go down the chain of deferred expressions and evaluate them all. `_DeferredExpression` are immutable objects. ''' def __init__(self, fn, *args, **kwargs): self._fn = fn self._args = args self._kwargs = kwargs # We cache the value of the last evaluation inside a tuple. # We don't cache the value directly, because it can be any. self._cached = () self._return_cached = False def evaluate(self, cache=False): # Return the cached value (if any) if self._return_cached and not cache: return self._cached[0] elif cache: self._return_cached = cache fn_args = [] for arg in self._args: fn_args.append( arg.evaluate() if isinstance(arg, _DeferredExpression) else arg ) fn_kwargs = {} for k, v in self._kwargs.items(): fn_kwargs[k] = ( v.evaluate() if isinstance(v, _DeferredExpression) else v ) ret = self._fn(*fn_args, **fn_kwargs) # Evaluate the return for as long as a deferred expression returns # another deferred expression. while isinstance(ret, _DeferredExpression): ret = ret.evaluate() # Cache the results for any subsequent evaluate calls. self._cached = (ret,) return ret def __bool__(self): '''The truthy value of a deferred expression. This causes the immediate evaluation of the deferred expression. ''' return builtins.bool(self.evaluate()) def __str__(self): '''Evaluate the deferred expresion and return its string representation.''' return str(self.evaluate()) def __iter__(self): '''Evaluate the deferred expression and iterate over the result.''' return iter(self.evaluate()) def __rfm_json_encode__(self): if self._cached == (): return None else: return self._cached[0] # Overload Python operators to be able to defer any expression # # NOTE: In the following we are not using `self` for denoting the first # argument. These operators are just there to capture and defer the # corresponding expression. When we evaluate this deferred expression, they # will be called with the real arguments of the originally captured # expression. That's why, in order to avoid confusion, we do not use `self` # as a formal argument. For example: # # D = make_deferrable(1) # D' = D == 2 --> this calls _DeferredExpression.__eq__(D, 2) # evaluate(D') --> this eventually calls _DeferredExpression.__eq__(1, 2) @deferrable def __eq__(a, b): return a == b @deferrable def __ne__(a, b): return a != b @deferrable def __lt__(a, b): return a < b @deferrable def __le__(a, b): return a <= b @deferrable def __gt__(a, b): return a > b @deferrable def __ge__(a, b): return a >= b @deferrable def __getitem__(seq, key): return seq[key] @deferrable def __contains__(seq, key): '''This method triggers the evaluation of the resulting expression. If you want a really deferred check, you should use `reframe.utility.sanity.contains()`. This happens because Python always converts the result of `__contains__()` to a boolean value by calling `bool()`, which in our case it triggers the evaluation of the expression. ''' return key in seq @deferrable def __add__(a, b): return a + b @deferrable def __sub__(a, b): return a - b @deferrable def __mul__(a, b): return a * b @deferrable def __matmul__(a, b): return a @ b @deferrable def __truediv__(a, b): return a / b @deferrable def __floordiv__(a, b): return a // b @deferrable def __mod__(a, b): return a % b def __divmod__(self, other): '''This is not deferrable. Instead it returns a tuple of deferrables that compute the floordiv and the mod. ''' return (self.__floordiv__(other), self.__mod__(other)) @deferrable def __pow__(a, b): return a**b @deferrable def __lshift__(a, b): return a << b @deferrable def __rshift__(a, b): return a >> b @deferrable def __and__(a, b): return a & b @deferrable def __xor__(a, b): return a ^ b @deferrable def __or__(a, b): return a | b # Reflected operators @deferrable def __radd__(a, b): return b + a @deferrable def __rsub__(a, b): return b - a @deferrable def __rmul__(a, b): return b * a @deferrable def __rmatmul__(a, b): return b @ a @deferrable def __rtruediv__(a, b): return b / a @deferrable def __rfloordiv__(a, b): return b // a @deferrable def __rmod__(a, b): return b % a def __rdivmod__(self, other): '''This is not deferrable. Instead it returns a tuple of deferrables that compute the rfloordiv and the rmod. ''' return (self.__rfloordiv__(other), self.__rmod__(other)) @deferrable def __rpow__(a, b): return b**a @deferrable def __rlshift__(a, b): return b << a @deferrable def __rrshift__(a, b): return b >> a @deferrable def __rand__(a, b): return a & b @deferrable def __rxor__(a, b): return b ^ a @deferrable def __ror__(a, b): return b | a # Augmented operators # # NOTE: These are usually part of mutable objects, however # _DeferredExpression remains immutable, since it eventually delegates # their evaluation to the objects it wraps @deferrable def __iadd__(a, b): a += b return a @deferrable def __isub__(a, b): a -= b return a @deferrable def __imul__(a, b): a *= b return a @deferrable def __imatmul__(a, b): a @= b return a @deferrable def __itruediv__(a, b): a /= b return a @deferrable def __ifloordiv__(a, b): a //= b return a @deferrable def __imod__(a, b): a %= b return a @deferrable def __ipow__(a, b): a **= b return a @deferrable def __ilshift__(a, b): a <<= b return a @deferrable def __irshift__(a, b): a >>= b return a @deferrable def __iand__(a, b): a &= b return a @deferrable def __ixor__(a, b): a ^= b return a @deferrable def __ior__(a, b): a |= b return a # Unary operators @deferrable def __neg__(a): return -a @deferrable def __pos__(a): return +a @deferrable def __abs__(a): return abs(a) @deferrable def __invert__(a): return ~a class _DeferredPerformanceExpression(_DeferredExpression): '''Represents a performance function whose evaluation has been deferred. It extends the :class:`_DeferredExpression` class by adding the ``unit`` attribute. This attribute represents the unit of the performance metric to be extracted by the performance function. ''' def __init__(self, fn, unit, *args, **kwargs): super().__init__(fn, *args, **kwargs) if not isinstance(unit, str): raise TypeError( 'performance units must be a string' ) self._unit = unit @classmethod def construct_from_deferred_expr(cls, expr, unit): if not isinstance(expr, _DeferredExpression): raise TypeError("'expr' argument is not an instance of the " "_DeferredExpression class") return cls(expr._fn, unit, *(expr._args), **(expr._kwargs)) @property def unit(self): return self._unit