Immer broken pipe bei TCP?

[Alfons Mittelmeyer]

Habe jetzt Server und Client umgeschrieben. Das mit den Steuerzeichen wird nicht benötigt, es genügt: if len data == 0

client.py:

Code: Alles auswählen

import socket

HOST, PORT = "localhost", 9999

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect((HOST, PORT))

while True:
    data = input(">")
    if len(data) == 0: break
    sock.sendall(bytes(data + "\n", "utf-8"))
sock.close()

server.py:

Code: Alles auswählen

import socketserver

HOST, PORT = "localhost", 9999

def process_data(data):
    print(str(data,"utf-8"))

class MyTCPHandler(socketserver.StreamRequestHandler):

    def handle(self):

        while True:       
            data = self.rfile.readline()
            if len(data) == 0: break
            process_data(data)
        
        print("Connection closed")
       
class ThreadedTCPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
    allow_reuse_address = True

server = ThreadedTCPServer((HOST, PORT), MyTCPHandler)
server.serve_forever()

[Alfons Mittelmeyer]

Also die Lösung wäre das: wenn eine Anwendung senden will, muss sie erst zum Router eine Liste der Message Ids (Strings) senden. Der gibt ihr dann zu den Strings Integer Werte zurück. Die muss dann die Anwendung zum Senden verwenden. Und die Anwendung, die Messages erhalten will, bekommt vom Router auch die Liste der MessageIds. Dann muss die zu routende Message auch gar nicht mehr die Message ID als String enthalten.

[Alfons Mittelmeyer]

Ich denke, Alfons möchte vom Prinzip her eigentlich Sessions implementieren. Er will hier aber vermutlich mal wieder im typischen Alfons-Stil seinen eigenen Weg gehen, anstatt sich auf bewährte Lösungen zu verlassen...

Ich benutze doch nur eine bewährte Lösung. Es ist socketserver.ThreadingMixIn. Oder meinst Du dass das keine bewährte Lösung ist?

[Alfons Mittelmeyer]

@Alfons Mittelmeyer: was soll der Server überhaupt können? Da TCP schon von sich aus reliable ist, braucht man nicht alles mit einem sinnfreien Ok bestätigen. Was gesendet wird, kommt auch an, oder man bekommt ein broken Pipe. Genauso die Sache mit dem loop. Nromalerweise sendet man Befehle, bis man keine Lust mehr hat und sendet dann entweder ein QUIT o.ä. oder man schließt einfach die Verbindung, dann ist auch klar, dass da wohl keine weiteren Befehle mehr kommen.

Was der Server können soll?

Es handelt sich nicht um einen Server und Clients, sondern um einen Router und Clients des Routers. Der Router dient dazu, dass Clients Messages miteinander austauschen können, ohne sich gegenseitig zu kennen. Es geht dabei um schnellen Messageaustausch zwischen einer überschaubaren Anzahl von Clients, also nicht dass tausende von Clients nur jeweils kurzzeitig einen Request wollen, sondern eher um eine verteilte Applikation in der vielleicht nur fünf Clients schnell miteinander kommunizieren können.

Dazu soll nicht bei jeder Message ein neuer Verbindungsaufbau erfolgen, sondern es sollen Sessions sein, bei denen die Verbindungen so lange bestehen, bis sie getrennt werden. Die Kommunikations soll nicht nach dem Schema Request mit Warten auf Response erfolgen, sondern einfach die Message senden, ohne Response. Zehn Bytes nach Amerika senden und dann auf ein OK warten bis man die nächste Message sendet, soll nicht sein. Wenn es Antworten gibt, dann sollen diese mit anderen dafür vorgesehenen Messages erfolgen.

Requirements für Verbindungen:
- sichere, dauerhafte Verbindungen mit Versenden von Messages ohne Warten auf Response

Requirements für den Router:
- die Messages haben eine Message ID. Der Router soll eine empfangene Message an alle Clients versenden, welche sich dafür registriert haben, Messages mit dieser Message ID zu empfangen
- der Router soll schnell sein und soll die Message nicht erst parsen, um daraus die Message ID zu extrahieren
- da es sich bei bei den Messages um mit JSON serialisierte Strings als Byte Array handelt, gedacht einer Python Funktion als Parameter zu übergeben, genügt es, die Messages mit readline zu lesen. Mehrmaliges Lesen in verschiedene Buffer ist daher nicht nötig.

Wenn man allerdings nicht mit readline arbeiten würde, könnte man auch Urlaubsbilder im JPEG Format übertragen oder MP3 Dateien.
Aber ich denke, wenn jemand so etwas tun will, soll er nicht den Router damit belasten, sondern über den Router von anderen Clients erfragen, wie er das dann ohne den Router tun kann. Und so etwas würde man dann wohl am Besten mit bereits existerenden Lösungen tun (weil sich da einige beschwert hatten, dass ich nicht bereits existierende Lösungen verwenden würde)

Lösung für die Verbindungen:

- Sicherheit: Verwendung von TCP
- Dauerhaft: Verbindung pro Client durch Verwendung von socketserver.ThreadingMixIn und durch eine Schleife, bis der Client die Verbindung beendet:

Code: Alles auswählen

while True:       
    message = self.rfile.readline()
    if len(message) == 0: break

- Ohne Response sondern über Messages zum Client: die Clients haben einen Server. Pro Thread von socketserver.ThreadingMixIn wird eine socketVerbindung zum Client eröffnet.
Die erste Message eines Clients an den Router ist seine IP und Portadresse und der Router baut dann eine Socket Verbindung zum Server des Clients auf:

Code: Alles auswählen

message = str(self.rfile.readline().strip(), "utf-8")
json_message = json.loads(message)
receiver_address=(json_message[0],json_message[1])
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client_socket.connect(receiver_address)

Lösung für den Router:

- Versenden von Messages an Clients: mit dem Routerkommando Message REGISTER senden Clients eine Liste der Message IDs an den Router, für die sie Messages empfangen wollen
- Message IDs ohne parsen: mit dem Routerkommando ID_REQUEST senden Clients eine Liste der Message IDs an den Router, für die sie Messages senden wollen. Der Router gibt den Clients dann eine Tupelliste zurück mit den Message IDs der Clients und vom Router generierten Message IDs dafür, welche die Clients für diese Messages an den Router verwenden und der Message voranstellen sollen. Die vom Router generierten Message IDs haben eine feste Länge – zur Zeit von vier Bytes. Und so erhält der Router dann die Message ID von einer an ihn gesendeten Message:

Code: Alles auswählen

message_id = message[0:4] # first 4 bytes are the message id

- Lesen ohne Queues und Bufferung: denn es genügt readline:

Code: Alles auswählen

message = self.rfile.readline()

Nicht alle Messages erfolgen ohne Response. Routerkommandos etwa gibt es nach dem Prinzip einfach Senden oder mit Warten auf Response:
- Die Message REGISTER, mit der Clients Message IDs versenden, für die sie Messages erhalten wollen, ist ohne Response
- Die Message ID_REQUEST allerdings ist mit Response. Wäre sie es nicht und eine Anwendung würde unmittelbar nach ID_REQUEST bereits senden, bevor der Router geantwortet hat, dann würde die Message ins Leere gehen und verworfen werden. Daher wartet ID_REQUEST auf die Antwort des Routers. Und unmittelbar folgene Sendebefehle erreichen ihr Ziel, vorausgesetzt, es haben sich bereits Empfänger dafür registriert.

[Sirius3]

@Alfons Mittelmeyer: Du willst also einen simplen Message-Broker mit Publish-Subscribe-Pattern nachprogrammieren. Soll das nun eine Übung sein um Threading, Sockets, oder Python allgemein zu lernen? Wenn es eine Übung zu Socket-Programmierung sein soll, würde ich mir an Deiner Stelle erst mal ein Buch besorgen, das sich nur damit beschäftigt (muß nicht Python sein, weil die Prinzipien sind sprachunabhängig). Ich habe Socket-Programmierung mit einem Buch aus den 80ern gelernt. Allgemeine Pythonbücher oder irgendwelche Quellen aus dem Internet sind mit hoher Wahrscheinlichkeit so schlecht, dass die mehr Schaden anrichten, als irgendetwas nützen. Dann schau Dir an, wie andere Message-Broker Protokolle entworfen haben. Dann versuche, den Server sowohl mit Threading, als auch asynchron zu entwerfen, damit Du die Unterschiede kennenlernst. Die meiste Zeit wirst Du mit Verstehen der Konzepte und dem Designen eigener Konzepte verwenden und nicht mit programmieren.

[Alfons Mittelmeyer]

Dann nennt sich das also ein simpler Messagebroker, den ich programmiert habe und zwar asynchron mit threading mixin und offenen Verbindungen. Gedacht für schnellen Messageaustausch weniger Subscriber.

Benützte Kommandomessages waren:

INFO
ECHO
ID_REQUEST
REGISTER
UNREGISTER
UNREGISTER_ALL

INFO und ECHO sind unwichtig. Eine INFO Message sendet der Messagebroker dem Subscriber, nachdem die Verbindung herhestellt wurde. Der Subscriber kann sie ausgeben oder einfach ignorieren. ECHO sendet die empfange Message an den Subscriber zurück. Kann man verwenden für Testzwecke. Die eigentlichen Kommandos sind ID_REQUEST, REGISTER, UNREGISTER und UNREGISTER_ALL. Die letzteren drei Kommandos sollte ich wohl dann beser nennen: SUBSCRIBE, UNSUBSCRIBE, UNSUBSCRIBE_ALL

MIT SUBSCRIBE gibt der Subscriber dem Messagebroker eine Liste, welche Messages er empfangen will. Ob man UNSUBSCRIBE braucht, weiss ich nicht, damit kann mittels einer Liste wieder Messages wegnehmen.
UNSUBSCRIBE_ALL dagegen ist wichtig. Das kann der Subscriber dem Messagebroker senden, wenn er herunterfährt. Allerdings führt der Messagebroker das auch selber aus, wenn der Subscriber die Verbindung löst.

Das habe ich dann auch mit einem kleinen Chatprogrammm getestet:

Code: Alles auswählen

# router and client address
ROUTERHOST, ROUTERPORT = "localhost", 9999
MY_IP, MY_PORT = "localhost", 0

# starting client
import client as myclient
client = myclient.Client((ROUTERHOST, ROUTERPORT),(MY_IP, MY_PORT))

# configure router and client
client.receive_ids(("CHAT","WHO_IS_ONLINE","IS_ONLINE")) # tell router, which messages shall be received
client.send_ids(("CHAT","WHO_IS_ONLINE","IS_ONLINE")) # tell client and router, which messages shall be sent

# install reception functions ===========================

name = input("Name: ") # enter name

# define function for reception of message "CHAT": print name and message, if it's not the sender self
def tell(message):
	if message[0] != name: print(message[0]+": "+message[1])

client.do_receive("CHAT",tell)

# define function for reception of message "WHO_IS_ONLINE": send message "IS_ONLINE" if it'ts not the sender of "WHO_IS_ONLINE"
def who_is_online(message):
    if message != name: client.send("IS_ONLINE",(message,name))

client.do_receive("WHO_IS_ONLINE",who_is_online)
	
# define function for reception of message "IS_ONLINE": print("online:",name) if it is the sender, which had sent "WHO_IS_ONLINE
def is_online(message):
    if message[0] == name: print("online: "+message[1])

client.do_receive("IS_ONLINE",is_online)

# chat programm  ==============================
client.send("CHAT",(name,"Nice to meet you")) # greeting
client.send("WHO_IS_ONLINE",name) # ask, who is online

# chat with participiants
while True:
	a = input()
	if a == "": break
	client.send("CHAT",(name,a))

# say good bye, when leaving
client.send("CHAT",(name,"Good bye"))

# wait a little bit, so that the message before will be delivered before shutdown happens
# (shutdown has higher priority - and we don't want to offer a function send_vip, because otherwise some people would use it any time)
import time
time.sleep(0.1)

# proper shutdown - but ending the program would also be sufficient
client.shutdown()

Allerdings für einen Chat von 50 Teilnehmern ist asynchron mit offenen Verbindungen wohl nicht ganz das Richtige. Eine TCP Verbindung hat schon zwei Kanäle, nämlich einen Kanal für den Request und einen Rückkanal für den Response. Und das habe ich doppelt für zurück vom Broker zum Subscriber. Bei 50 Teilnehmern wäre das 200 Kanäle. Keine Ahnung ob das das Netzwerk und das Betriebssystem verkraften.

Für so einen Fall würde ich asynchron und synchron mischen. Asynchron mit Threading Mixin und mit Schließen der Verbindung für Steuerkommandos (ID_REQUEST, SUBSCRIBE). Synchron mit Schließen der Verbindung für Messages vom Publisher zu den Subscribern. Das geht dann zwar langsamer, belastet aber nicht das System mit vielen offenen Kanälen.

War mir noch eingefallen: statt time.sleep könnte man noch eine Funktion implementieren, zu warten bis alle Messages gesendet wurden. Aber nützt auch nichts, denn beim Messagebroker hat shutdown (UNSUBSCRIBE_ALL) auch eine höhere Priorität.
Man könnte, wenn man send_vip nicht will, dem shutdown noch eine Message mitgeben, damit diese noch mit hoher Priorität ausgeführt wird, bevor das Sytem herunterfährt. Bzw. man könnte auch bei shutdown noch alle Messages in der Queue mit hoher Priorität senden. Soll man das tun? Habe ich mir noch nicht überlegt. Aber dann könnten diese Messages andere zuvor überholen. Ich lasse das dann doch besser.

[DasIch]

Allerdings für einen Chat von 50 Teilnehmern ist asynchron mit offenen Verbindungen wohl nicht ganz das Richtige. Eine TCP Verbindung hat schon zwei Kanäle, nämlich einen Kanal für den Request und einen Rückkanal für den Response. Und das habe ich doppelt für zurück vom Broker zum Subscriber. Bei 50 Teilnehmern wäre das 200 Kanäle. Keine Ahnung ob das das Netzwerk und das Betriebssystem verkraften.

Ein asynchroner Ansatz ist nicht nur das Richtige, er ist sogar der optimale Ansatz um möglichst viele Verbindungen gleichzeitig aufrecht zu erhalten. Darüberhinaus hast du nicht mit vielen Verbindungen zu tun, 10.000 offene Sockets wären viel und immer noch kein Problem für dein Betriebssystem und erst recht nicht für dein Netzwerk.

[Alfons Mittelmeyer]

@DasIch Danke, das hatte ich nicht gewußt, was ein Betriebssystem verträgt. Dann brauche ich da also nichts zu ändern. Und für 10.000 Teilnehmer ist es ja wirklich nicht gedacht. Eventuell für eine Schulklasse von 30 Teilnehmern, denen man von einem Rechner aus PDF Dateien senden will oder für einen Chat von 50 Teilnehmern. Also nichts richtig World Wide Web mäßiges wie Facebook und Co.