python - Python でのマルチスレッド ソケット プログラミング

Question

Pythonを使用してクライアント/サーバーソケットプログラミングに取り組んでいます。問題なく設定できましたが、スレッドがあると途方に暮れます。その部分を完成させる方法がわかりません。Pythonのドキュメントも見ましたが、期待していたものが見つかりませんでした。

私が達成しようとしているのは、サーバーがクライアントからの接続を受け入れ、ファイルの作成のためにディレクトリを監視していることです。両方を並行して実行しようとしていますが、うまくいきませんでした。

#!/usr/bin/python           # This is server.py file

import socket               # Import socket module
import sys,os
import time
from watchdog.observers import Observer
from watchdog.events import PatternMatchingEventHandler

class MyHandler(PatternMatchingEventHandler):
    patterns = ["*.challenger", "*.challenger"]

    def process(self, event):
        """
        event.event_type
            'modified' | 'created' | 'moved' | 'deleted'
        event.is_directory
            True
        event.src_path
            /home/abcd/Maildir/new/
        """
        # the file will be processed there
        print event.src_path, event.event_type  # print now only for degug

    def on_modified(self, event):
        self.process(event)
        flag = '1'
        print flag

    def on_created(self, event):
        self.process(event)

    def on_any_event(self,event):
        if event.event_type == 'created':
            send()


s = socket.socket()         # Create a socket object
host = socket.gethostname() # Get local machine name
port = 12346                # Reserve a port for your service.
s.bind((host, port))        # Bind to the port
flag = '0'
s.listen(5)                 # Now wait for client connection.
while True:
    c, addr = s.accept()     # Establish connection with client.
    print 'Got connection from', addr

#   c.send('Thank you for connecting')
#   c.send(flag)        

    if __name__ == '__main__':

       args = sys.argv[1:]
       observer = Observer()
       observer.schedule(MyHandler(), path='/home/abcd/Maildir/new')
       observer.start()

    try:
        while True:
          time.sleep(1)

    except KeyboardInterrupt:
         observer.stop()

observer.join()


c.send(flag)
c.close()                # Close the connection

スレッドではまったく機能しなかったため、投稿したコードにはスレッドがありません。

ウォッチドッグを使用して /home/abcd/Maildir/new を監視し、作成された新しい電子メール ファイルを監視しています。作成された場合は、電子メールに含まれているクライアントに確認 (この場合はフラグ) を送信したいと考えています。受け取った。ソケット接続とウォッチドッグは単体ではうまくいきましたが、連携しない理由がわかりません。:(

これらのメソッドを異なるスレッドに配置して、並列に実行するにはどうすればよいですか? ご協力いただきありがとうございます。

編集:クレイグの入力後のコード:

import threading
import socket               # Import socket module
import sys,os
import time
from watchdog.observers import Observer
from watchdog.events import PatternMatchingEventHandler

def send_message(conn, flag):
    conn.send(flag)


class MyHandler(PatternMatchingEventHandler):
    patterns = ["*.challenger", "*.challenger"]

    def __init__(self, conn, *args, **kwargs):
        super(MyHandler, self).__init__(*args, **kwargs)
        self.conn_list = conn

    def process(self, event):
        """
        event.event_type
            'modified' | 'created' | 'moved' | 'deleted'
        event.is_directory
            True
        event.src_path
            /home/abcd/Maildir/new/
        """
        print "I reached here too"
        time.sleep(5)
        print event.src_path, event.event_type  # print now only for degug

    def on_created(self, event):
        flag = '1'
        threads = [threading.Thread(target=send_message, args=(conn, flag)) for conn in self.conn_list]
        for t in threads:
            t.start()
        print "on_created"
        self.process(event)
        for t in threads:
            t.join()


from collections import deque
conn_list = deque()

if __name__ == '__main__':
    s = socket.socket()         # Create a socket object
    host = socket.gethostname()  # Get local machine name
    port = 12346                # Reserve a port for your service.
    s.bind((host, port))        # Bind to the port
    flag = '0'
    s.listen(5)                 # Now wait for client connection.
    observer = Observer()
    observer.schedule(MyHandler(conn_list), path='/home/abcd/Maildir/new/')
    observer.start()
    print "Before True"
while True:
     try:
        c, addr = s.accept()     # Establish connection with client.
        print 'Got connection from', addr
        conn_list.append(c)
        time.sleep(1)
        print "I started Observer"
     except KeyboardInterrupt:
        observer.stop()
        while conn_list:
            conn_list.pop().close()
        print "Connections closed"
        break

observer.join()

編集2：

クライアントからサーバーに電子メールを送信すると、最初は正しい結果が得られますが、もう一度接続すると、出力が奇妙になります (「メッセージ内」が 2 回出力され、フラグ値も 2 回出力され、パイプの破損エラーが発生します)

Craig の入力後に Server.py を変更しました。

import threading
import socket               # Import socket module
import sys,os
import time
from watchdog.observers import Observer
from watchdog.events import PatternMatchingEventHandler


def send_message(conn, flag):
    print "In message"
    conn.send(flag)
    print "flag"+flag

class MyHandler(PatternMatchingEventHandler):
    patterns = ["*.challenger", "*.challenger"]

    def __init__(self, conn, *args, **kwargs):
        super(MyHandler, self).__init__(*args, **kwargs)
        self.conn_list = conn

    def process(self, event):
        time.sleep(5)
        print "In process"
        print event.src_path, event.event_type  # print now only for degug

    def on_created(self, event):
        flag = '1'
        print "before process event"
        self.process(event)
        print "after process, before thread target"
        threads = [threading.Thread(target=send_message, args=(conn, flag)) for$
        flag = '0'
        for t in threads:
            t.start()
#        print "Before process(Event)"
#        self.process(event)
        print "after process event"
        for t in threads:
            t.join()

s = socket.socket()         # Create a socket object
host = socket.gethostname()  # Get local machine name
port = 12345                # Reserve a port for your service.
s.bind((host, port))        # Bind to the port
flag = '0'
s.listen(5)                 # Now wait for client connection.
print "flag before: "+flag

from collections import deque
conn_list = deque()

if __name__ == '__main__':
    observer = Observer()
    observer.start()
    args = sys.argv[1:]
    observer.schedule(MyHandler(conn_list), path='/home/abcd/Maildir/new')
while True:
    try:
        c, addr = s.accept()     # Establish connection with client.
        print 'Got connection from', addr
        flag = '0'
        print flag
        conn_list.append(c)

    except KeyboardInterrupt:
        observer.stop()
        while conn_list:
            conn_list.pop().close()
        print "Connections closed"
        break

observer.join()

出力は次のとおりです。

Got connection from ('72.123.27.223', 39844)
0
before process event
In process
/home/abcd/Maildir/new/1425403821.V801I2ac232cM275759.challenger created
after process, before thread target
In message
In message
flag1In message
 after process event

 flag1
Exception in thread Thread-7:
Traceback (most recent call last):
  File "/usr/lib/python2.7/threading.py", line 810, in __bootstrap_inner
    self.run()
  File "/usr/lib/python2.7/threading.py", line 763, in run
    self.__target(*self.__args, **self.__kwargs)
  File "/home/paras/server42.py", line 11, in send_message
    conn.send(flag)
error: [Errno 32] Broken pipe

ここに私の client.py があります - 私がやろうとしているのは、クライアントからサーバーへの電子メールの遅延をチェックし、サーバーからの応答を取得することだけです。

#!/usr/bin/python           # This is client.py file
import sys,smtplib
import socket               # Import socket module
import threading,time
import urllib
import random
import datetime
#class myThread(threading.Thread):
#    def __init__ (self,name):
#        threading.Thread.__init__(self)
#   self.name=name

#    def run(self):
#        connection()    
#        print "Starting"+self.name

def connection():
    s = socket.socket()         # Create a socket object
    host = socket.gethostbyname('server.edu') # Get local machine name
    port = 12345                # Reserve a port for your service.
    try:
        s.connect((host, port))
#        print s.recv(1024)
        t1 = datetime.datetime.now()
        sendmail()
    t2 = datetime.datetime.now()
        print str(t1),str(t2) 
        print "Time taken to send e-mail from client to server: "+str(t2-t1)
        print "came back"
        flag = s.recv(1024)
        print flag
        if (flag=='1'):
            t3=datetime.datetime.now()   
        print "Time taken to receive response from server: "+str(t3-t2)
        s.close                     # Close the socket when done
    except KeyboardInterrupt:
        s.close

def sendmail():
    fromaddr = 'xyz@gmail.com'

    toaddrs = 'email@server.edu'
    url = "http://www.google.com"
    seq = str(random.randint(1,9))
    msg = 'URL: '+ url  + '\n'+'SEQ:'+seq+'\n'
    print "In sendmail"
#print msg

# Credentials (if needed)
    username = 'xyz@gmail.com'
    password = 'somepassword'
#The actual mail send
    server = smtplib.SMTP('smtp.gmail.com:587')
#server = smtplib.SMTP('localhost')
    server.starttls()
    server.login(username,password)
    server.sendmail(fromaddr, toaddrs, msg)
    server.quit()
    print "email sent"
    return

if __name__ == "__main__":
    connection()

私を助けてください。ありがとう。