비 차단 클라이언트 FTP 검색 및 쓰기

Question

Twisted를 사용하여 FTP 테스트 서버 및 클라이언트를 구축 중입니다. 서버가 잘 돌아갑니다. 기본적으로 Twisted ftpserver.py 예제와 같습니다. 클라이언트는 파일 검색 및 작성 중 일부 블로킹 문제가있는 곳입니다. 내가 몇 가지 빠른 트위스트 스레딩 유틸리티를 통해 그것을 해결하기 위해 노력했지만 아무 소용이 없습니다. 여기

내 서버입니다 :

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

클라이언트 쪽이와 결합, 당신이 검색 할 파일의 이름을 작성하는 텍스트 상자를 표시하는 간단한 wxPython에의 GUI 인이 GUI 내에서. 메서드는 50 밀리 초마다 실행되는 wx.Timer이 있습니다. 이것은 내 FTP 파일 검색을 차단하는 것입니다. 주 스레드가 사용 중이기 때문에 데이터를 수신하는 프로토콜에 딸꾹질이 발생했습니다. 내가 왜이 설정을 가지고 있는지 궁금해하신다면 훨씬 더 큰 프로젝트의 유스 케이스를 시뮬레이션하고 있습니다.

내 문제를 해결하기 위해 파일을 검색해야 할 때 특정 지점에서 deferToThread을 사용했습니다. 그러나 현재 스레드를 인쇄하여 데이터를 수신하는 프로토콜이 주 스레드에서 실행 중임을 알았습니다. 이것이 내가 풀려고하는 문제입니다. 어떤 도움이라도 대단히 감사합니다.

내 클라이언트 코드 :

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

당신은 deferToThread(function_that_uses_twisted_apis) 수 없습니다. 꼬인 API는 거의 모든 스레드로부터 안전합니다. 반응기 스레드에서만 사용해야합니다 (예외는 몇 가지 스레드 스케줄링 관련 API입니다).

대신 차단 코드를 제거하십시오. 을 다른 스레드 또는 다른 프로세스에 넣거나 비 블로킹으로 다시 작성하십시오.