phi_func.py

import math
import time

from atomos.atomic import AtomicReference

from src._heartbeat_history import _HeartbeatHistory
from src._state import _State


class PhiAccrualFailureDetector:
    """
    The φ Accrual Failure Detector implementation.

    See:
        https://github.com/akka/akka/blob/master/akka-remote/src/main/scala/akka/remote/PhiAccrualFailureDetector.scala
    """

    # ...
    
    def _phi(self, timestamp: float) -> float:
        """
        Calculate the φ based on the current state and the timestamp.
        """
        last_state = self._state
        last_timestamp = last_state.get().timestamp

        if last_timestamp is None:
            return 0.0

        time_diff = timestamp - last_timestamp
        last_history = last_state.get().history
        mean = last_history.mean()
        std_dev = self._ensure_valid_std_deviation(last_history.std_dev())

        return self._calc_phi(time_diff, mean + self.acceptable_heartbeat_pause_ms, std_dev)


    @classmethod
    def _calc_phi(cls, time_diff: float, mean: float, std_dev: float) -> float:
        """
        Core method for calculating the phi φ coefficient. It uses a logistic approximation to the cumulative normal
        distribution.
        See: https://github.com/akka/akka/issues/1821
        """
        y = (time_diff - mean) / std_dev
        
        try:
            e = math.exp(-y * (1.5976 + 0.070566 * y * y))
        except OverflowError:
            e = float('inf')

        if time_diff > mean:
            return -math.log10(e / (1.0 + e))
        else:
            return -math.log10(1.0 - 1.0 / (1.0 + e))