Python asyncore / asynchat 基本传输实验 - Jacky Liu's Blog

Python asyncore / asynchat 基本传输实验

Jacky Liu posted @ 2012年7月03日 10:18 in Python with tags python asyncore asynchat , 4995 阅读

    ---- 自从上回实验了 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

 

Avatar_small
依云 说:
2012年7月03日 15:34

那个「死循环」,叫作「无限等待」更好。这时系统在等待永远也不可能到来的事件,和死锁差不多了。Node.js 似乎不会出现这种问题。

Avatar_small
Jacky Liu 说:
2012年7月03日 17:47

@依云: 我不懂 js,曾经试着看过些概念也不得要领。我只是感觉作为脚本语言,Python 遵循的是很原生态的哲学,标准库里的那些特性真的非常非常的谦虚内敛。这个好也不好,对我这种菜鸟来说可能还是好处多些。如果其它脚本语言提供更加高级的功能,应该是一点都不奇怪的,尤其 js 这种高层而又有针对性的东西。

另外想问一下,你所知道的 tornado 是怎么对付那两个问题的,如果有连接挂了怎么办?还有能不能动态地添加或删除连接呢?

Avatar_small
Jacky Liu 说:
2012年7月03日 17:54

@依云: 或者不一定 tornado,任何网络传输库都行,因为它们在底层用的应该是一样的 非阻塞 socket + polling loop 的特性。

Avatar_small
依云 说:
2012年7月03日 18:03

@Jacky Liu: Tornado 写应用的话,一般都是持续跑的,所以不用担心没事做干等待了。至于连接断掉,写 Web 的话框架已经处理好了,其它的自己 try 嘛。

Tornado 是可以任何时候添加/修改/移除文件描述符的(几乎所有I/O主循环的库都是这样)。

Tornado 在底层使用的是:Linux->epoll,大多数 BSD->kqueue,其余 fallback 到 select。

Avatar_small
Jacky Liu 说:
2012年7月03日 18:14

@依云: 果然!这再一次验证了 Python 标准库的原生态,还有菜鸟的苦逼!

Avatar_small
Jacky Liu 说:
2012年7月03日 19:49

@依云: 刚才看见 Tornado 的源码,还好还好。大部分也是 python 写的,也不算太大。现在对 python 不那么怵。看不懂,但关键应该在 ioloop 和 iostream。有空研究研究 。。。


登录 *


loading captcha image...
(输入验证码)
or Ctrl+Enter
Host by is-Programmer.com | Power by Chito 1.3.3 beta | © 2007 LinuxGem | Design by Matthew "Agent Spork" McGee