heartbeat_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 heartbeat(self) -> None:
        """
        Add an heartbeat to the state of the instance.
        """
        timestamp = self._get_time()
        old_state = self._state.get()

        if old_state.timestamp is None:
            new_history = self._first_heartbeat()
        else:
            latest_timestamp = old_state.timestamp
            interval = timestamp - latest_timestamp
            new_history = old_state.history

            if self._is_available(timestamp):
                new_history += interval

        new_state = _State(history=new_history, timestamp=timestamp)

        # Eventually it retry the heartbeat operation when a concurrent process modify the state
        if not self._state.compare_and_set(old_state, new_state):
            self.heartbeat()