epoll转发进程

1380阅读　0评论2013-06-19　lolizeppelin
分类：Python/Ruby

这个应该是转载的

点击(此处)折叠或打开

  1. #! /usr/bin/python
  2. # -*- coding: UTF-8 -*-

  4. import sys,time,select,threading
  5. import struct,Queue,copy

  7. class TalkServer(object):
  8.     MinMassageLen = 400
  9.     MaxMassageLen = 1200
  10.     def __init__(self):
  11.         self.sock = None
  12.         self.bindEopll = None
  13.         self.clientEopll = None
  14.         self.connectionPoll = {}
  15.         self.clientLock = None
  16.         self.cond = None
  17.         self.dataQueue = None

  19.     def __start__(self):
  20.         self.bindEopll = select.epoll()
  21.         self.clientEopll = select.epoll()
  22.         self.bindEopll.register(self.sock.fileno(), select.EPOLLIN)
  23.         self.clientLock = threading.Lock()
  24.         self.cond = threading.Condition()
  25.         self.dataQueue = Queue.Queue()

  27.     def __bindDeamon(self):
  28.         while True:
  29.             events = self.bindEopll.poll(1)
  30.             for fileno, event in events:
  31.                 if fileno == self.sock.fileno():
  32.                     self.clientLock.acquire()
  33.                     newConnection, address = self.sock.accept()
  34.                     self.clientEopll.register(newConnection.fileno(), select.EPOLLIN)
  35. # 保存客户端socket文件符fileno及其对应的连接,同时初始化一个0字节的str,用于存放分包数据
  36.                     self.connectionPoll[newConnection.fileno()] = [newConnection,'']
  37.                     self.clientLock.release()
  38.                 else:
  39.                     self.clientLock.acquire()
  40.                     for clientFileno in self.connectionPoll.keys():
  41.                         self.clientEopll.unregister(clientFileno)
  42.                         self.connectionPoll[clientFileno][0].close()
  43.                     self.connectionPoll.clear()
  44.                     self.clientEopll.close()
  45.                     self.bindEopll.unregister(self.sock.fileno())
  46.                     self.bindEopll.close()
  47.                     self.sock.close()
  48.                     self.clientLock.release()

  50.     def __clientDeamon(self):
  51.         while True:
  52.             events = self.clientEopll.poll(0.01)
  53.             for fileno, event in events:
  54.                 if event & select.EPOLLIN:
  55.                     data = self.connectionPoll[fileno].recv(1024)
  56.                     if len(data) > 0:
  57.                         self.cond.acquire()
  58.                         self.__unpackDataToQueue(data, fileno)
  59.                         self.cond.release()
  60.                 else:
  61.                     self.clientLock.acquire()
  62.                     self.clientEopll.unregister(fileno)
  63.                     self.connectionPoll[fileno][0].close()
  64.                     del self.connectionPoll[fileno]
  65.                     self.clientLock.release()

  67.     def __unpackDataToQueue(self,data,sockFileno):
  68.         """An iteraor to unpack data from socket buffer"""
  69.         if self.connectionPoll[sockFileno][1] != '':
  70. # 上次解包有未处理的分包数据
  71.             if len(self.connectionPoll[sockFileno][1])<8:
  72. # 上次遗留的分包长度小于包头长度
  73.                 if len(self.connectionPoll[sockFileno][1]) + len(data) < 8:
  74. # 合并后长度依旧不小于包头长度,直接合并两次数据包
  75.                     self.connectionPoll[sockFileno][1] += data
  76.                     return
  77.                 try:
  78. # 尝试将上次遗留数据与新数据合并
  79.                     packMark,dataLen = struct.unpack('Ii',self.connectionPoll[sockFileno][1] + data)
  80. # 检查合并数据包头,包头错误直接丢弃上次数据
  81.                     if (packMark != 1346 or ((dataLen > MaxMassageLen - 4) or (dataLen < MinMassageLen + 4))):
  82.                         self.connectionPoll[sockFileno][1] = ''
  83.                         __unpackDataToQueue(data,sockFileno)
  84.                         return
  85.                     else:
  86. # 包头正确合并两数据包并清除保存遗留数据的变量,递归调用解包函数
  87.                         data = self.connectionPoll[sockFileno][1] + data
  88.                         self.connectionPoll[sockFileno][1] = ''
  89.                         __unpackDataToQueue(data,sockFileno)
  90.                         return
  91.                 except:
  92.                     self.connectionPoll[sockFileno][1] = ''
  93.                     __unpackDataToQueue(data,sockFileno)
  94.                     return
  95.             else:
  96.                 data = self.connectionPoll[sockFileno][1] + data
  97.                 self.connectionPoll[sockFileno][1] = ''
  98.                 __unpackDataToQueue(data,sockFileno)
  99.                 return
  100.         if len(data)>8:
  101. # 数据包长度大于包头长度
  102.             try:
  103.                 packMark,dataLen = struct.unpack('Ii',data)
  104.                 if (packMark != 1346 or ((dataLen > MaxMassageLen - 4) or (dataLen < MinMassageLen + 4))):
  105.                     return
  106.             except:
  107.                 return
  108.             if dataLen > len(data) - 8:
  109.                 self.connectionPoll[sockFileno][1] = data
  110.                 return
  111.             elif (dataLen == len(data)):
  112. # self.cond.acquire()
  113.                 self.dataQueue.put((data[8:],sockFileno),False)
  114. # self.cond.release()
  115.                 return
  116.             elif dataLen < len(data) - 8:
  117. # self.cond.acquire()
  118.                 self.dataQueue.put((data[8:dataLen + 8],sockFileno),False)
  119. # self.cond.release()
  120.                 __unpackDataToQueue(data[dataLen + 8:],sockFileno)
  121.                 return
  122.         else:
  123. # 数据包长度小于包头长度,保存当前数据分包
  124.             self.connectionPoll[sockFileno][1] = data
  125.             return


  128.     def __talkDeamon(self):
  129.         time.sleep(5)
  130.         while True:
  131.             self.cond.acquire()
  132.             if(self.dataQueue.empty()):
  133.                 self.cond.wait()
  134.             while(not self.dataQueue.empty()):
  135.                 package = self.dataQueue.get()
  136.             self.cond.release()


1
上一篇：man查询
下一篇：终于找到high resolution timer的相关内核启动参数了