Sockets/select: how to deal with multiple connections in threads?
Irmen de Jong
I have something that works for now, but I'd like your advice & ideas.
I have a socket server that has to deal with multiple concurrent connections.
(FYI, the SocketServer module is not appropriate).
What I'm currently doing is the following:
main loop uses select() on the current connection list.
if a new connection arrives, add this new socket to the connection list.
if data is available on a socket, spawn a new thread to read the data.
There are some subtleties:
If I don't remove the socket from the connection list while the thread
is reading the data, the select() in the main loop keeps on returing
with that socket (probably to be expected, because the thread hasn't
read all data yet, so the socket is still ready for reading).
So what I do, just before spawning the thread, I remove the socket
from the connection list. The thread adds it back to the list when
it has read all data. Next time around, the select() in the main loop
will also wait on that socket again.
Problems arise, because even though the thread has added the socket
back to the connection list, the select() in the main loop is still
waiting on the *old* list, that hasn't got the socket! My app blocks!
Adding a timeout to the select fixes things, but introduces an
unwanted delay between socket activity (only after ~delay/2 the
select() times out and reenters the loop with the new connection list).
My solution is currently to have *another* server socket, and a
loopback socket to that socket. The select() will wait on this 'signal'
server socket too. The thread will write a character to the loopback
socket, and so the select() in the main loop breaks and my cycle continues.
I'm not too happy with this solution. Who has a better idea?
What is usually done to abort a select()? (has to work on Windows too)
Thanks!!!
Irmen de Jong
PS example code below:
#! /usr/bin/env python
import sys
from socket import *
import select
from threading import Thread
import signal
def read(c,conns,s):
print 'READING FROM',c.getpeername()
m=''
while len(m)<10:
d=c.recv(10)
if not d:
print 'connection lost'
return
else:
m+=d
print ' GOT :',m
conns.append(c)
s.send("!") # signal to break select()
def main():
port = int(eval(sys.argv[1]))
# create server socket 's'
s = socket(AF_INET, SOCK_STREAM)
s.bind(('', port))
s.listen(1)
# create signal socket to break select()
ssock=socket(AF_INET,SOCK_STREAM)
ssock.bind(('',port+1))
ssock.listen(1)
sigsock=socket(AF_INET,SOCK_STREAM)
sigsock.connect(ssock.getsockname())
sigssock,foo = ssock.accept()
conns=[s,sigssock]
while 1:
ins,outs,exs=select.select(conns,[],[])
if ins:
print 'GOT INS',ins
if sigssock in ins:
print 'SIGNAL!'
ins.remove(sigssock)
sigssock.recv(10) # ignore any data
if s in ins:
conn, addr = s.accept()
print 'connected by', addr
conns.append(conn)
ins.remove(s)
for c in ins:
conns.remove(c)
Thread(target=read,
args=(c,conns,sigsock)).start()
if __name__=='__main__':
main()
Irmen de Jong
> I have something that works for now, but I'd like your advice & ideas.
To answer myself; I got some suggestions by email:
- use asyncore / Medusa and forget about threading
- don't use a thread on each request, instead, create
a thread and let it live. Let the main thread deal *only*
with accept()-ing new connections.
Because my software (Pyro) currently is structured around threads,
I'm trying the second suggestion. First results look promising :-)
Irmen
Donn Cave
...
| My solution is currently to have *another* server socket, and a
| loopback socket to that socket. The select() will wait on this 'signal'
| server socket too. The thread will write a character to the loopback
| socket, and so the select() in the main loop breaks and my cycle continues.
|
| I'm not too happy with this solution. Who has a better idea?
| What is usually done to abort a select()? (has to work on Windows too)
It's a nuisance, but in principle I think your design is not a bad one,
to make I/O the common base of your event handling. If you can find
a way around it that's more convenient, that's great, but I don't think
you'll find a more powerful and portable event dispatching architecture.
Donn Cave, do...@drizzle.com
Irmen de Jong
> It's a nuisance, but in principle I think your design is not a bad one,
> to make I/O the common base of your event handling. If you can find
> a way around it that's more convenient, that's great, but I don't think
> you'll find a more powerful and portable event dispatching architecture.
I was recommended to use asyncore/Medusa (that doesn't use threads at all).
But since my system (Pyro) is currently structured around threads, that
would require a major rewrite.
So I optimized my thread solution: don't use a thread per request, just
spawn a new thread and let it live. The main loop only handles new
accept()s. First results show a speed increase of up to 2 times :)
Irmen
J.Jacob
I have been using threads combined with select, maybe this is of some help?
--------------------------------------------------------------------------
import threading
import select
from Tkinter import *
def __init__(self, ...
# Set up the thread to do asynchronous I/O, more threads possible
self.listening = 1
self.thread1 = threading.Thread(target=self.listenThread)
self.thread1.start()
# Set up stop for threads
self.master.protocol('WM_DELETE_WINDOW', self.stopThreads)
def listenThread(self):
"Remember that the thread has to yield control."
while self.listening:
is_readable = [self.cserver.socket]
is_writable = []
is_error = []
r, w, e = select.select(is_readable, is_writable, is_error, 1.0)
if r:
print 'handling request', len(r), len(w), len(e)
self.cserver.handle_request()
def stopThreads(self):
self.listening = 0
sys.exit(1)
--------------------------------------------------------------------------
A help wanted add for a photo journalist asked the rhetorical question:
If you found yourself in a situation where you could either save
a drowning man, or you could take a Pulitzer prize winning
photograph of him drowning, what shutter speed and setting would
you use?
-- Paul Harvey
Chris Reay
>
> I was recommended to use asyncore/Medusa (that doesn't use threads at all).
> But since my system (Pyro) is currently structured around threads, that
> would require a major rewrite.
>
> So I optimized my thread solution: don't use a thread per request, just
> spawn a new thread and let it live. The main loop only handles new
> accept()s. First results show a speed increase of up to 2 times :)
>
> Irmen
Hi Irmen
My own multiple server class is unthreaded. It has separate methods
for accepting new connections, for reading, for writing, and for
closing "dead" connections. The reading method uses select with a
timeout of 0 ie:
readList, w, e = select(self.clientList, [], [], 0)
The method then reads each client's data completely, one by one. This
method seems to work quickly, simply, and well.
Hth, fwiw
Chris
Kragen Sitaker
> main loop uses select() on the current connection list.
> if a new connection arrives, add this new socket to the connection list.
> if data is available on a socket, spawn a new thread to read the data.
My suggestions:
First, use asyncore. Second, don't accept() or read() or write() or
send() or recv() data in any thread but the thread that runs the
select() loop --- when data is available on a socket, don't spawn a
new thread to read it, just read it (and spawn a new htread if
necessary to handle it, or pass it to an existing thread).