def check_crashed_processes()

in dataflux_core/fast_list.py [0:0]

• 15 lines of code
• 9 McCabe index (conditional complexity)

    def check_crashed_processes(self) -> bool:
        """Checks if any processes have crashed.

        Returns:
          A boolean indicating if any processes have crashed after initialization.
          If this function returns true, it indicates a need to restart the listing
          operation.
        """
        logging.debug("checking for crashed procs...")
        now = time.time()
        crashed = []
        # Wait at least 60 seconds or 2 times the API call retry delay for check-ins,
        # otherwise processes might appear to be crashed while retrying API calls.
        checkin_wait = 2 * self.retry_config._maximum if self.retry_config else 0
        checkin_wait = max(checkin_wait, 60)
        for inited_worker, last_checkin in self.checkins.items():
            if now - last_checkin > checkin_wait:
                crashed.append(inited_worker)
            for proc in crashed:
                if proc in self.inited:
                    logging.error(
                        "process crash detected, ending list procedure...")
                    return True
        return False