Basic multithreaded ping with python (on Linux systems)

multithreaded_ping.py

from dataclasses import dataclass
from datetime import datetime
from queue import Empty, Queue
import re
import subprocess
from threading import Thread
import time


@dataclass
class PingResult:
    """A class to store a ping test result."""
    ip: str
    is_up: bool
    rtt_ms: float = None


def ping_thread(addrs_q: Queue, results_q: Queue) -> None:
    """Thread code to process (i.e. ping) addresses in addrs_q queue, append result to results_q queue."""
    # thread main loop
    while True:
        # get an IP address from queue
        ip = addrs_q.get()
        # ping it
        args = ['/bin/ping', '-c', '1', '-W', '5', str(ip)]
        p_ping = subprocess.Popen(args, shell=False, stdout=subprocess.PIPE)
        # save ping stdout
        p_ping_out = p_ping.communicate()[0].decode('utf-8')
        # up
        if p_ping.wait() == 0:
            search = re.search(r'rtt min/avg/max/mdev = (.*)/(.*)/(.*)/(.*) ms', p_ping_out, re.M | re.I)
            ping_rtt = float(search.group(2))
            results_q.put(PingResult(ip=ip, is_up=True, rtt_ms=ping_rtt))
        # down
        else:
            results_q.put(PingResult(ip=ip, is_up=False))
        # update queue (this address is processed)
        addrs_q.task_done()


if __name__ == '__main__':
    # init an addresses queue and a results queue
    addrs_q: Queue[str] = Queue()
    results_q: Queue[PingResult] = Queue()
    # start a pool of 50 threads to process it
    for _ in range(50):
        Thread(target=ping_thread, args=(addrs_q, results_q, ), daemon=True).start()
    # infinite tests loop
    while True:
        # populate queue with some IPs addresses
        for lsb in range(1, 255):
            addrs_q.put(f'192.168.1.{lsb}')
        # wait until all addresses are processed by threads
        addrs_q.join()
        # process all "up" results produce by ping threads
        print(f'some IPs up at {datetime.now().isoformat()}:')
        while True:
            try:
                ping_result = results_q.get_nowait()
                if ping_result.is_up:
                    print(ping_result)
            except Empty:
                break
        # wait before next test
        time.sleep(60.0)