Python : How to close a UDP socket while is waiting for data in recv?

Question

let's consider this code in python:

import socket
import threading
import sys
import select


class UDPServer:
    def __init__(self):
        self.s=None
        self.t=None
    def start(self,port=8888):
        if not self.s:
            self.s=socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            self.s.bind(("",port))
            self.t=threading.Thread(target=self.run)
            self.t.start()
    def stop(self):
        if self.s:
            self.s.close()
            self.t.join()
            self.t=None
    def run(self):
        while True:
            try:
                #receive data
                data,addr=self.s.recvfrom(1024)
                self.onPacket(addr,data)
            except:
                break
        self.s=None
    def onPacket(self,addr,data):
        print addr,data


us=UDPServer()
while True:
    sys.stdout.write("UDP server> ")
    cmd=sys.stdin.readline()
    if cmd=="start\n":
        print "starting server..."
        us.start(8888)
        print "done"
    elif cmd=="stop\n":
        print "stopping server..."
        us.stop()
        print "done"
    elif cmd=="quit\n":
        print "Quitting ..."
        us.stop()
        break;

print "bye bye"

It runs an interactive shell with which I can start and stop an UDP server. The server is implemented through a class which launches a thread in which there's a infinite loop of recv/onPacket callback inside a try/except block which should detect the error and the exits from the loop. What I expect is that when I type "stop" on the shell the socket is closed and an exception is raised by the recvfrom function because of the invalidation of the file descriptor. Instead, it seems that recvfrom still to block the thread waiting for data even after the close call. Why this strange behavior ? I've always used this patter to implements an UDP server in C++ and JAVA and it always worked.

I've tried also with a "select" passing a list with the socket to the xread argument, in order to get an event of file descriptor disruption from select instead that from recvfrom, but select seems to be "insensible" to the close too.

I need to have a unique code which maintain the same behavior on Linux and Windows with python 2.5 - 2.6.

Thanks.

Answer 1

The usual solution is to have a pipe tell the worker thread when to die.

1. Create a pipe using os.pipe. This gives you a socket with both the reading and writing ends in the same program. It returns raw file descriptors, which you can use as-is (os.read and os.write) or turn into Python file objects using os.fdopen.

2. The worker thread waits on both the network socket and the read end of the pipe using select.select. When the pipe becomes readable, the worker thread cleans up and exits. Don't read the data, ignore it: its arrival is the message.

3. When the master thread wants to kill the worker, it writes a byte (any value) to the write end of the pipe. The master thread then joins the worker thread, then closes the pipe (remember to close both ends).

P.S. Closing an in-use socket is a bad idea in a multi-threaded program. The Linux close(2) manpage says:

It is probably unwise to close file descriptors while they may be in use by system calls in other threads in the same process. Since a file descriptor may be re-used, there are some obscure race conditions that may cause unintended side effects.

So it's lucky your first approach did not work!

Answer 2

This is not java. Good hints:

• Don't use threads. Use asynchronous IO.
• Use a higher level networking framework

Here's an example using twisted:

from twisted.internet.protocol import DatagramProtocol
from twisted.internet import reactor, stdio
from twisted.protocols.basic import LineReceiver

class UDPLogger(DatagramProtocol):    
    def datagramReceived(self, data, (host, port)):
        print "received %r from %s:%d" % (data, host, port)


class ConsoleCommands(LineReceiver):
    delimiter = '\n'
    prompt_string = 'myserver> '

    def connectionMade(self):
        self.sendLine('My Server Admin Console!')
        self.transport.write(self.prompt_string)

    def lineReceived(self, line):
        line = line.strip()
        if line:
            if line == 'quit':
                reactor.stop()
            elif line == 'start':
                reactor.listenUDP(8888, UDPLogger())
                self.sendLine('listening on udp 8888')
            else:
                self.sendLine('Unknown command: %r' % (line,))
        self.transport.write(self.prompt_string)

stdio.StandardIO(ConsoleCommands())
reactor.run()

Example session:

My Server Admin Console!
myserver> foo  
Unknown command: 'foo'
myserver> start
listening on udp 8888
myserver> quit