Python多线程与Socket编程出现'现有连接被远程主机强制关闭'错误求解

如果您想直接查看最优解释，可跳转至本问题最末尾。

我开发了一个小型Python客户端/服务端项目，连接两端均使用同一个类，代码如下：

from channel import Channel
import time
import constants
import sys
is_py2 = sys.version[0] == '2'
if is_py2:
    import Queue as queue
else:
    import queue as queue

class OneSide:
  def __init__(self, address1, address2, target1, target2):
    self.inQueue1 = queue.Queue()
    self.channel1 = Channel(address1,target1,self.inQueue1)
    time.sleep(.500)
    self.inQueue2 = queue.Queue()
    self.channel2 = Channel(address2,target2,self.inQueue2)

  def close(self):
    self.channel1.stopListening();
    self.channel2.stopListening();

  def from1(self, msg):
    self.channel1.write(msg)

  def from2(self, msg):
    self.channel2.write(msg)

  def read1(self):
    while not self.inQueue1.empty():
      print(self.inQueue1.get())

  def read2(self):
    while not self.inQueue2.empty():
      print(self.inQueue2.get())

为保证信息完整，我也附上Channel类的代码：

import threading
import socket
import sys
is_py2 = sys.version[0] == '2'
if is_py2:
    import Queue as queue
else:
    import queue as queue

class Channel:
  def __init__(self, address, target, outQueue):
    self.address = address
    self.target = target
    self.connection = socket.socket(family=socket.AF_INET, type=socket.SOCK_DGRAM)
    self.connection.bind(address)
    self.connection.settimeout(5.0)
    self.outQueue = outQueue
####THIS IS THE PROBLEM####
    self.serverThread = threading.Thread(target=self.serverFunction)
    self.startListening()
####THIS IS THE PROBLEM####

  def startListening(self):
    self.running = True
    self.serverThread.start()

  def stopListening(self):
    print("Shutting down server thread port {}".format(self.address[1]))
    self.running = False

  def serverFunction(self):
    # time.sleep(.100) # sleep 100 ms
    print("Server thread started on port {}".format(self.address[1]))
    while self.running:
      try:
        self.outQueue.put(self.read(1024))
      except socket.timeout:
        #no error; this can occur if the read times out
        pass
      except Exception as e:
        # must have been forced close
        print("Server thread failed on port {}".format(self.address[1]))
        print("  Error: {}".format(e))
        self.running = False
    self.connection.shutdown(socket.SHUT_RDWR)
    self.connection.close()
    print("Closed server thread port {}".format(self.address[1]))

  def write(self, bytes):
    self.connection.sendto(bytes, self.target)

  def read(self, bufferSize):
    bytesAddressPair = self.connection.recvfrom(bufferSize)
    message = bytesAddressPair[0]
    return bytearray(message)

我最初写了一个main.py文件，同时创建两个对象互相通信，代码如下：

from oneside import OneSide
import time

c1 = OneSide(("127.0.0.1",2000),("127.0.0.1",2001),("127.0.0.1",1000),("127.0.0.1",1001));
time.sleep(.500)
c2 = OneSide(("127.0.0.1",1000),("127.0.0.1",1001),("127.0.0.1",2000),("127.0.0.1",2001));
time.sleep(.500)

c1.from1("bytes1")
c1.from2("bytes2")

c2.from1("reverse1")
c2.from2("reverse2")

c1.read1()
c1.read2()

c2.read1()
c2.read2()

c2.close()
c1.close()

这段代码运行完全正常。后来我想分开运行两个实例，就拆分出两个独立的main文件分别运行，其中一个的代码如下：

from oneside import OneSide
import time

c1 = OneSide(("127.0.0.1",2000),("127.0.0.1",2001),("127.0.0.1",1000),("127.0.0.1",1001));
time.sleep(.500)

c1.from1("bytes1")
c1.from2("bytes2")

c1.read1()
c1.read2()

c1.close()

但运行时就会报错，哪怕只运行其中一个或者同时运行两个都会出现该错误：

Error: [Errno 10054] An existing connection was forcibly closed by the remote host

我有三个疑问：

1. 该错误的触发原因是什么？
2. 如何修复这个问题？
3. 为什么同时运行两个服务端的时候不会出现该错误？

补充信息 #

我后续调试时两个OneSide对象之间的通信已经可以正常运行，但改用多线程模式运行时又会出现同样的错误。

最终编辑 #

请注意我仍然在寻找相关原理说明，目前我没有找到任何文档解释为什么多线程和Socket不能混用。不过我自己排查出了以下结论：
我是逐步迭代开发该程序的，最初我给Channel设置了一个线程化的server函数用于读取数据并打印，之后改成将读取到的数据存入队列供其他线程读取，后来我又把读队列的逻辑也放进了独立线程，全程Socket都是在主线程中创建的。

当我意识到Socket不应该在创建它的线程之外被访问后，情况有所好转。我重写了多线程版本的STOneSide（线程安全OneSide类），改为在子线程内部创建Channel对象，代码如下：

def startup(self, channel_id, address, target):
        self.channels[channel_id] = Channel(address, target, self.queues[channel_id])
        port = self.addresses[channel_id][1]
        safe_print("Port {} Stage 1".format(port))

        while not self.states[channel_id] == CONNECTED:
            self.states[channel_id] = STAGE1
...

此时还是会偶现报错，但已经可以偶尔连接成功。我当时还没意识到Channel内部也会创建新线程，当我发现这点后，我去掉了Channel内部的线程，临时修改代码，直接在创建Channel和Socket的STOneSide子线程函数中读取数据，问题就彻底解决了，所有连接都正常，不再报错。修改后的代码如下：

#From STOneSide
    def read_all(self, channel_id):
        self.channels[channel_id].read_all()
        output = []
        while not self.queues[channel_id].empty():
            output.append(self.queues[channel_id].get())
        return output

# From Channel
  def read_all(self):
    running = True
    while running:
      try:
        self.outQueue.put(self.read(1024))
      except socket.timeout:
        #end of all data
        running = False
        pass
      except Exception as e:
        # must have been forced close
        safe_print("Server thread failed on port {}".format(self.address[1]))
        safe_print("  Error: {}".format(e))
        running = False

现在我已经知道问题所在和修复方法，但还是希望能获得官方文档说明，解释为什么不能跨线程访问Socket。

1. 10054错误触发原因 #

你使用的是UDP协议，本身属于无连接协议，但Windows系统内核会缓存ICMP端口不可达报文：当你发出去的UDP报文对应的目标端口没有进程监听时，对端会返回ICMP不可达，内核会把这个错误关联到对应的socket上，下一次调用该socket的recvfrom方法时就会抛出10054错误。
单独运行单个实例的时候，你发送的报文没有对应接收端监听端口，自然会触发这个错误；单进程同时运行两个实例的时候，双方端口都有进程监听，不会返回ICMP不可达报文，所以不会报错。

2. 跨线程操作socket的官方说明 #

Python标准库的socket对象本身是线程不安全的，官方文档明确标注：socket对象的所有方法都没有内部同步机制，如果多个线程同时操作同一个socket对象，需要用户自行实现互斥锁保证访问顺序，否则会出现不可预期的状态错误。
你之前的问题本质是多个线程（主线程创建socket，子线程读、主线程写/关闭）同时操作同一个socket，没有做同步，导致socket状态混乱，才会偶发报错。你后续改成所有socket操作都在创建它的同一个线程内完成，天然避免了竞态条件，所以问题消失。

3. 修复方案 #

和你排查的结果一致，两种方案可选：

• 保证每个socket的所有操作（创建、读、写、关闭）都在同一个线程内完成
• 给所有跨线程的socket操作加互斥锁，避免并发访问

内容的提问来源于stack exchange，提问作者BSD