Блог

Я использую twisted для создания «клиента / сервера», программы, которая действует как клиент для получения некоторых данных и как сервер для повторной отправки этих данных или просто для отправки других данных клиентам.

Я использую twistd, и мой код для инициализации моих фабрик выглядит следующим образом:

 application = service.Application('monitorD', uid=0, gid=0)
factoryMonitord = ServerFactory(p)
internet.TCPServer(9000, factoryMonitord).setServiceParent(service.IServiceCollection(application))
#because I need to send some datas from client factory to clients of serverfactory
factoryScamd = ClientFactory(factoryMonitord)
internet.TCPClient("localhost", 8001, factoryScamd).setServiceParent(service.IServiceCollection(application))
  

Моя проблема в том, что клиентская часть не может подключиться (потому что сервер недоступен), похоже, что она «блокирует» всю мою серверную часть. Все еще возможно обмениваться данными между моей серверной частью и другими клиентами, но это действительно медленно … (и когда моя клиентская часть может быть подключена, она просто работает нормально).

Заранее спасибо за любую помощь.

Редактировать:

Вот мой ServerFactory-код (много бесполезного кода …):

 class ServerFactory(protocol.ServerFactory):
    protocol = ServerProtocol

    def __init__(self, portal):
        #self.tp = ClientFactory.tp 
        self.tp = []
        self.portal = portal
        self.loop_vol_raid_areca = LoopingCall(self.checkVolRaidAreca)
        self.loop_vol_raid_areca.start(30)
        self.loop_services = LoopingCall(self.checkServices)
        self.loop_services.start(30)

    def buildProtocol(self, addr):
        p = protocol.ServerFactory.buildProtocol(self, addr)
        p.portal = self.portal
        return p

    def sendList(self, data):
        if data:
            if isinstance(data, list):
                for element in data:
                    if isinstance(element, list):
                        self.clean_data = "".join(element)
                self.sendToClients(self.clean_data)

    def sendToClients(self, data):
        print "SEND to MonitorC from MonitorD, tp:", self.tp
        if data:
            for tp in self.tp:
                self.protocol.sendLine(tp, data)

    def checkServices(self):
        self.scamd = threads.deferToThread(checkScamd)
        self.scamd.addCallback(self.sendToClients)
        self.httpd = threads.deferToThread(checkHttpd)
        self.httpd.addCallback(self.sendToClients)
        self.postgres = threads.deferToThread(checkPostgres)
        self.postgres.addCallback(self.sendToClients)

    def checkVolRaidAreca(self):
        self.vol = threads.deferToThread(check_vol_areca)
        self.vol.addCallback(self.sendList)
        self.event = threads.deferToThread(get_last_event_areca)
        self.event.addCallback(self.sendList)
  

И вот клиентская фабрика с большим количеством бесполезного кода:

 class ClientFactory(protocol.ClientFactory):
    protocol = ClientProtocol

    def __init__(self, MonitordFactory):
        self.tp = MonitordFactory.tp

    def clientConnectionFailed(self, connector, reason):
        print "Connection to scamd failed - retrying..."
        time.sleep(30)
        connector.connect()

    def clientConnectionLost(self, connector, reason):
        print "Connection to scamd lost - retrying..."
        time.sleep(30)
        connector.connect()

    def getCamList(self, data):
        cams_list = data.split("x00")
        self.cams_array = []
        for i in cams_list:
            if str.find(i, "camera") != -1:
                i = i.split(" ")
                i = filter(lambda x: len(x)>0, i)
                self.cams_array.append(i)

    def checkCams(self, data):
        data = data.split(" ")
        for i in self.cams_array:
            if i[1] == data[2]:
                if data[3] == "-1":
                    msg = i[6]   " ERREUR -1"
                if data[3] == "0":
                    msg = i[6]   " ERREUR 0"
                if data[3] == "1":
                    msg = i[6]   " ERREUR 1"
                if data[3] == "2":
                    msg = i[6]   " ERREUR 2 (RECO)"
        return msg
  

Если потребуется дополнительная информация, я опубликую весь код в pastebin.
И я новичок в python И twisted (но раньше я кодировал на C или C )

Есть две вещи, которые вы должны быть в состоянии изменить сразу. Избавьтесь от time.sleep(30) вызовов. Кроме того, вы интенсивно используете threadpool, поэтому, вероятно, захотите увеличить его размер. Каждые 30 секунд вы создаете 5 потоков, и поскольку размер пула потоков reactor по умолчанию равен 5, любые другие потоки, которые вы создаете, в конечном итоге будут ждать за этими 5. Я думаю, что наличие множества ожидающих потоков в очереди может быть причиной того, что ваш сервис работает медленнее или даже блокируется.

В краткосрочной перспективе вы можете просто создать свой собственный twisted.python.threadpool.Threadpool в своей ServerFactory:

 class ServerFactory(protocol.ServerFactory):
    protocol = ServerProtocol

    def __init__(self, portal):
        # ...
        self.threadpool = ThreadPool(25, 50)

    def checkServices(self):
        # ...
        self.postgres = threads.deferToThreadPool(reactor, self.threadpool, checkPostgres)
  

Однако в долгосрочной перспективе, где это возможно, вам следует рассмотреть возможность использования асинхронного API для работы с вашими сервисами.

Например, ваш checkHttpd код можно было бы сделать асинхронным с помощью twisted.web.client.Агент. Для PostgreSQL вы можете использовать txPostgres. Я не знаком с вашей scamd службой выше, поэтому подключение к ней с помощью асинхронного ввода-вывода может быть более сложным (т. Е. вам, возможно, придется написать асинхронную версию протокола для этой службы).