Sending data through a socket from another thread does not work in Python

This is my 'game server'. It's nothing serious, I thought this was a nice way of learning a few things about python and sockets.

First the server class initialized the server. Then, when someone connects, we create a client thread. In this thread we continually listen on our socket.

Once a certain command comes in (I12345001001, for example) it spawns another thread.

The purpose of this last thread is to send updates to the client. But even though I see the server is performing this code, the data isn't actually being sent.

Could anyone tell where it's going wrong? It's like I have to receive something before I'm able to send. So I guess somewhere I'm missing something

#!/usr/bin/env python

"""
An echo server that uses threads to handle multiple clients at a time.
Entering any line of input at the terminal will exit the server.
"""

import select
import socket
import sys
import threading
import time
import Queue

globuser = {}
queue = Queue.Queue()

class Server:
    def __init__(self):
        self.host = ''
        self.port = 2000
        self.backlog = 5
        self.size = 1024
        self.server = None
        self.threads = []

    def open_socket(self):
        try:
            self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.server.bind((self.host,self.port))
            self.server.listen(5)
        except socket.error, (value,message):
            if self.server:
                self.server.close()
            print "Could not open socket: " + message
            sys.exit(1)

    def run(self):
        self.open_socket()
        input = [self.server,sys.stdin]
        running = 1
        while running:
            inputready,outputready,exceptready = select.select(input,[],[])

            for s in inputready:

                if s == self.server:
                    # handle the server socket
                    c = Client(self.server.accept(), queue)
                    c.start()
                    self.threads.append(c)

                elif s == sys.stdin:
                    # handle standard input
                    junk = sys.stdin.readline()
                    running = 0

        # close all threads

        self.server.close()
        for c in self.threads:
            c.join()

class Client(threading.Thread):
    initialized=0

    def __init__(self,(client,address), queue):
        threading.Thread.__init__(self)
        self.client = client
  开发者_StackOverflow社区      self.address = address
        self.size = 1024
        self.queue = queue
        print 'Client thread created!'


    def run(self):
        running = 10
        isdata2=0
        receivedonce=0

        while running > 0:

            if receivedonce == 0:
                print 'Wait for initialisation message'
                data = self.client.recv(self.size)
                receivedonce = 1

            if self.queue.empty():
                print 'Queue is empty'
            else:
                print 'Queue has information'
                data2 = self.queue.get(1, 1)
                isdata2 = 1
                if data2 == 'Exit':
                    running = 0
                    print 'Client is being closed'
                    self.client.close()

            if data:
                print 'Data received through socket! First char: "' + data[0] + '"'
                if data[0] == 'I':
                    print 'Initializing user'
                    user = {'uid': data[1:6] ,'x': data[6:9], 'y': data[9:12]}
                    globuser[user['uid']] = user
                    print globuser
                    initialized=1
                    self.client.send('Beginning - Initialized'+';')
                    m=updateClient(user['uid'], queue)
                    m.start()
                else:
                    print 'Reset receivedonce'
                    receivedonce = 0

                print 'Sending client data'
                self.client.send('Feedback: ' +data+';')
                print 'Client Data sent: ' + data

            data=None

            if isdata2 == 1:
                print 'Data2 received: ' + data2
                self.client.sendall(data2)
                self.queue.task_done()
                isdata2 = 0

            time.sleep(1)
            running = running - 1

        print 'Client has stopped'


class updateClient(threading.Thread):

    def __init__(self,uid, queue):
        threading.Thread.__init__(self)
        self.uid = uid
        self.queue = queue
        global globuser
        print 'updateClient thread started!'

    def run(self):
        running = 20
        test=0
        while running > 0:
            test = test + 1
            self.queue.put('Test Queue Data #' + str(test))
            running = running - 1
            time.sleep(1)

        print 'Updateclient has stopped'


if __name__ == "__main__":
    s = Server()
    s.run() 

I don't understand your logic -- in particular, why you deliberately set up two threads writing at the same time on the same socket (which they both call self.client), without any synchronization or coordination, an arrangement that seems guaranteed to cause problems.

Anyway, a definite bug in your code is you use of the send method -- you appear to believe that it guarantees to send all of its argument string, but that's very definitely not the case, see the docs:

Returns the number of bytes sent. Applications are responsible for checking that all data has been sent; if only some of the data was transmitted, the application needs to attempt delivery of the remaining data.

sendall is the method that you probably want:

Unlike send(), this method continues to send data from string until either all data has been sent or an error occurs.

Other problems include the fact that updateClient is apparently designed to never terminate (differently from the other two thread classes -- when those terminate, updateClient instances won't, and they'll just keep running and keep the process alive), redundant global statements (innocuous, just confusing), some threads trying to read a dict (via the iteritems method) while other threads are changing it, again without any locking or coordination, etc, etc -- I'm sure there may be even more bugs or problems, but, after spotting several, one's eyes tend to start to glaze over;-).

You have three major problems. The first problem is likely the answer to your question.

Blocking (Question Problem)

The socket.recv is blocking. This means that execution is halted and the thread goes to sleep until it can read data from the socket. So your third update thread just fills the queue up but it only gets emptied when you get a message. The queue is also emptied by one message at a time.

This is likely why it will not send data unless you send it data.

Message Protocol On Stream Protocol

You are trying to use the socket stream like a message stream. What I mean is you have:

 self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

The SOCK_STREAM part says it is a stream not a message such as SOCK_DGRAM. However, TCP does not support message. So what you have to do is build messages such as:

 data =struct.pack('I', len(msg)) + msg
 socket.sendall(data)

Then the receiving end will looking for the length field and read the data into a buffer. Once enough data is in the buffer it can grab out the entire message.

Your current setup is working because your messages are small enough to all be placed into the same packet and also placed into the socket buffer together. However, once you start sending large data over multiple calls with socket.send or socket.sendall you are going to start having multiple messages and partial messages being read unless you implement a message protocol on top of the socket byte stream.

Threads

Even though threads can be easier to use when starting out they come with a lot of problems and can degrade performance if used incorrectly especially in Python. I love threads so do not get me wrong. Python also has a problem with the GIL (global interpreter lock) so you get bad performance when using threads that are CPU bound. Your code is mostly I/O bound at the moment, but in the future it may become CPU bound. Also you have to worry about locking with threading. A thread can be a quick fix but may not be the best fix. There are circumstances where threading is quite simply the easiest way to break some time consuming process. So do not discard threads as evil or terrible. In Python they are considered bad mainly because of the GIL, and in other languages (including Python) because of concurrency issues so most people recommend you to use multiple processes with Python or use asynchronous code. The subject of to use a thread or not is very complex as it depends on the language (way your code is run), the system (single or multiple processors), and contention (trying to share a resource with locking), and other factors, but generally asynchronous code is faster because it utilizes more CPU with less overhead especially if you are not CPU bound.

The solution is the usage of the select module in Python, or something similar. It will tell you when a socket has data to be read, and you can set a timeout parameter.

You can gain more performance by doing asynchronous work (asynchronous sockets). To turn a socket into asynchronous mode you simply call socket.settimeout(0) which will make it not block. However, you will constantly eat CPU spinning waiting for data. The select module and friends will prevent you from spinning.

Generally for performance you want to do as much asynchronous (same thread) as possible, and then expand with more threads that are also doing as much asynchronously as possible. However as previously noted Python is an exception to this rule because of the GIL (global interpreter lock) which can actually degrade performance from what I have read. If you are interesting you should try writing a test case and find out!

You should also check out the thread locking primitives in the threading module. They are Lock, RLock, and Condition. They can help multiple threads share data with out problems.

lock = threading.Lock()
def myfunction(arg):
    with lock:
        arg.do_something()

Some Python objects are thread safe and others are not.

Sending Updates Asynchronously (Improvement)

Instead of using a third thread only to send updates you could instead use the client thread to send updates by checking the current time with the last time an update was sent. This would remove the usage of a Queue and a Thread. Also to do this you must convert your client code into asynchronous code and have a timeout on your select so that you can at interval check the current time to see if an update is needed.

Summary

I would recommend you rewrite your code using asynchronous socket code. I would also recommend that you use a single thread for all clients and the server. This will improve performance and decrease latency. It would make debugging easier because you would have no possible concurrency issues like you have with threads. Also, fix your message protocol before it fails.