python - TCP 线程 python 服务器未按预期处理信号

Question

我需要运行一个服务器，一次最多处理 3 个请求。我的设计是我有一个 TCP 服务器正在运行，并且将运行 3 个线程来处理这些请求。服务器将接受请求并将这些请求传递给使用相应锁队列的线程。我也有适当的队列锁。我的问题是，即使我有一个信号处理程序来在主进程必须退出时使用标志向线程发出信号。我不明白错误是什么，因为它没有按预期正常退出。输出结果如下:

vm:~/Desktop$ python multi_threaded_queueing.py
About to kickoff
About to kickoff
Starting Thread-1
About to kickoff
Starting Thread-2
Starting Thread-3
^CTraceback (most recent call last):
  File "multi_threaded_queueing.py", line 94, in <module>
    conn, addr = s.accept()
  File "/usr/lib/python2.7/socket.py", line 202, in accept
    sock, addr = self._sock.accept()
socket.error: [Errno 4] Interrupted system call

代码如下:

#!/usr/bin/python

import Queue
import threading
import time
import sys
import socket
import signal


HOST = '127.0.0.1'
PORT = 50007              # Arbitrary non-privileged port
s = None

exitFlag = 0
#signal handler for control C
def signal_handler(signal, frame):
    print "Control+C has been pressed"
    #setting the exit flag so that all the threads can get notified
    exitFlag = 1
    #wait till all the threads have finished processing and can gracefully exit
    #I maintain an array for each thread to set the corresponding index when
    #it has finished its processing. I and all the elements to see if its 0
    #and based on which I will exit or wait
    while 1:
        num = 0
        for ele in exitList:
            num &= ele
        if ele == 0:
            sys.exit(0)

class myThread (threading.Thread):
    #have a queue, thread ID and name for every thread.
    def __init__(self, threadID, name, q):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.q = q
    def run(self):
        print "Starting " + self.name
        process_data(self.name, self.q, self.threadID)
        print "Exiting " + self.name

def process_data(threadName, q, threadID):
    #while exit flag is not set by the main thread keep processing the data
    #present in the queue.
    while not exitFlag:
        queueLock[threadID].acquire()
        if not workQueue[threadID].empty():
            data = q[threadID].get()
            queueLock[threadID].release()
            print "%s processing %s" % (threadName, data)
        else:
            queueLock[threadID].release()
        time.sleep(1)
    exitThread[threadID] = 1

threadList = ["Thread-1", "Thread-2", "Thread-3"]
nameList = ["One", "Two", "Three", "Four", "Five"]
queueLock = []
workQueue = []
threads = []
threadID = 0
exitList = []
size = 3
request = 0
signal.signal(signal.SIGINT, signal_handler)

# Create new threads
#by default hard coding the number of threads to 3
for tName in threadList:
    workQueue.append(Queue.Queue(10))
    queueLock.append(threading.Lock())
    exitList.append(0)
    thread = myThread(threadID, tName, workQueue)
    print "About to kickoff"
    thread.start()
    threads.append(thread)
    threadID += 1


for res in socket.getaddrinfo(HOST, PORT, socket.AF_UNSPEC, socket.SOCK_STREAM, 0, socket.AI_PASSIVE):
    af, socktype, proto, canonname, sa = res
    try:
        socket.setdefaulttimeout(10)
        s = socket.socket(af, socktype, proto)
    except socket.error, msg:
        s = None
        continue
    try:
        s.bind(sa)
        s.listen(1)
    except socket.error, msg:
        s.close()
        s = None
        continue
    break
if s is None:
    print 'could not open socket'
    sys.exit(1)
while 1:
    conn, addr = s.accept()
    print 'Connected by', addr
    request += 1
    #round robin scheduling for each thread
    thread_index = request % size
    while 1:
        data = conn.recv(1024)
        if not data: break
        # Fill the queue with the request received
        queueLock[thread_index].acquire()
        for word in nameList:
            workQueue[thread_index].put(word)
        queueLock[thread_index].release()
        # Wait for queue to empty
        while not workQueue[thread_index].empty():
            pass
        conn.send(data)
    conn.close()



# Notify threads it's time to exit
exitFlag = 1
print "setting the exitFlag"
# Wait for all threads to complete
for t in threads:
    t.join()
print "Exiting Main Thread"

Answer 1

有几件事情正在发生。

1. signal_handler(signal, frame): 未设置全局 exitFlag。您需要将 global exitFlag 添加到函数的顶部。

2. sys.exit() 并没有真正退出 - 它只是引发了一个 KeyboardInterrupt 错误。

3. socket.error: [Errno 4] Interrupted system call 是个好东西，它可以防止您的程序卡在 conn, addr = s.accept()。您应该捕获 socket.error 异常并使用它们来跳出 while 循环。