---- 自从上回实验了 Python socket 的基本传输之后又受了些启发，于是想试试基于 non-blocking socket 的通信机制。Python 标准库里的 asynchat 自然是第一步。昨天写的实验程序实现了基于 asynchat 的基本通信，把前因后果总结一下:

    实验前的考虑

    ---- 用 non-blocking socket 最基本的考虑自然是传输效率，尤其是遇到一大坨数据过来的时候，希望它能尽快传送完，这时最好其它线程通通都停掉，在传完之前不要再无谓地换来换去（反正其它线程此时应该也没甚实际工作要干），当然这只是愿望而已。既然做不到，就只能追求尽量快速的传输方式了。

    ---- 查资料得出的印象是 non-blocking socket 当然比较快，但是好的东西总是有代价，这是源于我个人的印象。因为非阻塞意味着没有延时，而没有延时环节的无限循环跑起来可能会很耗 CPU。但是另一方面来讲，既然非阻塞 socket 已经用得这么多，意味着底层实现很可能已经解决了这个问题，我的担心很可能是不成立的。这个还是要试试才知道。

    结果总结

    1. 关于 CPU 使用

    ---- 实验的结果，当服务端非正常退出以后，使用 asynchat 的客户端好像就会陷入死循环，没有任何异常的迹象，所以此时 polling loop 应该还是在运行的，而 CPU 使用并没有明显高于平时的水平，所以 non-blocking socket + polling loop 应该对 CPU 并没有特别大的压力，当然如果开十个 polling loop 可能会是另外一回事。

    2. 对坏连接的处理

    ---- 关于死循环，这实际上就是一个网络通信模块有没有能力对付坏连接的问题。如果用 blocking socket，上一篇已经讲了，会抛出异常或返回 b''，编程时就可以做相应处理，而死循环显然不是个很讨喜的行为方式。我想了一下如果直接用底层的 select() 函数时会发生什么。因为我曾经在进程间通信的特性中体验过 select() / pooling loop 的用法，这里用作网络传输时应该也差不多。

    ---- 如果一个进程（或者网络连接）还没吱一声就挂了，那么与它相关的 fd（或 socket）应该仍然会被 select() 放在有效列表里返回，但是后续的 read()（或 recv()）操作就会读到一个空的对象，这时你就知道这个进程或连接已经挂了。再看 asynchat 提供的那些寥寥的接口: 好像 collect_incoming_data() 跟 recv() 也差不多，能不能在里面判断，当 data 参数是空的就说明连接已经挂了呢？试了一下，不行。collect_incoming_data() 的 data 参数总是有效的，也就是说读不到内容时这个函数根本不会被调用。

    ---- 如果再要深究下去大概要去看 asynchat.py 了，但问题是 asynchat 完全是用 Python 写的而且只有几百行而已，它应该木有那么多考虑在里面，我想有这功夫还不如直接去试验 select()。

    3. 在运行中更新连接

    ---- 另一个很重要的问题，我觉得网络通信模块应该提供在 pooling loop 运行的时候动态添加或删除连接的功能，但是很显然 asynchat 并没有这个特性的接口。试着动点脑筋 —— 因为连接对象是通过 asyncore.loop() 的一个参数: map 来传递的，能不能动态修改它呢？文档里说了 map 参数是普通的 dict 类型，dict 当然没有线程安全性，也就是说从 asyncore.loop() 所在的线程之外的地方动它肯定会让 loop() 崩溃。如果在同一个线程内呢？行不行呢？这大概又要去看 asyncore.py 了，卟啦卟啦 。。。如果有这功夫，卟啦卟啦 。。。

    所以，总结的总结:

    ---- 我觉得上面的 2 跟 3 对于一个网络传输组件来说应该算是很基本的功能。如果缺了这两个功能，那这个网络传输组件应该没太大实际用处，大概只能用在一次性程序里。但是实话讲我对异步网络传输的了解还浅得很，俩月前我还不知道 socket 是什么，也不知道局域网该怎么连，所以以上这些都不是很确定。眼下先跑起来，以后有机会再修正吧。

    ---- 实验程序:
 

# -*- encoding: utf-8 -*-



'''
实验平台:
	Ubuntu 12.04 LTS / Linux 3.2.0-25-generic-pae

实验目标:
	1. asynchat 模块的基本传输机制
	2. 让服务端在未关闭 socket 的情况下终止运行，模拟连接异常中断的情况，检视后续行为。

实验过程:
	1. 另起一个进程作为服务端，服务端仍然用 blocking socket + 多线程的方式，针对每个连接开启一个
	发送线程和一个接收线程。服务端的任务是向客户端发送数据（一次 3 Mb）并接收客户端的回执。

	2. 主进程内建立客户端对象并向服务端发起指定数目的连接。客户端使用 asynchat 模块，背后的机制
	是 non-blocking socket + select()。连接建立以后向服务端发送一条 'start' 消息，通知服务端开始
	发送，之后从服务端接收数据，收完一条之后发送一个回执。

	3. 服务端完成指定次数的发送以后，在未关闭 socket 的情况下退出运行，检视后续的行为。

实验结果:
	1. 客户端使用 asynchat 模块实现了基本传输功能。服务端非正常退出以后 CPU 使用一直保持低水平，
	此时 polling loop 应该是在运行的，所以 non-blocking socket + polling loop 的方式应该对 CPU
	并没有太大压力。

	2.1. 如果服务端发送线程选择未关闭 socket 就退出，那么 asyncore.loop() 就会死循环，不抛出异常
	2.2. 如果服务端发送线程选择先关闭 socket 再退出，那么 asyncore.loop() 就会抛出异常
'''



import asyncore
import asynchat

import socket
import multiprocessing
import time
import threading



__host_server__= '127.0.0.1'
__port_server__= 9000

__host_client__= '127.0.0.1'



__消息内容__= b'head' + b'.' * (3 * 2**20 - 8) + b'tail'	# 一次 3 Mb
__结束标志__= b'\r<--EOM-->\r'
__发送次数__= 3
__连接个数__= 3	
__接收缓存__= 2**15
__消息编码__= 'utf-8'



#================================================== ↓ 服务端 ↓ ==================================================

# 全局 socket，最后在主进程内关闭。
sock_server= socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock_server.bind((__host_server__, __port_server__))
sock_server.listen(1)	# 开始监听



def 服务端进程():

	服务端对象= 服务端()

	for i in range(__连接个数__):
		try:
			sock, addr= sock_server.accept()
		except:
			print('服务端进程() -- server socket 已关闭，监听进程退出 ...')
			break

		print('服务端进程() -- 接收到来自 ' + str(addr) + ' 的连接。')

		服务端对象.添加连接(addr=addr, sock=sock)

	time.sleep(0.3)		# 等所有发送线程开始运行

	服务端对象.join()

	# 服务端退出 ...
	print('服务端进程() -- 所有连接已终止，服务端进程退出 ...')



class 服务端:

	def __init__(self):
		self._连接集= {}

	def 添加连接(self, addr, sock):
		新连接= 服务端连接(addr=addr, sock=sock)
		self._连接集[addr]= 新连接

	def join(self):
		for conn in self._连接集.values():
			conn.join()



class 服务端连接:

	def __init__(self, addr, sock):

		self._addr= addr
		self._sock= sock
		self._buff= b''

		self._发送= False
		self._发送线程= threading.Thread(target=self._服务端发送线程, name='发送线程_'+str(self._addr), kwargs={})
		self._接收线程= threading.Thread(target=self._服务端接收线程, name='接收线程_'+str(self._addr), kwargs={})

		self._发送线程.start()
		self._接收线程.start()



	def _服务端发送线程(self):
		'''

		'''
		sock= self._sock
		addr= self._addr

		while not self._发送:
			time.sleep(0.3)

		print('_服务端发送线程() -- 与 ' + str(addr) + ' 的连接开始发送 ...')

		for i in range(1, __发送次数__+1):
			try:
				sock.sendall(__消息内容__ + __结束标志__)
			except:
				print('_服务端发送线程() -- 与 ' + str(addr) + ' 的连接已损坏，发送线程终止。')
				break
			print('_服务端发送线程() -- 与 ' + str(addr) + ' 的连接已发 ' + str(i) + ' 条。')

		print('_服务端发送线程() -- 与 ' + str(addr) + ' 的连接发送完毕，发送线程退出 ... ')

		time.sleep(3)

		# XXX: 非正常退出，未关闭 socket 的情况下就退出
		#	try: sock.shutdown(socket.SHUT_RDWR)
		#	except: pass
		#	sock.close()



	def _服务端接收线程(self):
		'''

		'''
		sock= self._sock
		addr= self._addr

		print('_服务端接收线程() -- 与 ' + str(addr) + ' 的连接开始接收 ...')

		while True:
			data= sock.recv(8192)

			if not data:
				print('_服务端接收线程() -- 与 ' + str(addr) + ' 的连接已损坏，接收线程终止。')
				break

			sidx= max(len(self._buff) - len(__结束标志__), 0)
			self._buff += data

			if __结束标志__ in self._buff[sidx:]:
				mlist= self._buff.split(__结束标志__)
				self._buff= mlist.pop(-1)

				for msg in mlist:
					msg= msg.decode(__消息编码__)
					print('_服务端接收线程() -- 接收到 ' + str(addr) + ' 发来的信息: "' + msg + '"')
					self._处理所收消息(msg=msg)

		try: sock.shutdown(socket.SHUT_RDWR)
		except: pass
		sock.close()



	def _处理所收消息(self, msg):

		if msg == 'start':
			self._发送= True



	def join(self):
		self._发送线程.join()
		self._接收线程.join()



#================================================== ↓ 客户端 ↓ ==================================================

class 客户端:

	def __init__(self):
		'''

		'''
		self._连接集= {}



	def 建立连接(self):
		'''

		'''
		for i in range(1, __连接个数__+1):
			addr= (__host_client__, 9000+i)
			self._连接集[addr]= 客户端连接(addr=addr)



	def start(self):
		'''

		'''
		for conn in self._连接集.values():
			conn.start()

		map= {conn._sock.fileno(): conn for conn in self._连接集.values()}
		polling_loop= threading.Thread(target=asyncore.loop, name='polling_loop', kwargs={'map':map})
		polling_loop.start()

		print('客户端.start() -- 客户端已开始运行 ...')



class 客户端连接(asynchat.async_chat):

	def __init__(self, addr):

		self._addr= addr
		#	self._sock= self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
		self._sock= socket.socket(socket.AF_INET, socket.SOCK_STREAM)
		self._sock.bind(addr)
		asynchat.async_chat.__init__(self, self._sock)
		self._buff= b''
		self._msgs= []
		self.ac_in_buffer_size= __接收缓存__
		self.set_terminator(__结束标志__)
		self.connect( (__host_server__, __port_server__) )



	def collect_incoming_data(self, data):
		if not data:	# 说明连接已损坏
			print('客户端连接.collect_incoming_data() -- 连接 ' + str(self._addr) + ' 已损坏。')

		self._buff += data



	def found_terminator(self):

		msg= self._buff.decode(__消息编码__)
		self._msgs.append(msg)
		self._buff= b''

		print('客户端连接.found_terminator() -- 连接 ' + str(self._addr) + ' 共收到 ' + str(len(self._msgs)) + ' 条。')

		self.push( ('已收 ' + str(len(self._msgs)) + ' 条。').encode(__消息编码__) + __结束标志__ )



	def start(self):
		self.push(b'start' + __结束标志__)



#================================================== ↓ 主进程 ↓ ==================================================

# 启动监听进程
服务端进程对象= multiprocessing.Process(name='服务端进程', target=服务端进程)
服务端进程对象.start()



客户端对象= 客户端()
客户端对象.建立连接()
time.sleep(0.3)
客户端对象.start()

    ---- 以下是运行结果:

        服务端进程() -- 接收到来自 ('127.0.0.1', 9001) 的连接。
        服务端进程() -- 接收到来自 ('127.0.0.1', 9002) 的连接。
        _服务端接收线程() -- 与 ('127.0.0.1', 9001) 的连接开始接收 ...
        _服务端接收线程() -- 与 ('127.0.0.1', 9002) 的连接开始接收 ...
        客户端.start() -- 客户端已开始运行 ...
        _服务端接收线程() -- 接收到 ('127.0.0.1', 9001) 发来的信息: "start"
        服务端进程() -- 接收到来自 ('127.0.0.1', 9003) 的连接。
        _服务端接收线程() -- 接收到 ('127.0.0.1', 9002) 发来的信息: "start"
        _服务端接收线程() -- 与 ('127.0.0.1', 9003) 的连接开始接收 ...
        _服务端接收线程() -- 接收到 ('127.0.0.1', 9003) 发来的信息: "start"
        _服务端发送线程() -- 与 ('127.0.0.1', 9001) 的连接开始发送 ...
        _服务端发送线程() -- 与 ('127.0.0.1', 9002) 的连接开始发送 ...
        _服务端发送线程() -- 与 ('127.0.0.1', 9002) 的连接已发 1 条。
        _服务端发送线程() -- 与 ('127.0.0.1', 9001) 的连接已发 1 条。
        _服务端发送线程() -- 与 ('127.0.0.1', 9003) 的连接开始发送 ...
        客户端连接.found_terminator() -- 连接 ('127.0.0.1', 9002) 共收到 1 条。
        _服务端接收线程() -- 接收到 ('127.0.0.1', 9002) 发来的信息: "已收 1 条。"
        客户端连接.found_terminator() -- 连接 ('127.0.0.1', 9001) 共收到 1 条。
        _服务端接收线程() -- 接收到 ('127.0.0.1', 9001) 发来的信息: "已收 1 条。"
        _服务端发送线程() -- 与 ('127.0.0.1', 9002) 的连接已发 2 条。
        _服务端发送线程() -- 与 ('127.0.0.1', 9003) 的连接已发 1 条。
        _服务端发送线程() -- 与 ('127.0.0.1', 9001) 的连接已发 2 条。
        客户端连接.found_terminator() -- 连接 ('127.0.0.1', 9002) 共收到 2 条。
        _服务端接收线程() -- 接收到 ('127.0.0.1', 9002) 发来的信息: "已收 2 条。"
        客户端连接.found_terminator() -- 连接 ('127.0.0.1', 9003) 共收到 1 条。
        _服务端接收线程() -- 接收到 ('127.0.0.1', 9003) 发来的信息: "已收 1 条。"
        客户端连接.found_terminator() -- 连接 ('127.0.0.1', 9001) 共收到 2 条。
        _服务端接收线程() -- 接收到 ('127.0.0.1', 9001) 发来的信息: "已收 2 条。"
        _服务端发送线程() -- 与 ('127.0.0.1', 9002) 的连接已发 3 条。
        _服务端发送线程() -- 与 ('127.0.0.1', 9002) 的连接发送完毕，发送线程退出 ...
        _服务端发送线程() -- 与 ('127.0.0.1', 9003) 的连接已发 2 条。
        _服务端发送线程() -- 与 ('127.0.0.1', 9001) 的连接已发 3 条。
        _服务端发送线程() -- 与 ('127.0.0.1', 9001) 的连接发送完毕，发送线程退出 ...
        客户端连接.found_terminator() -- 连接 ('127.0.0.1', 9002) 共收到 3 条。
        _服务端接收线程() -- 接收到 ('127.0.0.1', 9002) 发来的信息: "已收 3 条。"
        客户端连接.found_terminator() -- 连接 ('127.0.0.1', 9003) 共收到 2 条。
        _服务端接收线程() -- 接收到 ('127.0.0.1', 9003) 发来的信息: "已收 2 条。"
        客户端连接.found_terminator() -- 连接 ('127.0.0.1', 9001) 共收到 3 条。
        _服务端接收线程() -- 接收到 ('127.0.0.1', 9001) 发来的信息: "已收 3 条。"
        _服务端发送线程() -- 与 ('127.0.0.1', 9003) 的连接已发 3 条。
        _服务端发送线程() -- 与 ('127.0.0.1', 9003) 的连接发送完毕，发送线程退出 ...
        客户端连接.found_terminator() -- 连接 ('127.0.0.1', 9003) 共收到 3 条。
        _服务端接收线程() -- 接收到 ('127.0.0.1', 9003) 发来的信息: "已收 3 条。"


        ^CException KeyboardInterrupt: KeyboardInterrupt() in <module 'threading' from '/usr/lib/python3.2/threading.py'> ignored
        Process 服务端进程:
        Traceback (most recent call last):
          File "/usr/lib/python3.2/multiprocessing/process.py", line 267, in _bootstrap
            self.run()
          File "/usr/lib/python3.2/multiprocessing/process.py", line 116, in run
            self._target(*self._args, **self._kwargs)
          File "asynchat基本传输.py", line 83, in 服务端进程
            服务端对象.join()
          File "asynchat基本传输.py", line 101, in join
            conn.join()
          File "asynchat基本传输.py", line 196, in join
            self._接收线程.join()
          File "/usr/lib/python3.2/threading.py", line 854, in join
            self._block.wait()
          File "/usr/lib/python3.2/threading.py", line 235, in wait
            waiter.acquire()
        KeyboardInterrupt
 

• 全局相关文章
• 相关文章