Как обнюхать сетевой интерфейс с помощью Twisted?

Вы можете запустить распределенную систему и передавать данные через центральную систему очередей. Возьмите философию Unix и создайте одно приложение, которое выполняет несколько задач и выполняет их хорошо. Создайте одно приложение, которое обнюхивает пакеты (вы можете использовать здесь scapy, поскольку это не имеет особого значения, если вы что-то блокируете), затем отправляет их в очередь (RabitMQ, Redis, SQS и т. д.), А другое приложение обрабатывает пакет из очереди. . Этот метод должен вызвать наименьшую головную боль.

Если вам нужно запустить все в одном приложении, то единственный вариант - потоки / многопроцессорность. Но есть некоторые шаблоны проектирования, которым вы захотите следовать. Вы также можете разбить следующий код на отдельные функции и использовать выделенную систему очередей.

from threading import Thread
from time import sleep
from twisted.internet import defer, reactor

class Sniffer(Thread):
    def __init__(self, _reactor, shared_queue):
        super().__init__()
        self.reactor = _reactor
        self.shared_queue = shared_queue

    def run(self):
        """
        Sniffer logic here
        """
        while True:
            self.reactor.callFromThread(self.shared_queue.put, 'hello world')
            sleep(5)

@defer.inlineCallbacks
def consume_from_queue(_id, _reactor, shared_queue):
    item = yield shared_queue.get()
    print(str(_id), item)
    _reactor.callLater(0, consume_from_queue, _id, _reactor, shared_queue)

def main():
    shared_queue = defer.DeferredQueue()
    sniffer = Sniffer(reactor, shared_queue)
    sniffer.daemon = True
    sniffer.start()

    workers = 4
    for i in range(workers):
        consume_from_queue(i+1, reactor, shared_queue)

    reactor.run()

main()

Класс Sniffer запускается вне контроля Twisted. Обратите внимание на sniffer.daemon = True, это сделано для того, чтобы поток останавливался, когда остановился основной поток. Если он был установлен на False (по умолчанию), то приложение выйдет только в том случае, если все потоки подошли к концу. В зависимости от поставленной задачи это может быть, а может и не всегда быть возможным. Если вы можете сделать перерывы в прослушивании, чтобы проверить событие потока, вы могли бы остановить поток более безопасным способом.

self.reactor.callFromThread(self.shared_queue.put, 'hello world') необходим, чтобы элемент, помещаемый в очередь, происходил в основном потоке реактора, а не в потоке, который выполняет Sniffer. Основным преимуществом этого будет то, что будет какая-то синхронизация сообщений, поступающих из потоков (при условии, что вы планируете масштабирование для прослушивания нескольких интерфейсов). Кроме того, я не был уверен, что объекты DeferredQueue являются потокобезопасными :) Я относился к ним так, как будто они не были.

Поскольку Twisted не управляет потоками в этом случае, жизненно важно, чтобы это делал разработчик. Обратите внимание на петлю worker и consume_from_queue(i+1, reactor, shared_queue). Этот цикл гарантирует, что с задачами будет работать только желаемое количество рабочих. Внутри функции consume_from_queue()shared_queue.get() будет ждать (неблокирующе), пока элемент не будет помещен в очередь, распечатает элемент, а затем планирует другой consume_from_queue().