greenio.py 18 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537
  1. from eventlet.support import get_errno
  2. from eventlet.hubs import trampoline
  3. BUFFER_SIZE = 4096
  4. import errno
  5. import os
  6. import socket
  7. from socket import socket as _original_socket
  8. import sys
  9. import time
  10. import warnings
  11. __all__ = ['GreenSocket', 'GreenPipe', 'shutdown_safe']
  12. CONNECT_ERR = set((errno.EINPROGRESS, errno.EALREADY, errno.EWOULDBLOCK))
  13. CONNECT_SUCCESS = set((0, errno.EISCONN))
  14. if sys.platform[:3]=="win":
  15. CONNECT_ERR.add(errno.WSAEINVAL) # Bug 67
  16. # Emulate _fileobject class in 3.x implementation
  17. # Eventually this internal socket structure could be replaced with makefile calls.
  18. try:
  19. _fileobject = socket._fileobject
  20. except AttributeError:
  21. def _fileobject(sock, *args, **kwargs):
  22. return _original_socket.makefile(sock, *args, **kwargs)
  23. def socket_connect(descriptor, address):
  24. """
  25. Attempts to connect to the address, returns the descriptor if it succeeds,
  26. returns None if it needs to trampoline, and raises any exceptions.
  27. """
  28. err = descriptor.connect_ex(address)
  29. if err in CONNECT_ERR:
  30. return None
  31. if err not in CONNECT_SUCCESS:
  32. raise socket.error(err, errno.errorcode[err])
  33. return descriptor
  34. def socket_checkerr(descriptor):
  35. err = descriptor.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
  36. if err not in CONNECT_SUCCESS:
  37. raise socket.error(err, errno.errorcode[err])
  38. def socket_accept(descriptor):
  39. """
  40. Attempts to accept() on the descriptor, returns a client,address tuple
  41. if it succeeds; returns None if it needs to trampoline, and raises
  42. any exceptions.
  43. """
  44. try:
  45. return descriptor.accept()
  46. except socket.error, e:
  47. if get_errno(e) == errno.EWOULDBLOCK:
  48. return None
  49. raise
  50. if sys.platform[:3]=="win":
  51. # winsock sometimes throws ENOTCONN
  52. SOCKET_BLOCKING = set((errno.EWOULDBLOCK,))
  53. SOCKET_CLOSED = set((errno.ECONNRESET, errno.ENOTCONN, errno.ESHUTDOWN))
  54. else:
  55. # oddly, on linux/darwin, an unconnected socket is expected to block,
  56. # so we treat ENOTCONN the same as EWOULDBLOCK
  57. SOCKET_BLOCKING = set((errno.EWOULDBLOCK, errno.ENOTCONN))
  58. SOCKET_CLOSED = set((errno.ECONNRESET, errno.ESHUTDOWN, errno.EPIPE))
  59. def set_nonblocking(fd):
  60. """
  61. Sets the descriptor to be nonblocking. Works on many file-like
  62. objects as well as sockets. Only sockets can be nonblocking on
  63. Windows, however.
  64. """
  65. try:
  66. setblocking = fd.setblocking
  67. except AttributeError:
  68. # fd has no setblocking() method. It could be that this version of
  69. # Python predates socket.setblocking(). In that case, we can still set
  70. # the flag "by hand" on the underlying OS fileno using the fcntl
  71. # module.
  72. try:
  73. import fcntl
  74. except ImportError:
  75. # Whoops, Windows has no fcntl module. This might not be a socket
  76. # at all, but rather a file-like object with no setblocking()
  77. # method. In particular, on Windows, pipes don't support
  78. # non-blocking I/O and therefore don't have that method. Which
  79. # means fcntl wouldn't help even if we could load it.
  80. raise NotImplementedError("set_nonblocking() on a file object "
  81. "with no setblocking() method "
  82. "(Windows pipes don't support non-blocking I/O)")
  83. # We managed to import fcntl.
  84. fileno = fd.fileno()
  85. flags = fcntl.fcntl(fileno, fcntl.F_GETFL)
  86. fcntl.fcntl(fileno, fcntl.F_SETFL, flags | os.O_NONBLOCK)
  87. else:
  88. # socket supports setblocking()
  89. setblocking(0)
  90. try:
  91. from socket import _GLOBAL_DEFAULT_TIMEOUT
  92. except ImportError:
  93. _GLOBAL_DEFAULT_TIMEOUT = object()
  94. class GreenSocket(object):
  95. """
  96. Green version of socket.socket class, that is intended to be 100%
  97. API-compatible.
  98. """
  99. def __init__(self, family_or_realsock=socket.AF_INET, *args, **kwargs):
  100. if isinstance(family_or_realsock, (int, long)):
  101. fd = _original_socket(family_or_realsock, *args, **kwargs)
  102. else:
  103. fd = family_or_realsock
  104. assert not args, args
  105. assert not kwargs, kwargs
  106. # import timeout from other socket, if it was there
  107. try:
  108. self._timeout = fd.gettimeout() or socket.getdefaulttimeout()
  109. except AttributeError:
  110. self._timeout = socket.getdefaulttimeout()
  111. set_nonblocking(fd)
  112. self.fd = fd
  113. # when client calls setblocking(0) or settimeout(0) the socket must
  114. # act non-blocking
  115. self.act_non_blocking = False
  116. @property
  117. def _sock(self):
  118. return self
  119. #forward unknown attibutes to fd
  120. # cache the value for future use.
  121. # I do not see any simple attribute which could be changed
  122. # so caching everything in self is fine,
  123. # If we find such attributes - only attributes having __get__ might be cahed.
  124. # For now - I do not want to complicate it.
  125. def __getattr__(self, name):
  126. attr = getattr(self.fd, name)
  127. setattr(self, name, attr)
  128. return attr
  129. def accept(self):
  130. if self.act_non_blocking:
  131. return self.fd.accept()
  132. fd = self.fd
  133. while True:
  134. res = socket_accept(fd)
  135. if res is not None:
  136. client, addr = res
  137. set_nonblocking(client)
  138. return type(self)(client), addr
  139. trampoline(fd, read=True, timeout=self.gettimeout(),
  140. timeout_exc=socket.timeout("timed out"))
  141. def connect(self, address):
  142. if self.act_non_blocking:
  143. return self.fd.connect(address)
  144. fd = self.fd
  145. if self.gettimeout() is None:
  146. while not socket_connect(fd, address):
  147. trampoline(fd, write=True)
  148. socket_checkerr(fd)
  149. else:
  150. end = time.time() + self.gettimeout()
  151. while True:
  152. if socket_connect(fd, address):
  153. return
  154. if time.time() >= end:
  155. raise socket.timeout("timed out")
  156. trampoline(fd, write=True, timeout=end-time.time(),
  157. timeout_exc=socket.timeout("timed out"))
  158. socket_checkerr(fd)
  159. def connect_ex(self, address):
  160. if self.act_non_blocking:
  161. return self.fd.connect_ex(address)
  162. fd = self.fd
  163. if self.gettimeout() is None:
  164. while not socket_connect(fd, address):
  165. try:
  166. trampoline(fd, write=True)
  167. socket_checkerr(fd)
  168. except socket.error, ex:
  169. return get_errno(ex)
  170. else:
  171. end = time.time() + self.gettimeout()
  172. while True:
  173. try:
  174. if socket_connect(fd, address):
  175. return 0
  176. if time.time() >= end:
  177. raise socket.timeout(errno.EAGAIN)
  178. trampoline(fd, write=True, timeout=end-time.time(),
  179. timeout_exc=socket.timeout(errno.EAGAIN))
  180. socket_checkerr(fd)
  181. except socket.error, ex:
  182. return get_errno(ex)
  183. def dup(self, *args, **kw):
  184. sock = self.fd.dup(*args, **kw)
  185. set_nonblocking(sock)
  186. newsock = type(self)(sock)
  187. newsock.settimeout(self.gettimeout())
  188. return newsock
  189. def makefile(self, *args, **kw):
  190. return _fileobject(self.dup(), *args, **kw)
  191. def makeGreenFile(self, *args, **kw):
  192. warnings.warn("makeGreenFile has been deprecated, please use "
  193. "makefile instead", DeprecationWarning, stacklevel=2)
  194. return self.makefile(*args, **kw)
  195. def recv(self, buflen, flags=0):
  196. fd = self.fd
  197. if self.act_non_blocking:
  198. return fd.recv(buflen, flags)
  199. while True:
  200. try:
  201. return fd.recv(buflen, flags)
  202. except socket.error, e:
  203. if get_errno(e) in SOCKET_BLOCKING:
  204. pass
  205. elif get_errno(e) in SOCKET_CLOSED:
  206. return ''
  207. else:
  208. raise
  209. trampoline(fd,
  210. read=True,
  211. timeout=self.gettimeout(),
  212. timeout_exc=socket.timeout("timed out"))
  213. def recvfrom(self, *args):
  214. if not self.act_non_blocking:
  215. trampoline(self.fd, read=True, timeout=self.gettimeout(),
  216. timeout_exc=socket.timeout("timed out"))
  217. return self.fd.recvfrom(*args)
  218. def recvfrom_into(self, *args):
  219. if not self.act_non_blocking:
  220. trampoline(self.fd, read=True, timeout=self.gettimeout(),
  221. timeout_exc=socket.timeout("timed out"))
  222. return self.fd.recvfrom_into(*args)
  223. def recv_into(self, *args):
  224. if not self.act_non_blocking:
  225. trampoline(self.fd, read=True, timeout=self.gettimeout(),
  226. timeout_exc=socket.timeout("timed out"))
  227. return self.fd.recv_into(*args)
  228. def send(self, data, flags=0):
  229. fd = self.fd
  230. if self.act_non_blocking:
  231. return fd.send(data, flags)
  232. # blocking socket behavior - sends all, blocks if the buffer is full
  233. total_sent = 0
  234. len_data = len(data)
  235. while 1:
  236. try:
  237. total_sent += fd.send(data[total_sent:], flags)
  238. except socket.error, e:
  239. if get_errno(e) not in SOCKET_BLOCKING and get_errno(e) not in SOCKET_CLOSED:
  240. raise
  241. if total_sent == len_data:
  242. break
  243. trampoline(self.fd, write=True, timeout=self.gettimeout(),
  244. timeout_exc=socket.timeout("timed out"))
  245. return total_sent
  246. def sendall(self, data, flags=0):
  247. tail = self.send(data, flags)
  248. len_data = len(data)
  249. while tail < len_data:
  250. tail += self.send(data[tail:], flags)
  251. def sendto(self, *args):
  252. trampoline(self.fd, write=True)
  253. return self.fd.sendto(*args)
  254. def setblocking(self, flag):
  255. if flag:
  256. self.act_non_blocking = False
  257. self._timeout = None
  258. else:
  259. self.act_non_blocking = True
  260. self._timeout = 0.0
  261. def settimeout(self, howlong):
  262. if howlong is None or howlong == _GLOBAL_DEFAULT_TIMEOUT:
  263. self.setblocking(True)
  264. return
  265. try:
  266. f = howlong.__float__
  267. except AttributeError:
  268. raise TypeError('a float is required')
  269. howlong = f()
  270. if howlong < 0.0:
  271. raise ValueError('Timeout value out of range')
  272. if howlong == 0.0:
  273. self.setblocking(howlong)
  274. else:
  275. self._timeout = howlong
  276. def gettimeout(self):
  277. return self._timeout
  278. class _SocketDuckForFd(object):
  279. """ Class implementing all socket method used by _fileobject in cooperative manner using low level os I/O calls."""
  280. def __init__(self, fileno):
  281. self._fileno = fileno
  282. @property
  283. def _sock(self):
  284. return self
  285. def fileno(self):
  286. return self._fileno
  287. def recv(self, buflen):
  288. while True:
  289. try:
  290. data = os.read(self._fileno, buflen)
  291. return data
  292. except OSError, e:
  293. if get_errno(e) != errno.EAGAIN:
  294. raise IOError(*e.args)
  295. trampoline(self, read=True)
  296. def sendall(self, data):
  297. len_data = len(data)
  298. os_write = os.write
  299. fileno = self._fileno
  300. try:
  301. total_sent = os_write(fileno, data)
  302. except OSError, e:
  303. if get_errno(e) != errno.EAGAIN:
  304. raise IOError(*e.args)
  305. total_sent = 0
  306. while total_sent <len_data:
  307. trampoline(self, write=True)
  308. try:
  309. total_sent += os_write(fileno, data[total_sent:])
  310. except OSError, e:
  311. if get_errno(e) != errno. EAGAIN:
  312. raise IOError(*e.args)
  313. def __del__(self):
  314. try:
  315. os.close(self._fileno)
  316. except:
  317. # os.close may fail if __init__ didn't complete (i.e file dscriptor passed to popen was invalid
  318. pass
  319. def __repr__(self):
  320. return "%s:%d" % (self.__class__.__name__, self._fileno)
  321. def _operationOnClosedFile(*args, **kwargs):
  322. raise ValueError("I/O operation on closed file")
  323. class GreenPipe(_fileobject):
  324. """
  325. GreenPipe is a cooperative replacement for file class.
  326. It will cooperate on pipes. It will block on regular file.
  327. Differneces from file class:
  328. - mode is r/w property. Should re r/o
  329. - encoding property not implemented
  330. - write/writelines will not raise TypeError exception when non-string data is written
  331. it will write str(data) instead
  332. - Universal new lines are not supported and newlines property not implementeded
  333. - file argument can be descriptor, file name or file object.
  334. """
  335. def __init__(self, f, mode='r', bufsize=-1):
  336. if not isinstance(f, (basestring, int, file)):
  337. raise TypeError('f(ile) should be int, str, unicode or file, not %r' % f)
  338. if isinstance(f, basestring):
  339. f = open(f, mode, 0)
  340. if isinstance(f, int):
  341. fileno = f
  342. self._name = "<fd:%d>" % fileno
  343. else:
  344. fileno = os.dup(f.fileno())
  345. self._name = f.name
  346. if f.mode != mode:
  347. raise ValueError('file.mode %r does not match mode parameter %r' % (f.mode, mode))
  348. self._name = f.name
  349. f.close()
  350. super(GreenPipe, self).__init__(_SocketDuckForFd(fileno), mode, bufsize)
  351. set_nonblocking(self)
  352. self.softspace = 0
  353. @property
  354. def name(self): return self._name
  355. def __repr__(self):
  356. return "<%s %s %r, mode %r at 0x%x>" % (
  357. self.closed and 'closed' or 'open',
  358. self.__class__.__name__,
  359. self.name,
  360. self.mode,
  361. (id(self) < 0) and (sys.maxint +id(self)) or id(self))
  362. def close(self):
  363. super(GreenPipe, self).close()
  364. for method in ['fileno', 'flush', 'isatty', 'next', 'read', 'readinto',
  365. 'readline', 'readlines', 'seek', 'tell', 'truncate',
  366. 'write', 'xreadlines', '__iter__', 'writelines']:
  367. setattr(self, method, _operationOnClosedFile)
  368. if getattr(file, '__enter__', None):
  369. def __enter__(self):
  370. return self
  371. def __exit__(self, *args):
  372. self.close()
  373. def xreadlines(self, buffer):
  374. return iterator(self)
  375. def readinto(self, buf):
  376. data = self.read(len(buf)) #FIXME could it be done without allocating intermediate?
  377. n = len(data)
  378. try:
  379. buf[:n] = data
  380. except TypeError, err:
  381. if not isinstance(buf, array.array):
  382. raise err
  383. buf[:n] = array.array('c', data)
  384. return n
  385. def _get_readahead_len(self):
  386. try:
  387. return len(self._rbuf.getvalue()) # StringIO in 2.5
  388. except AttributeError:
  389. return len(self._rbuf) # str in 2.4
  390. def _clear_readahead_buf(self):
  391. len = self._get_readahead_len()
  392. if len>0:
  393. self.read(len)
  394. def tell(self):
  395. self.flush()
  396. try:
  397. return os.lseek(self.fileno(), 0, 1) - self._get_readahead_len()
  398. except OSError, e:
  399. raise IOError(*e.args)
  400. def seek(self, offset, whence=0):
  401. self.flush()
  402. if whence == 1 and offset==0: # tell synonym
  403. return self.tell()
  404. if whence == 1: # adjust offset by what is read ahead
  405. offset -= self.get_readahead_len()
  406. try:
  407. rv = os.lseek(self.fileno(), offset, whence)
  408. except OSError, e:
  409. raise IOError(*e.args)
  410. else:
  411. self._clear_readahead_buf()
  412. return rv
  413. if getattr(file, "truncate", None): # not all OSes implement truncate
  414. def truncate(self, size=-1):
  415. self.flush()
  416. if size ==-1:
  417. size = self.tell()
  418. try:
  419. rv = os.ftruncate(self.fileno(), size)
  420. except OSError, e:
  421. raise IOError(*e.args)
  422. else:
  423. self.seek(size) # move position&clear buffer
  424. return rv
  425. def isatty(self):
  426. try:
  427. return os.isatty(self.fileno())
  428. except OSError, e:
  429. raise IOError(*e.args)
  430. # import SSL module here so we can refer to greenio.SSL.exceptionclass
  431. try:
  432. from OpenSSL import SSL
  433. except ImportError:
  434. # pyOpenSSL not installed, define exceptions anyway for convenience
  435. class SSL(object):
  436. class WantWriteError(object):
  437. pass
  438. class WantReadError(object):
  439. pass
  440. class ZeroReturnError(object):
  441. pass
  442. class SysCallError(object):
  443. pass
  444. def shutdown_safe(sock):
  445. """ Shuts down the socket. This is a convenience method for
  446. code that wants to gracefully handle regular sockets, SSL.Connection
  447. sockets from PyOpenSSL and ssl.SSLSocket objects from Python 2.6
  448. interchangeably. Both types of ssl socket require a shutdown() before
  449. close, but they have different arity on their shutdown method.
  450. Regular sockets don't need a shutdown before close, but it doesn't hurt.
  451. """
  452. try:
  453. try:
  454. # socket, ssl.SSLSocket
  455. return sock.shutdown(socket.SHUT_RDWR)
  456. except TypeError:
  457. # SSL.Connection
  458. return sock.shutdown()
  459. except socket.error, e:
  460. # we don't care if the socket is already closed;
  461. # this will often be the case in an http server context
  462. if get_errno(e) != errno.ENOTCONN:
  463. raise