I wrote this server to handle incoming messages in a process using
multiprocessing named "handler", and sending message in a Thread named
"sender", 'cause I think the async_chat object can not pass between
processes.
My project is a network gate server with many complex logic handler
behind, so I use multiprocessing to handle them separately and send
back the clients later when done.
To use the server multicore cpu I tried to separate the send and
receive function in different process but it seems can not be done :)
I just get questions about this design:
1. is async_chat.push thread-safe? 'Cause I found random errors
reporting push fifo queue out of index 0 sometimes
2. is the whole design odd in any way?
here is my code
import asyncore, asynchat
import os, socket, string
from multiprocessing import Process,Manager
import pickle
import _thread
PORT = 80
policyRequest = b"<policy-file-request/>"
policyReturn = b"""<cross-domain-policy>
<allow-access-from domain="*" to-ports="*" />
</cross-domain-policy\x00"""
def handler(taskLis t,msgList):
while 1:
print('getting task')
item = pickle.loads(ta skList.get())
print('item before handle ', item)
item['msg'] += b' hanlded done'
msgList.put(pic kle.dumps(item) )
def findClient(id):
for item in clients:
if item.idx == id:
return item
def sender():
global msgList
while 1:
item = pickle.loads(ms gList.get())
#print time()
c = findClient(item['cid'])
#print time()
c.push(item['msg'])
print('msg sent ',item['msg'])
#print time()
class HTTPChannel(asy nchat.async_cha t):
def __init__(self, server, sock, addr):
global cid;
asynchat.async_ chat.__init__(s elf, sock)
self.set_termin ator(b"\x00")
self.data = b""
cid += 1
self.idx = cid
if not self in clients:
clients.append( self)
def collect_incomin g_data(self, data):
self.data = self.data + data
print(data)
def found_terminato r(self):
global taskList
print("found",s elf.data)
if self.data == policyRequest:
self.push(polic yReturn)
else:
d = {'cid':self.idx ,'msg':self.dat a}
taskList.put(pi ckle.dumps(d))
self.data = b""
def handle_close(se lf):
if self in clients:
clients.remove( self)
class HTTPServer(asyn core.dispatcher ):
def __init__(self, port):
asyncore.dispat cher.__init__(s elf)
self.create_soc ket(socket.AF_I NET, socket.SOCK_STR EAM)
self.bind(("", port))
self.listen(5)
def handle_accept(s elf):
conn, addr = self.accept()
HTTPChannel(sel f, conn, addr)
#
# try it out
if __name__ == "__main__":
s = HTTPServer(PORT )
print ("serving at port", PORT, "...")
#clients sock obj list stored for further use
clients=[]
#client id auto increasement
cid = 0
manager = Manager()
taskList = manager.Queue()
msgList = manager.Queue()
h = Process(target= handler,args=(t askList,msgList ))
h.start()
_thread.start_n ew_thread(sende r,())
print('entering loop')
asyncore.loop()
