Порт udp заблокирован во всех процессах (python - pi)

Question

У меня есть сеть контроллеров Raspberry PI в домашней системе безопасности. Каждый запускает несколько независимых процессов для различных функций, и MQTT-брокер связывает все вместе.

Каждый PI / процесс жестко закодирован с адресом брокера MQTT, и вместо этого я бы хотел, чтобы каждый запросил центральный сервер для определения местоположения брокера.

У меня каждый процесс открывает набор сокетов UDP для широковещательной рассылки, и каждый из них привязывается к своему порту прослушивания. При запуске он отправляет запрос (который включает в себя порт прослушивания) на центральный сервер, который отправляет ответ на правильный порт прослушивания.

Это прекрасно работает только для одного процесса, других (на том же PI) ) все время без ответа. Wireshark показывает, что только один из трех фактически отправляет запрос на провод, и любопытно, что работает последний процесс открытия сокета. Фрагмент из журнала отладки:

mpi@RPI02:~ $ cat /var/log/vlab.debug.log
2020-02-28 17:08:55,242 garagectlr.31: startup
2020-02-28 17:08:55,243 root.2:
2020-02-28 17:08:55,244 root.2: startup
2020-02-28 17:08:55,250 root: looking for broker on port 10417...
2020-02-28 17:08:55,255 garagectlr: looking for broker on port 10416...
2020-02-28 17:08:57,117 sms-receiver.18:
2020-02-28 17:08:57,118 sms-receiver.18: startup
2020-02-28 17:08:57,121 sms-receiver: looking for broker on port 10419...
2020-02-28 17:08:57,124 sms-receiver: rx [mqtt-broker=192.168.99.99]
2020-02-28 17:08:57,225 sms-receiver: located broker at 192.168.99.99
2020-02-28 17:08:57,255 root: FATAL: unable to locate MQTT broker
2020-02-28 17:08:57,260 garagectlr: FATAL: unable to locate MQTT broker
2020-02-28 17:08:57,270  * Running on http://192.168.99.162:6060/ (Press CTRL+C to quit)

Ниже приведен соответствующий код python для каждого PI, немного оторванный:

    #
    # start looking for broker
    #

    # get unique listening port
    listen_port = os.getpid() + 10000
    logging.debug("%s: looking for broker on port %d..."%(self.process,listen_port))
    try:
        sock = socket.socket(socket.AF_INET,socket.SOCK_DGRAM)
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
        sock.bind(('',listen_port))
    except Exception:
        logging.exception("%s: "%(self.process))

    tries = 0
    try:
        while self.broker == None and tries < 2:
            query = "query mqtt-broker:%s:%d"%(self.process,listen_port)
            sock.sendto(query,("192.168.99.255",1414))
            sock.settimeout(2)
            data,addr = sock.recvfrom(64)
            logging.debug("%s: rx [%s]"%(self.process,data))
            if data[0:5]=='mqtt-':
                self.broker = data[12:]
                time.sleep(.1)
            else:
                time.sleep(1)
                tries = tries + 1
    except Exception:
        print('timeout')
    sock.close()        
    if self.broker == None:
        logging.debug("%s: FATAL: unable to locate MQTT broker"%(self.process))
    else:    
        logging.debug("%s: located broker at %s"%(self.process,self.broker))
        self.sendBoot()            
        self.pubTest("mqtt@%s.%s startup"%(self.process,self.version))            
        m_thread = threading.Thread(target = self.mqtt_thread)
        m_thread.start()

Я думал, что, связываясь с уникальными портами для RX, у меня не было бы никаких помех между сокетами, но явно работает только один. Почему два других даже не отправляют запрос?

Answer 1

Я заметил, что WHILE l oop находится внутри блока TRY-CATCH, так что при первом тайм-ауте приемника время l oop было сломано.

Я поместил TRY в WHILE следующим образом:

    tries = 0
    query = "query mqtt-broker:%s:%d"%(self.process,listen_port)

    while self.broker == None and tries < 4:
        try:
            sock.sendto(query,("192.168.99.255",1414))
            sock.settimeout(2)
            data,addr = sock.recvfrom(64)
            if data[0:5]=='mqtt-':
                self.broker = data[12:]
                time.sleep(.1)
            else:
                time.sleep(1)
                tries = tries + 1
        except Exception:
            pass    ## timeout

    try:
        sock.close()        
    except Exception:
        logging.exception("%s: "%(self.process))

, думая, что у каждого процесса будет возможность повторить попытку в случае сбоя запроса. Во-первых, я не уверен, что было не так, но теперь это работает достаточно надежно:

mpi@RPI02:~ $ cat /var/log/vlab.debug.log
2020-02-29 17:07:47,981 root.2:
2020-02-29 17:07:47,982 root.2: startup
2020-02-29 17:07:47,986 garagectlr.31: startup
2020-02-29 17:07:47,992 garagectlr: looking for broker on port 10421...
2020-02-29 17:07:48,010 root: looking for broker on port 10424...
2020-02-29 17:07:49,895 sms-receiver.18:
2020-02-29 17:07:49,896 sms-receiver.18: startup
2020-02-29 17:07:49,901 sms-receiver: looking for broker on port 10423...
2020-02-29 17:07:50,004 sms-receiver: located broker at 192.168.99.99
2020-02-29 17:07:50,048 sms-receiver:  * Running on http://192.168.99.162:6060
2020-02-29 17:07:50,098 garagectlr: located broker at 192.168.99.99
2020-02-29 17:07:50,140 root: located broker at 192.168.99.99
mpi@RPI02:~ $

Мне кажется странным, что запросы составляются, а затем выполняются в обратном порядке. Но эй, это работает.