How to prevent Brokenpipe error in my server script

Question

This is my server script to recieve and send messages to all clients

import socket
import threading

HEADER = 64
PORT = 5050
SERVER = socket.gethostbyname(socket.gethostname())
ADDR = (SERVER, PORT)
FORMAT = 'utf-8'
DISCONNECT_MESSAGE = "!DISCONNECT"

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(ADDR)

clients = set()
clients_lock = threading.Lock()

def handle_client(conn, addr):
    name = conn.recv(HEADER).decode(FORMAT)
    if name:
        name = int(name)
        msg_name = conn.recv(name).decode(FORMAT)
    print(f"[NEW CONNECTION] {msg_name} connected.")
    connection_message = f"{msg_name} connected."
    with clients_lock:
        for c in clients:
            if c != conn:
                message = connection_message.encode(FORMAT)
                msg_length = len(message)
                send_length = str(msg_length).encode(FORMAT)
                send_length += b' ' * (HEADER - len(send_length))
                c.sendall(send_length)
                c.sendall(message)

    with clients_lock:
        clients.add(conn)

    connected = True
    try:
        while connected:
            msg_length = conn.recv(HEADER).decode(FORMAT)
            if msg_length:
                msg_length = int(msg_length)
                msg1 = conn.recv(msg_length).decode(FORMAT)
                msg = f"{msg_name}: {msg1}"
                if msg1 == DISCONNECT_MESSAGE:
                    connected = False
                print(f"{msg}")
                with clients_lock:
                    for c in clients:
                        if c != conn:
                            message = msg.encode(FORMAT)
                            msg_length = len(message)
                            send_length = str(msg_length).encode(FORMAT)
                            send_length += b' ' * (HEADER - len(send_length))
                            c.sendall(send_length)
                            c.sendall(message)
                msg = f"You: {msg1}"
                message = msg.encode(FORMAT)
                msg_length = len(message)
                send_length = str(msg_length).encode(FORMAT)
                send_length += b' ' * (HEADER - len(send_length))
                conn.send(send_length)
                conn.send(message)

    finally:
        with clients_lock:
            clients.remove(conn)
            conn.close()

def start():
    server.listen()
    print(f"[LISTENING] Server is listening on {SERVER}")
    while True:
        conn, addr = server.accept()
        thread = threading.Thread(target=handle_client, args=(conn, addr))
        thread.daemon = True
        thread.start()
        print(f"[ACTIVE CONNECTIONS] {threading.activeCount() - 1}")

print("[STARTING] server is starting...")
start()

The problem is, if a server disconnects without sending the DISCONNECT_MESSAGE, i.e. I forcefully closed the program before it finishes, I get Brokenpipe error.

This is the client script:

import socket
import threading
import tkinter as tk

def returnname():
    def receiving():
        receiving = True
        while receiving:
            msg_length = client.recv(HEADER).decode(FORMAT)
            if msg_length:
                msg_length = int(msg_length)
                msg = client.recv(msg_length).decode(FORMAT)
                TEXTAREA.insert("end", msg)
                TEXTAREA.see("end")

    def send(msg):
        message = msg.encode(FORMAT)
        msg_length = len(message)
        send_length = str(msg_length).encode(FORMAT)
        send_length += b' ' * (HEADER - len(send_length))
        client.send(send_length)
        client.send(message)

    def sendmessage():
        mess = MESSAGEFIELD.get()
        MESSAGEFIELD.delete(0, "end")
        send(mess)

    def quitmessage():
        send(DISCONNECT_MESSAGE)
        exit()

    name = FIELD.get()
    FIELD.pack_forget()
    BUTTON.pack_forget()
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client.connect(ADDR)
    message = name.encode(FORMAT)
    msg_length = len(message)
    send_length = str(msg_length).encode(FORMAT)
    send_length += b' ' * (HEADER - len(send_length))
    client.send(send_length)
    client.send(message)
    thread = threading.Thread(target=receiving)
    thread.daemon = True
    thread.start()
    MESSAGEFIELD = tk.Entry(TOP)
    SENDBUTTON = tk.Button(TOP, text="Send", command=sendmessage)
    QUITBUTTON = tk.Button(TOP, text="Quit", command=quitmessage)
    TEXTAREA = tk.Listbox(TOP)
    SCROLLBAR = tk.Scrollbar(TOP)
    MESSAGEFIELD.pack()
    SENDBUTTON.pack()
    QUITBUTTON.pack()
    TEXTAREA.pack(side="left", expand=True, fill="both")
    SCROLLBAR.pack(side="right", fill="both")
    SCROLLBAR.config(command=TEXTAREA.yview)

HEADER = 64
PORT = 5050
FORMAT = 'utf-8'
SERVER = "IP ADDRESS OF SERVER"
ADDR = (SERVER, PORT)
DISCONNECT_MESSAGE = "!DISCONNECT"

TOP = tk.Tk()
FIELD = tk.Entry(TOP)
FIELD.insert(0, "Enter Name Here")
BUTTON = tk.Button(TOP, text="Send", command=returnname)
FIELD.pack(expand=True)
BUTTON.pack(expand=True)
TOP.mainloop()

Can anyone tell me if there is any way I can fix this issue?

Sorry if my explanation of the problem was bad. I am bad at explaining stuff.

Does this answer your question? [How to prevent BrokenPipeError when doing a flush in Python?](https://stackoverflow.com/questions/26692284/how-to-prevent-brokenpipeerror-when-doing-a-flush-in-python), [How to handle a broken pipe (SIGPIPE) in python?](https://stackoverflow.com/questions/180095/how-to-handle-a-broken-pipe-sigpipe-in-python) — Steffen Ullrich, Apr 19 '20 at 17:32
I tried that. But the problem here is, I need a way in case of an exception detrmine the client which is not connected and remove it from the clients list. I am not able to figure that out — Prem Raj, Apr 19 '20 at 19:12
I don't understand why you are not able to figure out the client and your only unspecific information is that you can't - without details. Broken Pipe happens when sending, you know which client is sending so you should be able to figure out which client it is if you capture the exception at the place where it can happen. — Steffen Ullrich, Apr 19 '20 at 19:32
Yeah... I got the idea about it just a few seconds before... I will fix it and post an answer here. Thanks for the support. — Prem Raj, Apr 19 '20 at 21:30

score 0 · Accepted Answer · answered Apr 20 '20 at 03:23

There was a bunch of things I had to do to make this work.

First, the Broken pipe can occur at two different places in the server code. First is when a new client connects and the server tries to send message New Client Connected to all the clients and second is when an existing client sends a message. So, we need to handle the exception at both the places.

So, we put try/except on both the blocks. where it says if c != conn.

Now, about how to handle the exception.

As I first thought, simply removing the client c from the list on clients will work, but the for loop, for c in clients will throw out runtime error as we try to modify the set clients during iteration.

I tried different methods to get over this problem, but this is the most efficient working method I got.

I changed clients from a set() to an empty list [ ]

Then I changed clients.add to clients.append

Then I changed the for loop to for c in range(len(clients)) and used clients[c] to access clients.

But when I tried this, I saw that the if statement if clients[c] != conn may throw out list index out of bounds error if the program tries to go over a non existant client after removal. So I put it too in a try/except block and let the program continue on exception.

for c in range(len(clients)):
        try:
            if clients[c] != conn:
                try:
                    message = connection_message.encode(FORMAT)
                    msg_length = len(message)
                    send_length = str(msg_length).encode(FORMAT)
                    send_length += b' ' * (HEADER - len(send_length))
                    clients[c].sendall(send_length)
                    clients[c].sendall(message)
                except:
                    clients.remove(clients[c])
        except:
            continue

The last problem was that even after removal of the client, the thread is still alive, so the active thread count returns more than the number of clients connected. So, instead of printing out the number of active connection as number of alive threads - 1, I print len(clients) + 1, + 1 because on connection of a new client, it prints this line before appending the client to list.

print(f"[ACTIVE CONNECTIONS] {len(clients) + 1}")

So, the entire program now is:

import socket
import threading

HEADER = 64
PORT = 5050
SERVER = socket.gethostbyname(socket.gethostname())
ADDR = (SERVER, PORT)
FORMAT = 'utf-8'
DISCONNECT_MESSAGE = "!DISCONNECT"

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(ADDR)

clients = []
clients_lock = threading.Lock()

def handle_client(conn, addr):
    name = conn.recv(HEADER).decode(FORMAT)
    if name:
        name = int(name)
        msg_name = conn.recv(name).decode(FORMAT)
    print(f"[NEW CONNECTION] {msg_name} connected.")
    connection_message = f"{msg_name} connected."
    with clients_lock:
        for c in range(len(clients)):
            try:
                if clients[c] != conn:
                    try:
                        message = connection_message.encode(FORMAT)
                        msg_length = len(message)
                        send_length = str(msg_length).encode(FORMAT)
                        send_length += b' ' * (HEADER - len(send_length))
                        clients[c].sendall(send_length)
                        clients[c].sendall(message)
                    except:
                        clients.remove(clients[c])

            except:
                continue

    with clients_lock:
        clients.append(conn)

    connected = True
    try:
        while connected:
            msg_length = conn.recv(HEADER).decode(FORMAT)
            if msg_length:
                msg_length = int(msg_length)
                msg1 = conn.recv(msg_length).decode(FORMAT)
                msg = f"{msg_name}: {msg1}"
                if msg1 == DISCONNECT_MESSAGE:
                    connected = False
                print(f"{msg}")
                with clients_lock:
                    for c in range(len(clients)):
                        try:
                            if clients[c] != conn:
                                try:
                                    message = msg.encode(FORMAT)
                                    msg_length = len(message)
                                    send_length = str(msg_length).encode(FORMAT)
                                    send_length += b' ' * (HEADER - len(send_length))
                                    clients[c].sendall(send_length)
                                    clients[c].sendall(message)
                                except:
                                    clients.remove(clients[c])
                        except:
                            continue
                msg = f"You: {msg1}"
                message = msg.encode(FORMAT)
                msg_length = len(message)
                send_length = str(msg_length).encode(FORMAT)
                send_length += b' ' * (HEADER - len(send_length))
                conn.send(send_length)
                conn.send(message)

    finally:
        with clients_lock:
            clients.remove(conn)
            conn.close()

def start():
    server.listen()
    print(f"[LISTENING] Server is listening on {SERVER}")
    while True:
        conn, addr = server.accept()
        thread = threading.Thread(target=handle_client, args=(conn, addr))
        thread.daemon = True
        thread.start()
        print(f"[ACTIVE CONNECTIONS] {len(clients) + 1}")

print("[STARTING] server is starting...")
start()

How to prevent Brokenpipe error in my server script

1 Answers1