Multithreading-Sockets in Python

Question

Ich versuche, einen blockierungsfreien UDP-Server aufzubauen, der verschiedene Ports abhört und Datenpakete bis zum Timeout empfängt. Leider kann ich die Client-Seite nicht ändern und UDP ist obligatorisch. Empfangen von Dateien funktioniert einwandfrei. Das Problem ist, dass das Erstellen der Arbeiter eine Blockierung ist. Ich möchte es nicht blockierend haben, also werden alle Arbeiter parallel gerufen. Außerdem sollte jeder Arbeiter in einer Schleife wie True laufen, aber das war auch blockiert.

Hier ist mein Code:

#!/usr/bin/env python 
from socket import * 
import sys 
import select 
import threading 
threads = [] 

def worker(port): 
     host="192.168.88.51" 
     s = socket(AF_INET,SOCK_DGRAM) 
     s.bind((host,port)) 
     addr = (host,port) 
     buf=128 
     data,addr = s.recvfrom(buf) 
     filename = str(port)+".data" 
     print str(port)+" received File:" 
     f = open(filename,'wb') 

     data, addr = s.recvfrom(buf) 
     try: 
      while(data): 
       f.write(data) 
       s.settimeout(1) 
       data,addr = s.recvfrom(buf) 
     except timeout: 
      f.close() 
      s.close() 
      print "File Downloaded" 

for i in range(1300,1305): 
    wrk = worker(i) 
    threads.append(wrk) 

Answer 1

Dies funktioniert wie von Ihnen vorgesehen, außer dass die Datei jedes Mal überschrieben wird, wenn neue Daten ohne Zeitüberschreitung gesendet werden. Timeout steht für Ende der gesamten Verbindung. Aber Sie können dies leicht umarbeiten, um Daten zu derselben Datei hinzuzufügen oder eine neue Datei zu erstellen oder alles zu tun, was Sie brauchen.

#! /usr/bin/env python 
from socket import AF_INET, SOCK_DGRAM 
import socket 
import threading 

class Server (threading.Thread): 
    def __init__ (self, host="192.168.88.51", port=123, bufsize=128): 
     threading.Thread.__init__(self) 
     self.host = host 
     self.port = port 
     self.bufsize = bufsize 
     self.done = threading.Event() 

    def opensock (self): 
     s = socket.socket(AF_INET, SOCK_DGRAM) 
     s.bind((self.host, self.port)) 
     s.settimeout(0.001) 
     return s 

    def run (self): 
     host = self.host 
     port = self.port 
     self.s = s = self.opensock() 
     print "Waiting for connection on", host+":"+str(port) 
     while not self.done.isSet(): 
      try: 
       data, addr = s.recvfrom(self.bufsize) 
       print "Connection from", addr 
       s.settimeout(1) 
       self.recvdata(data, s, addr) 
       s.settimeout(0.001) 
      except socket.timeout: pass 
      except: 
       raise 
     self.done.set() 
     s.close() 
     print "Server on '%s:%s' stopped!" % (host, port) 

    def recvdata (self, initdata, conn, addr): 
     bufsize = self.bufsize 
     filename = str(self.port)+".data" 
     print "Opening file", filename 
     f = open(filename, "wb") 
     print "Receiving & writingrest of data from", addr 
     data = initdata 
     while data and not self.done.isSet(): 
      f.write(data) 
      try: 
       data, addr = conn.recvfrom(bufsize) 
      except socket.timeout: break 
     f.close() 
     if self.done.isSet(): 
      print "Forcefully interrupted transmission" 
     else: 
      print "File Downloaded" 

    def stop (self): 
     self.done.set() 
     self.s.close() 

servers = [] 
for port in xrange(123, 150): 
    try: 
     s = Server(port=port) 
     s.start() 
     servers.append(s) 
    except Exception as e: 
     print e 

raw_input("Press enter to send data to one of ports for testing . . . ") 
import random 
a = servers[0].host 
p = random.choice(servers).port 
print "data will be sent to port '%s:%i'" % (a, p) 
k = socket.socket(AF_INET, SOCK_DGRAM) 
k.connect((a, p)) 
k.send("1234567890") 
k.send("asdfghjkl") 
k.send("0987654321") 
k.close() 
raw_input("Press enter to close the program . . . ") 

# Stop all servers: 
for s in servers: 
    s.stop() 

# Make sure all of them terminated: 
for s in servers: 
    s.join() 

Answer 2

Das tat es. Ich habe es selbst herausgefunden.

#!/usr/bin/env python 
from socket import * 
import sys 
import select 
import multiprocessing 

def worker(port): 
     print "started: "+str(port) 
     host="192.168.88.51" 
     s = socket(AF_INET,SOCK_DGRAM) 
     s.bind((host,port)) 
     addr = (host,port) 
     buf=128 
     data,addr = s.recvfrom(buf) 
     filename = str(port)+".jpg" 
     print str(port)+" received File:" 
     f = open(filename,'wb') 

     data, addr = s.recvfrom(buf) 
     try: 
      while(data): 
       f.write(data) 
       s.settimeout(1) 
       data,addr = s.recvfrom(buf) 
     except timeout: 
      f.close() 
      s.close() 
      print "File Downloaded" 

for i in range(1300,1305): 
    multiprocessing.Process(target=worker, args=(i,)).start()