Non-blocking Client FTP-Abruf und Schreiben

Question

Ich baue einen FTP-Test-Server und Client mit Twisted. Der Server läuft super. Dies ist im Wesentlichen das gleiche wie im Beispiel Twisted ftpserver.py. Der Client ist, wo ich einige blockierende Probleme während meiner Dateiabruf und Schreiben habe. Ich habe versucht, es durch einige schnelle Twisted Threading-Dienstprogramme zu lösen, aber ohne Erfolg.

Hier mein Server ist:

#!/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() 

Die Client-Seite mit dieser gekoppelt ist, ist ein einfacher wxPython GUI, die ein Textfeld stellt den Namen der Datei, die Sie in Innerhalb dieses GUI abrufen mögen zu schreiben. Es gibt eine wx.Timer, die alle 50 Millisekunden eine Methode ausführt. Dies blockiert meinen FTP-Dateiabruf. Ich finde, dass, weil der Hauptfaden aufgebraucht ist, das Protokoll, das die Daten empfängt, Schluckauf hat. Wenn Sie sich fragen, warum ich dieses Setup habe, simuliere ich den Anwendungsfall für ein viel größeres Projekt.

Mein Versuch, dies zu lösen, war, deferToThread auf den spezifischen Punkt zu verwenden, wenn ich eine Datei abrufen muss. Beim Drucken des aktuellen Threads stelle ich jedoch fest, dass das Protokoll, das die Daten empfängt, im Hauptthread ausgeführt wird. Das ist das Problem, das ich versuche zu lösen. Jede Hilfe wird sehr geschätzt.

Mein Client-Code:

#!/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

Sie können nicht deferToThread(function_that_uses_twisted_apis). Twisted APIs sind fast alle nicht Thread-sicher. Sie müssen sie nur im Reactor-Thread verwenden (die Ausnahmen sind einige Thread-Scheduling-bezogene APIs).

Stattdessen werden Sie Ihren Sperrcode loswerden. Setzen Sie es in einen anderen Thread, einen anderen Prozess, oder schreiben Sie es neu, um nicht blockierend zu sein.