nbnet

  1 #!/usr/bin/env python
  2 # coding: utf-8
  3 
  4 from daemon import Daemon
  5 import socket
  6 import select
  7 import time
  8 import pdb
  9 
 10 __all__ = ["nbNet", "sendData_mh"]
 11 #DEBUG = True
 12 
 13 from nbNetUtils import *
 14 
 15 class nbNetBase:
 16     ‘‘‘non-blocking Net‘‘‘
 17     def setFd(self, sock):
 18         """sock is class object of socket"""
 19         #dbgPrint("\n -- setFd start!")
 20         tmp_state = STATE()
 21         tmp_state.sock_obj = sock
 22         self.conn_state[sock.fileno()] = tmp_state
 23         #self.conn_state[sock.fileno()].printState()
 24         #dbgPrint("\n -- setFd end!")
 25 
 26     def accept(self, fd): 
 27         """fd is fileno() of socket"""
 28         #dbgPrint("\n -- accept start!")
 29         sock_state = self.conn_state[fd]
 30         sock = sock_state.sock_obj
 31         conn, addr = sock.accept()
 32         # set to non-blocking: 0
 33         conn.setblocking(0)
 34         return conn
 35     
 36     def close(self, fd):
 37         """fd is fileno() of socket"""
 38         #pdb.set_trace()
 39         print "closing", fd, self.conn_state
 40         try:
 41             # cancel of listen to event
 42             sock = self.conn_state[fd].sock_obj
 43             self.epoll_sock.unregister(fd)
 44             sock.close()
 45             self.conn_state.pop(fd)
 46             tmp_pipe = self.popen_pipe
 47             self.popen_pipe = 0
 48             tmp_pipe.close()
 49         except:
 50             #dbgPrint("Close fd: %s abnormal" % fd)
 51             pass
 52     #@profile
 53     def read(self, fd):
 54         """fd is fileno() of socket"""
 55         #pdb.set_trace()
 56         try:
 57             sock_state = self.conn_state[fd]
 58             conn = sock_state.sock_obj
 59             if sock_state.need_read <= 0:
 60                 raise socket.error
 61 
 62             one_read = conn.recv(sock_state.need_read)
 63             #dbgPrint("\tread func fd: %d, one_read: %s, need_read: %d" % (fd, one_read, sock_state.need_read))
 64             if len(one_read) == 0:
 65                 raise socket.error
 66             # process received data
 67             sock_state.buff_read += one_read
 68             sock_state.have_read += len(one_read)
 69             sock_state.need_read -= len(one_read)
 70             #sock_state.printState()
 71 
 72             # read protocol header
 73             if sock_state.have_read == 10:
 74                 header_said_need_read = int(sock_state.buff_read)
 75                 if header_said_need_read <= 0:
 76                     raise socket.error
 77                 sock_state.need_read += header_said_need_read
 78                 sock_state.buff_read = ‘‘
 79                 # call state machine, current state is read. 
 80                 # after protocol header haven readed, read the real cmd content, 
 81                 # call machine instead of call read() it self in common.
 82                 #sock_state.printState()
 83                 return "readcontent"
 84             elif sock_state.need_read == 0:
 85                 # recv complete, change state to process it
 86                 return "process"
 87             else:
 88                 return "readmore"
 89         except (socket.error, ValueError), msg:
 90             try:
 91                 if msg.errno == 11:
 92                     #dbgPrint("11 " + msg)
 93                     return "retry"
 94             except:
 95                 pass
 96             return ‘closing‘
 97         
 98 
 99     #@profile
100     def write(self, fd):
101         sock_state = self.conn_state[fd]
102         conn = sock_state.sock_obj
103         #pdb.set_trace()
104         
105         if isinstance(sock_state.popen_pipe, file):
106             try:
107                 output = sock_state.popen_pipe.read()
108                 #print output
109             except (IOError, ValueError), msg:
110                 pass
111             #have_send = conn.send("%010d%s" % (len(output), output))
112             #todo
113 
114         else:
115             last_have_send = sock_state.have_write
116             try:
117                 # to send some Bytes, but have_send is the return num of .send()
118                 have_send = conn.send(sock_state.buff_write[last_have_send:])
119                 sock_state.have_write += have_send
120                 sock_state.need_write -= have_send
121                 if sock_state.need_write == 0 and sock_state.have_write != 0:
122                     # send complete, re init status, and listen re-read
123                     #sock_state.printState()
124                     #dbgPrint(‘\n write data completed!‘)
125                     return "writecomplete"
126                 else:
127                     return "writemore"
128             except socket.error, msg:
129                 return "closing"
130 
131 
132     def run(self):
133         while True:
134             #dbgPrint("\nrun func loop:")
135             # print conn_state
136             #for i in self.conn_state.iterkeys():
137                 #dbgPrint("\n - state of fd: %d" % i)
138                 #self.conn_state[i].printState()
139 
140             epoll_list = self.epoll_sock.poll()
141             for fd, events in epoll_list:
142                 #dbgPrint(‘\n-- run epoll return fd: %d. event: %s‘ % (fd, events))
143                 print self.conn_state
144                 print fd, events
145                 sock_state = self.conn_state[fd]
146                 if select.EPOLLHUP & events:
147                     #dbgPrint("EPOLLHUP")
148                     sock_state.state = "closing"
149                 elif select.EPOLLERR & events:
150                     #dbgPrint("EPOLLERR")
151                     sock_state.state = "closing"
152                 self.state_machine(fd)
153 
154     def state_machine(self, fd):
155         #time.sleep(0.1)
156         #dbgPrint("\n - state machine: fd: %d, status: %s" % (fd, self.conn_state[fd].state))
157         sock_state = self.conn_state[fd]
158         self.sm[sock_state.state](fd)
159 
160 class nbNet(nbNetBase):
161     def __init__(self, addr, port, logic):
162         #dbgPrint(‘\n__init__: start!‘)
163         self.conn_state = {}
164         self.listen_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
165         self.listen_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
166         self.listen_sock.bind((addr, port))
167         self.listen_sock.listen(10)
168         self.setFd(self.listen_sock)
169         self.epoll_sock = select.epoll()
170         # LT for default, ET add ‘ | select.EPOLLET ‘
171         self.epoll_sock.register(self.listen_sock.fileno(), select.EPOLLIN )
172         self.logic = logic
173         self.sm = {
174             "accept" : self.accept2read,
175             "read"   : self.read2process,
176             "write"  : self.write2read,
177             "process": self.process,
178             "closing": self.close,
179         }
180         #dbgPrint(‘\n__init__: end, register no: %s‘ % self.listen_sock.fileno() )
181 
182     #@profile
183     def process(self, fd):
184         sock_state = self.conn_state[fd]
185         response = self.logic(fd, sock_state.buff_read)
186         #pdb.set_trace()
187         if response == None:
188             conn = sock_state.sock_obj
189             self.setFd(conn)
190             self.conn_state[fd].state = "read"
191             self.epoll_sock.modify(fd, select.EPOLLIN)
192         else:  
193             sock_state.buff_write = "%010d%s" % (len(response), response)
194             sock_state.need_write = len(sock_state.buff_write)
195             #sock_state.printState()
196             #self.state_machine(fd)
197             sock_state.state = "write"
198             self.epoll_sock.modify(fd, select.EPOLLOUT)
199 
200              
201 
202     #@profile
203     def accept2read(self, fd):
204         conn = self.accept(fd)
205         self.epoll_sock.register(conn.fileno(), select.EPOLLIN)
206         # new client connection fd be initilized 
207         self.setFd(conn)
208         self.conn_state[conn.fileno()].state = "read"
209         # now end of accept, but the main process still on ‘accept‘ status
210         # waiting for new client to connect it.
211         #dbgPrint("\n -- accept end!")
212 
213     #@profile
214     def read2process(self, fd):
215         """fd is fileno() of socket"""
216         #pdb.set_trace()
217         read_ret = ""
218         try:
219             read_ret = self.read(fd)
220         except (Exception), msg:
221             #dbgPrint(msg)
222             read_ret = "closing"
223         if read_ret == "process":
224             # recv complete, change state to process it
225             #sock_state.state = "process"
226             self.process(fd)
227         elif read_ret == "readcontent":
228             pass
229         elif read_ret == "readmore":
230             pass
231         elif read_ret == "retry":
232             pass
233         elif read_ret == "closing":
234             self.conn_state[fd].state = ‘closing‘
235             # closing directly when error.
236             self.state_machine(fd)
237         else:
238             raise Exception("impossible state returned by self.read")
239 
240     #@profile
241     def write2read(self, fd):
242         try:
243             write_ret = self.write(fd)
244         except socket.error, msg:
245             write_ret = "closing"
246 
247         if write_ret == "writemore":
248             pass
249         elif write_ret == "writecomplete":
250             sock_state = self.conn_state[fd]
251             conn = sock_state.sock_obj
252             self.setFd(conn)
253             self.conn_state[fd].state = "read"
254             self.epoll_sock.modify(fd, select.EPOLLIN)
255         elif write_ret == "closing":
256             #dbgPrint(msg)
257             self.conn_state[fd].state = ‘closing‘
258             # closing directly when error.
259             self.state_machine(fd)
260     
261 counter = 0
262 if __name__ == ‘__main__‘:
263     
264     def logic(d_in):
265         global counter
266         counter += 1
267         if counter % 100000 == 0:
268             print counter, time.time()
269         return("a")
270 
271     reverseD = nbNet(‘0.0.0.0‘, 9099, logic)
272     reverseD.run()