Récupération FTP non bloquante du client et écriture

Question

Je construis un serveur de test FTP et un client avec Twisted. Le serveur fonctionne très bien. Il est fondamentalement le même à partir de l'exemple Twisted ftpserver.py. Le client est l'endroit où j'ai des problèmes de blocage pendant la récupération et l'écriture de mes fichiers. J'ai essayé de le résoudre grâce à quelques utilitaires Twisted rapides, mais en vain.

Voici mon serveur:

#!/usr/bin/env python2 
from __future__ import print_function, division, absolute_import 

# Import twisted things 

from twisted.protocols.ftp import FTPFactory 
from twisted.protocols.ftp import FTPRealm 
from twisted.internet import reactor 
from twisted.cred.portal import Portal 
from twisted.cred.checkers import AllowAnonymousAccess 

p = Portal(FTPRealm("test/"), [AllowAnonymousAccess()]) 

f = FTPFactory(p) 
f.timeOut = None 

reactor.listenTCP(5504, f) 
reactor.run() 

Le côté client, associé à cela, est un simple GUI wxPython qui présente une zone de texte pour écrire le nom du fichier que vous souhaitez récupérer dans l'intérieur de cette interface graphique. il y a un wx.Timer qui exécute une méthode toutes les 50 millisecondes. C'est ce qui bloque la récupération de mon fichier FTP. Je trouve que parce que le thread principal est utilisé, le protocole qui reçoit les données est en train d'avoir des hoquets. Si vous vous demandez pourquoi j'ai cette configuration, je simule le cas d'utilisation pour un projet beaucoup plus important.

Ma tentative pour résoudre cela a été d'utiliser deferToThread sur le point spécifique lorsque j'ai besoin de récupérer un fichier. Cependant, en imprimant le thread actuel, je trouve que le protocole qui reçoit les données est en cours d'exécution dans le thread principal. C'est le problème que j'essaie de résoudre. Toute aide est très appréciée.

Mon code client:

#!/usr/bin/env python2 
from __future__ import print_function, division, absolute_import 

import wx 
import sys 
import threading 

from twisted.internet import wxreactor 
wxreactor.install() 

from twisted.internet import reactor 

from twisted.protocols.ftp import FTPClient 

from twisted.internet import protocol 
from twisted.internet import threads 
from twisted.python import log 

# This is the GUI 
class TextSend(wx.Frame): 

    def __init__(self): 
     wx.Frame.__init__(self, None, -1, "Request Files", size=(200, 75)) 

     self.protocol = None # ftp client protocol 
     self.factory = None 

     panel = wx.Panel(self) 

     vertSizer = wx.BoxSizer(wx.VERTICAL) 
     horzSizer = wx.BoxSizer(wx.HORIZONTAL) 

     self.fileName = None 
     self.textbox = wx.TextCtrl(parent=panel, id=100, size=(100,-1)) 
     self.btn = wx.Button(panel, label="Retr.") 

     # timer and checkbox for timer 
     self.timer = wx.Timer(self, id=wx.ID_ANY) 
     self.check = wx.CheckBox(parent=panel, label="Start blocking") 

     #Bind 
     self.textbox.Bind(wx.EVT_TEXT, self.getText) 
     self.btn.Bind(wx.EVT_BUTTON, self.press) 
     self.check.Bind(wx.EVT_CHECKBOX, self.onCheck) 
     self.Bind(wx.EVT_TIMER, self.onTimer, self.timer) 

     horzSizer.Add(self.textbox, flag=wx.ALIGN_CENTER) 
     horzSizer.Add(self.btn, flag=wx.ALIGN_CENTER) 

     vertSizer.Add(horzSizer, flag=wx.ALIGN_CENTER) 
     vertSizer.Add(self.check, flag=wx.ALIGN_CENTER) 

     panel.SetSizer(vertSizer) 
     panel.Layout() 

    def getText(self, evt): 
     self.fileName = str(self.textbox.GetValue()) 

    def onCheck(self, evt): 
     yes = self.check.GetValue() 
     if yes: 
      print("Starting timer") 
      self.timer.Start(50) 
     else: # no 
      self.timer.Stop() 

    def onTimer(self, evt): 
     #print("Triggered timer") 
     pass 

    def press(self, evt): 
     print("Send:", self.fileName) 

     d = threads.deferToThread(self.retrieve) 
     d.addCallback(self.done) 

    def retrieve(self): 
     print(threading.current_thread()) 
     # This is what does the retrieving. Pass in FileWriter and 
     # FileWriter's dataReceived method is called by main thread 
     self.protocol.retrieveFile(self.fileName, FileWriter(self.fileName), offset=0).addCallbacks(self.done, self.fail) 
     return "Done with deferToThread" 

    def done(self, msg): 
     print(threading.current_thread()) 
     print("DONE Retrieving:", msg) 

    def fail(self, error): 
     print('Failed. Error was:') 
     print(error) 

# This writes to the file of a same name as the one retrieved. 
class FileWriter(protocol.Protocol): 

    def __init__(self, fileName): 
     self.f = open(fileName, 'wb') 
     print("FROM FileWriter __init__:", threading.current_thread()) 

    def dataReceived(self, data): 
     print("Byte size", len(data)) 
     print("FROM FileWriter dataReceived:", threading.current_thread()) 
     self.f.write(data) 

    def connectionLost(self, reason): 
     print("Writing closed and done") 
     print("FROM FileWriter connectionLost:", threading.current_thread()) 
     self.f.close() 

# Client FTP Protocol 
class TestClient(FTPClient, object): 

    def __init__(self, factory, username, password, passive): 
     super(TestClient, self).__init__(username=username, password=password, passive=passive) 
     self.factory = factory 

    def connectionMade(self): 
     print("hello") 
     gui = self.factory.gui 
     gui.protocol = self 

# Twisted Client Factory 
class FileClientFactory(protocol.ClientFactory): 

    def __init__(self, gui): 
     self.gui = gui 
     self.protocol = None 

    def buildProtocol(self, addr): 
     user = 'anonymous' 
     passwd = 'twisted@matrix.com' 
     self.protocol = TestClient(self, username=user, password=passwd, passive=1) 
     return self.protocol 

    def clientConnectionLost(self, transport, reason): 
     print("Connectiong lost normally:", reason) 

    def clientConnectionFailed(self, transport, reason): 
     print("Connection failed:", reason) 


if __name__ == "__main__": 
    # Initialize and show GUI 
    logger = log.startLogging(sys.stdout) 
    app = wx.App(False) 
    app.frame = TextSend() 
    app.frame.Show() 
    reactor.registerWxApp(app) 

    # Build Factory 
    f = FileClientFactory(app.frame) 

    # Connect to FTP server 
    reactor.connectTCP("localhost", 5504, f) 
    reactor.run() 

    wxPython main loop. 
    app.MainLoop() 

Answer 1

Vous ne pouvez pas deferToThread(function_that_uses_twisted_apis). Les API Twisted sont presque toutes non-thread-safe. Vous devez les utiliser uniquement dans le thread du réacteur (les exceptions sont quelques API liées à la planification de threads).

À la place, débarrassez-vous de votre code de blocage. Mettez dans un autre thread, un autre processus, ou réécrivez-le pour être non-bloquant.