PyTianQiService的核心网络服务模块
张可人
PyTianQiService的核心网络服务模块，是一个单线程，基于EPOLL事件循环的TCP通信框架。
---
#!/usr/bin/env python
# -*- coding: utf-8 -*-
##########################################################
# Teach Wisedom To Machine.
# Please Call Me Programming devil.
# Module Name: EpollServer
###################################################### #
from EpollComm import *
from ServiceStatus import *
from Utils import *
logger = logging.getLogger()
statuslogger = logging.getLogger("statusLogger")
netLogger = logging.getLogger("netLogger")
serviceRunningStatus = ServiceRunningStatus()
''' LogAcceptor
'''
############################################################################
class EpollServer(ServerInterface):
'''epoll-loop-based server
'''
# listen_addr_list is list of item as (ip,port)
def __init__(self, listen_addr_list=[]):
self._listen_addr_list = listen_addr_list
self._listen_backlog = 40000
self._epoll_time_out = 1
self._timer_time_out = 10
self._max_session_count = 1024
pass
# ServerInterface.start()
def start(self):
self._is_started = 0
self._epoll_loop = EpollLoop()
self._max_notify_event_count = 0
self._max_socket_fileno = 0
self._max_doaccept_count = 0 # 一次accept通知，可以accept的最大次数
self._listener_dict = {}
for listen_addr in self._listen_addr_list:
if not self._createListenSocket(listen_addr):
return False
self._acceptor_dict = {} # <acceptor_id=xxx.fileno(),acceptor=object()>
self._invalid_acceptorfd_set = set()
self._session_id = 0
self._is_started = 1
self._is_abort_serve_once = 0 # 是否终止serve_once操作
self._last_timestamp = time.time()
return True
# ServerInterface.stop()
def stop(self):
if self._is_started == 0:
return
self._is_started = 0
self._epoll_loop.close()
for listen_socket in self._listener_dict.values():
listen_socket.close()
self._listener_dict.clear()
for acceptor in self._acceptor_dict.values():
acceptor.close()
self._is_abort_serve_once = 1
pass
# ServerInterface.serve_once()
def serve_once(self):
""" Pool the ready event """
# epoll 进行 fd 扫描的地方 -- 未指定超时时间则为阻塞等待
if self._is_abort_serve_once == 1:
netLogger.warning("serve_once _is_abort_serve_once=1 ")
return False
now_timestamp = time.time()
if now_timestamp > self._last_timestamp + self._timer_time_out:
self._last_timestamp = now_timestamp
self._triggerTimerEvent()
epoll_list = self._epoll_loop.poll(self._epoll_time_out)
for fileno, events in epoll_list:
# print_epoll_events(fileno,events)
# 若为监听 fd 被激活
if fileno in self._listener_dict:
self._onFdAcceptable(listen_fd=fileno)
continue
acceptor = self._acceptor_dict.get(fileno, None)
if acceptor is None:
t = "cannot find fd:{0} from _acceptor_dict".format(fileno)
netLogger.warning(t)
continue
if acceptor.isConnectSucc():
self._checkSocketDataEvent(fileno, events)
pass
pass
if len(epoll_list) > 0:
if len(epoll_list) > self._max_notify_event_count:
self._max_notify_event_count = len(epoll_list)
# self._reportStatus()
# 可以延时处理失效的acceptor
self._checkInvalidAcceptors()
return True
#
def getAcceptor(self,acceptor_fd=0,acceptor_sessionid=0):
if acceptor_fd not in self._acceptor_dict:
netLogger.debug("getAcceptor NULL;fd:%d sessionid:%d",acceptor_fd,acceptor_sessionid)
return None
acceptor = self._acceptor_dict[acceptor_fd]
if acceptor.client_session_id != acceptor_sessionid:
netLogger.debug("getAcceptor failed;fd:%d sessionid:%d>currentsessionid:%d", acceptor_fd, acceptor_sessionid,acceptor.client_session_id)
return None
return acceptor
# Public Virtual Method
def onTcpConnectionEnter(self, session_id, client_socket, client_address):
raise NotImplementedError()
# acceptor = _Acceptor(session_id, client_socket, client_address)
# return acceptor
# Public Virtual Method
def onProcessTimerTask(self):
raise NotImplementedError()
# Public Virtual Method
def task_once(self):
raise NotImplementedError()
# Public Virtual Method
def feedback_once(self):
raise NotImplementedError()
def _reportStatus(self):
acceptors = self._acceptor_dict.values()
accept_succ_count = reduce(lambda x, y: x + y.get_intval(), acceptors, 0)
if accept_succ_count > 2:
t = "reportStatus>> acceptors:{0} accept_succ_count:{1} max_events:{2},maxfileno:{3}".format(
len(self._acceptor_dict),
accept_succ_count,
self._max_notify_event_count,
self._max_socket_fileno)
statuslogger.debug(t)
def _assignSessionId(self):
self._session_id += 1
return self._session_id
def _pushInvalidFd(self, client_fd):
self._invalid_acceptorfd_set.add(client_fd)
# 定时事件触发器
def _triggerTimerEvent(self):
for fileno, acceptor in self._acceptor_dict.items():
if not acceptor.onTimerEvent(self._last_timestamp):
self._pushInvalidFd(fileno)
self.onProcessTimerTask()
self._reportStatus()
global serviceRunningStatus
serviceRunningStatus.report()
pass
# 创建listen socket,并且注册到epoll,关注EPOLLIN
def _createListenSocket(self, listen_addr=()):
try:
listen_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
listen_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
listen_socket.setblocking(False)
listen_socket.bind(listen_addr)
listen_socket.listen(self._listen_backlog)
self._listener_dict[listen_socket.fileno()] = listen_socket
self._epoll_loop.register(listen_socket.fileno(), select.EPOLLIN)
t = "_createListenSocket ok addr:{0} pid:{1}".format(listen_addr, os.getpid())
netLogger.info(t)
return True
except socket.error, e:
t = "_createListenSocket,error,addr:{0} pid:{1} msg:{2}".format(listen_addr, os.getpid(), repr(e))
netLogger.critical(t)
sys.exit(1)
return False
def _checkSocketDataEvent(self, fileno, events):
# 异常事件
if events & (select.EPOLLHUP | select.EPOLLERR):
self._onFdExceptional(client_fd=fileno)
print_epoll_events(fileno, events)
return
# 合法事件
if events & select.EPOLLIN:
# <连接到达；有数据来临；>有 可读 事件激活
self._onFdReadable(client_fd=fileno)
return
if events & select.EPOLLPRI:
# < 外带数据>
netLogger.warning("_checkSocketEvent EPOLLPRI")
self._onFdReadable(client_fd=fileno)
return
if events & select.EPOLLOUT:
# <有数据要写>有 可写 事件激活
self._onFdWritable(client_fd=fileno)
return
# 没有预期的事件
# EPOLLERR 是服务器这边出错
# 对端正常关闭（程序里close()，shell下kill或ctr+c），触发EPOLLIN和EPOLLRDHUP，但是不触发EPOLLERR和EPOLLHUP
# 关于这点，以前一直以为会触发EPOLLERR或者EPOLLHUP。
# man epoll_ctl看下后两个事件的说明，这两个应该是本端（server端）出错才触发的。
# 对端异常断开连接（只测了拔网线），没触发任何事件。
t = "_checkSocketDataEvent <unhandle_event fileno:{0} events:{1}> ".format(fileno, events)
netLogger.critical(t)
pass
# Accept就绪
def _onFdAcceptable(self, listen_fd):
# 进行 accept -- 获得连接上来 client 的 ip 和 port，以及 socket 句柄
listen_socket = self._listener_dict[listen_fd]
accept_count = 0
while True:
try:
client_socket, client_address = listen_socket.accept()
if len(self._acceptor_dict) > self._max_session_count:
t = "_onFdAcceptable Failed;Too Many Sessions;Then Close It"
netLogger.critical(t)
client_socket.close()
return False
session_id = self._assignSessionId()
# print "accept fd:{0} sessionId:{1}".format(client_socket.fileno(),session_id )
# logger.debug("accept connection from %s, %d, fd = %d" % (addr[0], addr[1], conn.fileno()))
# 将连接 socket 设置为 非阻塞
client_socket.setblocking(0)
acceptor = self.onTcpConnectionEnter(session_id, client_socket, client_address)
self._max_socket_fileno = client_socket.fileno()
self._addTcpAcceptor(acceptor.client_socket.fileno(), acceptor)
accept_count += 1
except socket.error, e:
if e.errno == errno.EAGAIN or e.errno == errno.EWOULDBLOCK:
# 在 非阻塞 socket 上进行 recv 需要处理 读穿 的情况
# 这里实际上是利用 读穿 出 异常 的方式跳到这里进行后续处理
# print "_onFdAcceptable EAGAIN or EWOULDBLOCK,acceptor_count:{0}".format(acceptor_count)
if accept_count > self._max_doaccept_count:
self._max_doaccept_count = accept_count
return True
else:
t = "_onFdAcceptable unhandle socket.error:{0}".format(repr(e))
netLogger.critical(t)
return False
# Read就绪
def _onFdReadable(self, client_fd):
# print "_onFdReadable fd:{0}".format(client_fd)
acceptor = self._acceptor_dict.get(client_fd, None)
if acceptor is None:
netLogger.critical("_onFdReadable failed,NULL,client_fd:%d", client_fd)
return
if acceptor.onReadEvent():
pass
# if acceptor.isNeedSend():
# # 更新 epoll 句柄中连接d 注册事件为 可写
# #self._epoll_loop.modify(client_fd, select.EPOLLET|select.EPOLLOUT )
# #print "_onFdReadable modify to epoll_out"
# else:
# print "_onFdReadable not modify"
# pass
else:
# 出错
self._pushInvalidFd(client_fd)
if not acceptor.isClosedByClient():
t = "_onFdReadable error,client_fd:{0}".format(client_fd)
netLogger.critical(t)
# Write就绪
def _onFdWritable(self, client_fd):
# print "_onFdWritable fd:{0}".format(client_fd)
acceptor = self._acceptor_dict.get(client_fd, None)
if acceptor is None:
netLogger.critical("_onFdWritable failed,client_fd:%d", client_fd)
return
if acceptor.onWriteEvent():
# 更新 epoll 句柄中连接 fd 注册事件为 可读
# self._epoll_loop.modify(client_fd, select.EPOLLET | select.EPOLLIN)
pass
else:
# 出错
self._pushInvalidFd(client_fd)
t = "_onFdWritable error,client_fd:{0}".format(client_fd)
netLogger.critical(t)
# Exception通知
def _onFdExceptional(self, client_fd):
acceptor = self._acceptor_dict.get(client_fd, None)
if acceptor is None:
netLogger.critical("_onFdExceptional failed,client_fd:%d", client_fd)
return
# 出错
t = "_onFdExceptional client_fd:{0}".format(client_fd)
netLogger.critical(t)
self._pushInvalidFd(client_fd)
pass
# 检查失效的acceptors
def _checkInvalidAcceptors(self):
if not self._invalid_acceptorfd_set:
return
for client_fd in self._invalid_acceptorfd_set:
acceptor = self._acceptor_dict.get(client_fd, None)
if acceptor is None:
continue
self._delTcpAcceptor(client_fd, acceptor)
acceptor.onDisconnectEvent()
acceptor.close()
self._invalid_acceptorfd_set.clear()
def _addTcpAcceptor(self, acceptor_fd, acceptor):
# 向 epoll 句柄中注册 连接 socket 的 可读 事件
self._epoll_loop.register(acceptor_fd, select.EPOLLET | select.EPOLLOUT | select.EPOLLIN)
self._acceptor_dict[acceptor_fd] = acceptor
t = "_addTcpAcceptor sessionId:{0} fd:{1} addr:{2} conns:{3}".format(
acceptor.client_session_id,
acceptor_fd,
acceptor.client_addr,
len(self._acceptor_dict))
netLogger.debug(t)
def _delTcpAcceptor(self, acceptor_fd, acceptor):
if acceptor_fd in self._acceptor_dict:
self._epoll_loop.unregister(acceptor_fd)
del self._acceptor_dict[acceptor_fd]
t = "_delTcpAcceptor sessionId:{0} fd:{1} addr:{2} conns:{3}".format(
acceptor.client_session_id,
acceptor_fd,
acceptor.client_addr,
len(self._acceptor_dict))
netLogger.debug(t)
return
t = "_delTcpAcceptor nothing"
netLogger.warning(t)
############################################################################
class _Acceptor(ConnectionBase):
def __init__(self, client_session_id=-1, client_socket=None, client_addr=()):
super(_Acceptor, self).__init__(client_session_id, client_socket, client_addr)
self.connection_type = CONNECT_TYPE.IS_ACCEPTOR
self.connect_status = CONNECT_STATUS.CONNECT_SUCC
self.max_send_buffer_size = 1024 * 1024 * 10
self.max_recv_buffer_size = 1024 * 1024 * 10
self.max_keeplive_time = 600000000000
self.send_count = 0
def onDisconnectEvent(self):
if self.connect_status == CONNECT_STATUS.CONNECT_CLOSED:
return
print "_Acceptor::onDisconnectEvent fd:{0} reason:{1}".format(self.client_socket.fileno(), self.connect_status)
def onTimerEvent(self, current_time):
# if not super(_Acceptor,self).onTimerEvent(current_time):
# return False
#
# tm = time.time()
# data = '2'*1024*60 +"2"* 1024 * self.send_count
# self.send_count += 1
# self.sendData(data)
# print tm,"sendbytes:{0} recvbytes:{1}".format(self.send_bytes,self.recv_bytes)
return True
# 处理收到的数据
def _process_recv_buffer(self):
pid = str(os.getpid())
# print '_Acceptor_{0}-recv:'.format(pid)
# self.sendData(pid + "say:" + self.recv_buffer)
return len(self.recv_buffer)
if __name__ == '__main__':
listen_addr_list = [("0.0.0.0", 1234)]
serv = EpollServer(listen_addr_list)
InterruptableTaskLoop(serv, 0).startAsForver()---
类似资料：