def terminate_now()

in dataflux_core/fast_list.py [0:0]

• 7 lines of code
• 2 McCabe index (conditional complexity)

    def terminate_now(
            self, processes: "list[multiprocessing.Process]") -> RuntimeError:
        """Terminates all processes immediately.

        Args:
          processes: The full list of multiprocessing processes.

        Returns:
            RuntimeError indicating that one or more multiprocess processes has
            become unresponsive
        """
        for p in processes:
            p.terminate()
        raise RuntimeError(
            "multiprocessing child process became unresponsive; check logs for underlying error"
        )