stream.pyx 31 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016
  1. cdef enum:
  2. __PREALLOCED_BUFS = 4
  3. @cython.no_gc_clear
  4. @cython.freelist(DEFAULT_FREELIST_SIZE)
  5. cdef class _StreamWriteContext:
  6. # used to hold additional write request information for uv_write
  7. cdef:
  8. uv.uv_write_t req
  9. list buffers
  10. uv.uv_buf_t uv_bufs_sml[__PREALLOCED_BUFS]
  11. Py_buffer py_bufs_sml[__PREALLOCED_BUFS]
  12. bint py_bufs_sml_inuse
  13. uv.uv_buf_t* uv_bufs
  14. Py_buffer* py_bufs
  15. size_t py_bufs_len
  16. uv.uv_buf_t* uv_bufs_start
  17. size_t uv_bufs_len
  18. UVStream stream
  19. bint closed
  20. cdef free_bufs(self):
  21. cdef size_t i
  22. if self.uv_bufs is not NULL:
  23. PyMem_RawFree(self.uv_bufs)
  24. self.uv_bufs = NULL
  25. if UVLOOP_DEBUG:
  26. if self.py_bufs_sml_inuse:
  27. raise RuntimeError(
  28. '_StreamWriteContext.close: uv_bufs != NULL and '
  29. 'py_bufs_sml_inuse is True')
  30. if self.py_bufs is not NULL:
  31. for i from 0 <= i < self.py_bufs_len:
  32. PyBuffer_Release(&self.py_bufs[i])
  33. PyMem_RawFree(self.py_bufs)
  34. self.py_bufs = NULL
  35. if UVLOOP_DEBUG:
  36. if self.py_bufs_sml_inuse:
  37. raise RuntimeError(
  38. '_StreamWriteContext.close: py_bufs != NULL and '
  39. 'py_bufs_sml_inuse is True')
  40. if self.py_bufs_sml_inuse:
  41. for i from 0 <= i < self.py_bufs_len:
  42. PyBuffer_Release(&self.py_bufs_sml[i])
  43. self.py_bufs_sml_inuse = 0
  44. self.py_bufs_len = 0
  45. self.buffers = None
  46. cdef close(self):
  47. if self.closed:
  48. return
  49. self.closed = 1
  50. self.free_bufs()
  51. Py_DECREF(self)
  52. cdef advance_uv_buf(self, size_t sent):
  53. # Advance the pointer to first uv_buf and the
  54. # pointer to first byte in that buffer.
  55. #
  56. # We do this after a "uv_try_write" call, which
  57. # sometimes sends only a portion of data.
  58. # We then call "advance_uv_buf" on the write
  59. # context, and reuse it in a "uv_write" call.
  60. cdef:
  61. uv.uv_buf_t* buf
  62. size_t idx
  63. for idx from 0 <= idx < self.uv_bufs_len:
  64. buf = &self.uv_bufs_start[idx]
  65. if buf.len > sent:
  66. buf.len -= sent
  67. buf.base = buf.base + sent
  68. self.uv_bufs_start = buf
  69. self.uv_bufs_len -= idx
  70. return
  71. else:
  72. sent -= self.uv_bufs_start[idx].len
  73. if UVLOOP_DEBUG:
  74. if sent < 0:
  75. raise RuntimeError('fatal: sent < 0 in advance_uv_buf')
  76. raise RuntimeError('fatal: Could not advance _StreamWriteContext')
  77. @staticmethod
  78. cdef _StreamWriteContext new(UVStream stream, list buffers):
  79. cdef:
  80. _StreamWriteContext ctx
  81. int uv_bufs_idx = 0
  82. size_t py_bufs_len = 0
  83. int i
  84. Py_buffer* p_pybufs
  85. uv.uv_buf_t* p_uvbufs
  86. ctx = _StreamWriteContext.__new__(_StreamWriteContext)
  87. ctx.stream = None
  88. ctx.closed = 1
  89. ctx.py_bufs_len = 0
  90. ctx.py_bufs_sml_inuse = 0
  91. ctx.uv_bufs = NULL
  92. ctx.py_bufs = NULL
  93. ctx.buffers = buffers
  94. ctx.stream = stream
  95. if len(buffers) <= __PREALLOCED_BUFS:
  96. # We've got a small number of buffers to write, don't
  97. # need to use malloc.
  98. ctx.py_bufs_sml_inuse = 1
  99. p_pybufs = <Py_buffer*>&ctx.py_bufs_sml
  100. p_uvbufs = <uv.uv_buf_t*>&ctx.uv_bufs_sml
  101. else:
  102. for buf in buffers:
  103. if UVLOOP_DEBUG:
  104. if not isinstance(buf, (bytes, bytearray, memoryview)):
  105. raise RuntimeError(
  106. 'invalid data in writebuf: an instance of '
  107. 'bytes, bytearray or memoryview was expected, '
  108. 'got {}'.format(type(buf)))
  109. if not PyBytes_CheckExact(buf):
  110. py_bufs_len += 1
  111. if py_bufs_len > 0:
  112. ctx.py_bufs = <Py_buffer*>PyMem_RawMalloc(
  113. py_bufs_len * sizeof(Py_buffer))
  114. if ctx.py_bufs is NULL:
  115. raise MemoryError()
  116. ctx.uv_bufs = <uv.uv_buf_t*>PyMem_RawMalloc(
  117. len(buffers) * sizeof(uv.uv_buf_t))
  118. if ctx.uv_bufs is NULL:
  119. raise MemoryError()
  120. p_pybufs = ctx.py_bufs
  121. p_uvbufs = ctx.uv_bufs
  122. py_bufs_len = 0
  123. for buf in buffers:
  124. if PyBytes_CheckExact(buf):
  125. # We can only use this hack for bytes since it's
  126. # immutable. For everything else it is only safe to
  127. # use buffer protocol.
  128. p_uvbufs[uv_bufs_idx].base = PyBytes_AS_STRING(buf)
  129. p_uvbufs[uv_bufs_idx].len = Py_SIZE(buf)
  130. else:
  131. try:
  132. PyObject_GetBuffer(
  133. buf, &p_pybufs[py_bufs_len], PyBUF_SIMPLE)
  134. except Exception:
  135. # This shouldn't ever happen, as `UVStream._buffer_write`
  136. # casts non-bytes objects to `memoryviews`.
  137. ctx.py_bufs_len = py_bufs_len
  138. ctx.free_bufs()
  139. raise
  140. p_uvbufs[uv_bufs_idx].base = <char*>p_pybufs[py_bufs_len].buf
  141. p_uvbufs[uv_bufs_idx].len = p_pybufs[py_bufs_len].len
  142. py_bufs_len += 1
  143. uv_bufs_idx += 1
  144. ctx.uv_bufs_start = p_uvbufs
  145. ctx.uv_bufs_len = uv_bufs_idx
  146. ctx.py_bufs_len = py_bufs_len
  147. ctx.req.data = <void*> ctx
  148. if UVLOOP_DEBUG:
  149. stream._loop._debug_stream_write_ctx_total += 1
  150. stream._loop._debug_stream_write_ctx_cnt += 1
  151. # Do incref after everything else is done.
  152. # Under no circumstances we want `ctx` to be GCed while
  153. # libuv is still working with `ctx.uv_bufs`.
  154. Py_INCREF(ctx)
  155. ctx.closed = 0
  156. return ctx
  157. def __dealloc__(self):
  158. if not self.closed:
  159. # Because we do an INCREF in _StreamWriteContext.new,
  160. # __dealloc__ shouldn't ever happen with `self.closed == 1`
  161. raise RuntimeError(
  162. 'open _StreamWriteContext is being deallocated')
  163. if UVLOOP_DEBUG:
  164. if self.stream is not None:
  165. self.stream._loop._debug_stream_write_ctx_cnt -= 1
  166. self.stream = None
  167. @cython.no_gc_clear
  168. cdef class UVStream(UVBaseTransport):
  169. def __cinit__(self):
  170. self.__shutting_down = 0
  171. self.__reading = 0
  172. self.__read_error_close = 0
  173. self.__buffered = 0
  174. self._eof = 0
  175. self._buffer = []
  176. self._buffer_size = 0
  177. self._protocol_get_buffer = None
  178. self._protocol_buffer_updated = None
  179. self._read_pybuf_acquired = False
  180. cdef _set_protocol(self, object protocol):
  181. if protocol is None:
  182. raise TypeError('protocol is required')
  183. UVBaseTransport._set_protocol(self, protocol)
  184. if (hasattr(protocol, 'get_buffer') and
  185. not isinstance(protocol, aio_Protocol)):
  186. try:
  187. self._protocol_get_buffer = protocol.get_buffer
  188. self._protocol_buffer_updated = protocol.buffer_updated
  189. self.__buffered = 1
  190. except AttributeError:
  191. pass
  192. else:
  193. self.__buffered = 0
  194. cdef _clear_protocol(self):
  195. UVBaseTransport._clear_protocol(self)
  196. self._protocol_get_buffer = None
  197. self._protocol_buffer_updated = None
  198. self.__buffered = 0
  199. cdef inline _shutdown(self):
  200. cdef int err
  201. if self.__shutting_down:
  202. return
  203. self.__shutting_down = 1
  204. self._ensure_alive()
  205. self._shutdown_req.data = <void*> self
  206. err = uv.uv_shutdown(&self._shutdown_req,
  207. <uv.uv_stream_t*> self._handle,
  208. __uv_stream_on_shutdown)
  209. if err < 0:
  210. exc = convert_error(err)
  211. self._fatal_error(exc, True)
  212. return
  213. cdef inline _accept(self, UVStream server):
  214. cdef int err
  215. self._ensure_alive()
  216. err = uv.uv_accept(<uv.uv_stream_t*>server._handle,
  217. <uv.uv_stream_t*>self._handle)
  218. if err < 0:
  219. exc = convert_error(err)
  220. self._fatal_error(exc, True)
  221. return
  222. self._on_accept()
  223. cdef inline _close_on_read_error(self):
  224. self.__read_error_close = 1
  225. cdef bint _is_reading(self):
  226. return self.__reading
  227. cdef _start_reading(self):
  228. cdef int err
  229. if self._closing:
  230. return
  231. self._ensure_alive()
  232. if self.__reading:
  233. return
  234. if self.__buffered:
  235. err = uv.uv_read_start(<uv.uv_stream_t*>self._handle,
  236. __uv_stream_buffered_alloc,
  237. __uv_stream_buffered_on_read)
  238. else:
  239. err = uv.uv_read_start(<uv.uv_stream_t*>self._handle,
  240. __loop_alloc_buffer,
  241. __uv_stream_on_read)
  242. if err < 0:
  243. exc = convert_error(err)
  244. self._fatal_error(exc, True)
  245. return
  246. else:
  247. # UVStream must live until the read callback is called
  248. self.__reading_started()
  249. cdef inline __reading_started(self):
  250. if self.__reading:
  251. return
  252. self.__reading = 1
  253. Py_INCREF(self)
  254. cdef inline __reading_stopped(self):
  255. if not self.__reading:
  256. return
  257. self.__reading = 0
  258. Py_DECREF(self)
  259. cdef _stop_reading(self):
  260. cdef int err
  261. if not self.__reading:
  262. return
  263. self._ensure_alive()
  264. # From libuv docs:
  265. # This function is idempotent and may be safely
  266. # called on a stopped stream.
  267. err = uv.uv_read_stop(<uv.uv_stream_t*>self._handle)
  268. if err < 0:
  269. exc = convert_error(err)
  270. self._fatal_error(exc, True)
  271. return
  272. else:
  273. self.__reading_stopped()
  274. cdef inline _try_write(self, object data):
  275. cdef:
  276. ssize_t written
  277. bint used_buf = 0
  278. Py_buffer py_buf
  279. void* buf
  280. size_t blen
  281. int saved_errno
  282. int fd
  283. if (<uv.uv_stream_t*>self._handle).write_queue_size != 0:
  284. raise RuntimeError(
  285. 'UVStream._try_write called with data in uv buffers')
  286. if PyBytes_CheckExact(data):
  287. # We can only use this hack for bytes since it's
  288. # immutable. For everything else it is only safe to
  289. # use buffer protocol.
  290. buf = <void*>PyBytes_AS_STRING(data)
  291. blen = Py_SIZE(data)
  292. else:
  293. PyObject_GetBuffer(data, &py_buf, PyBUF_SIMPLE)
  294. used_buf = 1
  295. buf = py_buf.buf
  296. blen = py_buf.len
  297. if blen == 0:
  298. # Empty data, do nothing.
  299. return 0
  300. fd = self._fileno()
  301. # Use `unistd.h/write` directly, it's faster than
  302. # uv_try_write -- less layers of code. The error
  303. # checking logic is copied from libuv.
  304. written = system.write(fd, buf, blen)
  305. while written == -1 and (
  306. errno.errno == errno.EINTR or
  307. (system.PLATFORM_IS_APPLE and
  308. errno.errno == errno.EPROTOTYPE)):
  309. # From libuv code (unix/stream.c):
  310. # Due to a possible kernel bug at least in OS X 10.10 "Yosemite",
  311. # EPROTOTYPE can be returned while trying to write to a socket
  312. # that is shutting down. If we retry the write, we should get
  313. # the expected EPIPE instead.
  314. written = system.write(fd, buf, blen)
  315. saved_errno = errno.errno
  316. if used_buf:
  317. PyBuffer_Release(&py_buf)
  318. if written < 0:
  319. if saved_errno == errno.EAGAIN or \
  320. saved_errno == system.EWOULDBLOCK:
  321. return -1
  322. else:
  323. exc = convert_error(-saved_errno)
  324. self._fatal_error(exc, True)
  325. return
  326. if UVLOOP_DEBUG:
  327. self._loop._debug_stream_write_tries += 1
  328. if <size_t>written == blen:
  329. return 0
  330. return written
  331. cdef inline _buffer_write(self, object data):
  332. cdef int dlen
  333. if not PyBytes_CheckExact(data):
  334. data = memoryview(data).cast('b')
  335. dlen = len(data)
  336. if not dlen:
  337. return
  338. self._buffer_size += dlen
  339. self._buffer.append(data)
  340. cdef inline _initiate_write(self):
  341. if (not self._protocol_paused and
  342. (<uv.uv_stream_t*>self._handle).write_queue_size == 0 and
  343. self._buffer_size > self._high_water):
  344. # Fast-path. If:
  345. # - the protocol isn't yet paused,
  346. # - there is no data in libuv buffers for this stream,
  347. # - the protocol will be paused if we continue to buffer data
  348. #
  349. # Then:
  350. # - Try to write all buffered data right now.
  351. all_sent = self._exec_write()
  352. if UVLOOP_DEBUG:
  353. if self._buffer_size != 0 or self._buffer != []:
  354. raise RuntimeError(
  355. '_buffer_size is not 0 after a successful _exec_write')
  356. # There is no need to call `_queue_write` anymore,
  357. # as `uv_write` should be called already.
  358. if not all_sent:
  359. # If not all of the data was sent successfully,
  360. # we might need to pause the protocol.
  361. self._maybe_pause_protocol()
  362. elif self._buffer_size > 0:
  363. self._maybe_pause_protocol()
  364. self._loop._queue_write(self)
  365. cdef inline _exec_write(self):
  366. cdef:
  367. int err
  368. int buf_len
  369. _StreamWriteContext ctx = None
  370. if self._closed:
  371. # If the handle is closed, just return, it's too
  372. # late to do anything.
  373. return
  374. buf_len = len(self._buffer)
  375. if not buf_len:
  376. return
  377. if (<uv.uv_stream_t*>self._handle).write_queue_size == 0:
  378. # libuv internal write buffers for this stream are empty.
  379. if buf_len == 1:
  380. # If we only have one piece of data to send, let's
  381. # use our fast implementation of try_write.
  382. data = self._buffer[0]
  383. sent = self._try_write(data)
  384. if sent is None:
  385. # A `self._fatal_error` was called.
  386. # It might not raise an exception under some
  387. # conditions.
  388. self._buffer_size = 0
  389. self._buffer.clear()
  390. if not self._closing:
  391. # This should never happen.
  392. raise RuntimeError(
  393. 'stream is open after UVStream._try_write '
  394. 'returned None')
  395. return
  396. if sent == 0:
  397. # All data was successfully written.
  398. self._buffer_size = 0
  399. self._buffer.clear()
  400. # on_write will call "maybe_resume_protocol".
  401. self._on_write()
  402. return True
  403. if sent > 0:
  404. if UVLOOP_DEBUG:
  405. if sent == len(data):
  406. raise RuntimeError(
  407. '_try_write sent all data and returned '
  408. 'non-zero')
  409. if PyBytes_CheckExact(data):
  410. # Cast bytes to memoryview to avoid copying
  411. # data that wasn't sent.
  412. data = memoryview(data)
  413. data = data[sent:]
  414. self._buffer_size -= sent
  415. self._buffer[0] = data
  416. # At this point it's either data was sent partially,
  417. # or an EAGAIN has happened.
  418. else:
  419. ctx = _StreamWriteContext.new(self, self._buffer)
  420. err = uv.uv_try_write(<uv.uv_stream_t*>self._handle,
  421. ctx.uv_bufs_start,
  422. ctx.uv_bufs_len)
  423. if err > 0:
  424. # Some data was successfully sent.
  425. if <size_t>err == self._buffer_size:
  426. # Everything was sent.
  427. ctx.close()
  428. self._buffer.clear()
  429. self._buffer_size = 0
  430. # on_write will call "maybe_resume_protocol".
  431. self._on_write()
  432. return True
  433. try:
  434. # Advance pointers to uv_bufs in `ctx`,
  435. # we will reuse it soon for a uv_write
  436. # call.
  437. ctx.advance_uv_buf(<ssize_t>err)
  438. except Exception as ex: # This should never happen.
  439. # Let's try to close the `ctx` anyways.
  440. ctx.close()
  441. self._fatal_error(ex, True)
  442. self._buffer.clear()
  443. self._buffer_size = 0
  444. return
  445. elif err != uv.UV_EAGAIN:
  446. ctx.close()
  447. exc = convert_error(err)
  448. self._fatal_error(exc, True)
  449. self._buffer.clear()
  450. self._buffer_size = 0
  451. return
  452. # fall through
  453. if ctx is None:
  454. ctx = _StreamWriteContext.new(self, self._buffer)
  455. err = uv.uv_write(&ctx.req,
  456. <uv.uv_stream_t*>self._handle,
  457. ctx.uv_bufs_start,
  458. ctx.uv_bufs_len,
  459. __uv_stream_on_write)
  460. self._buffer_size = 0
  461. # Can't use `_buffer.clear()` here: `ctx` holds a reference to
  462. # the `_buffer`.
  463. self._buffer = []
  464. if err < 0:
  465. # close write context
  466. ctx.close()
  467. exc = convert_error(err)
  468. self._fatal_error(exc, True)
  469. return
  470. self._maybe_resume_protocol()
  471. cdef size_t _get_write_buffer_size(self):
  472. if self._handle is NULL:
  473. return 0
  474. return ((<uv.uv_stream_t*>self._handle).write_queue_size +
  475. self._buffer_size)
  476. cdef _close(self):
  477. try:
  478. if self._read_pybuf_acquired:
  479. # Should never happen. libuv always calls uv_alloc/uv_read
  480. # in pairs.
  481. self._loop.call_exception_handler({
  482. 'transport': self,
  483. 'message': 'XXX: an allocated buffer in transport._close()'
  484. })
  485. self._read_pybuf_acquired = 0
  486. PyBuffer_Release(&self._read_pybuf)
  487. self._stop_reading()
  488. finally:
  489. UVSocketHandle._close(<UVHandle>self)
  490. cdef inline _on_accept(self):
  491. # Ultimately called by __uv_stream_on_listen.
  492. self._init_protocol()
  493. cdef inline _on_eof(self):
  494. # Any exception raised here will be caught in
  495. # __uv_stream_on_read.
  496. try:
  497. meth = self._protocol.eof_received
  498. except AttributeError:
  499. keep_open = False
  500. else:
  501. keep_open = run_in_context(self.context, meth)
  502. if keep_open:
  503. # We're keeping the connection open so the
  504. # protocol can write more, but we still can't
  505. # receive more, so remove the reader callback.
  506. self._stop_reading()
  507. else:
  508. self.close()
  509. cdef inline _on_write(self):
  510. self._maybe_resume_protocol()
  511. if not self._get_write_buffer_size():
  512. if self._closing:
  513. self._schedule_call_connection_lost(None)
  514. elif self._eof:
  515. self._shutdown()
  516. cdef inline _init(self, Loop loop, object protocol, Server server,
  517. object waiter, object context):
  518. self.context = context
  519. self._set_protocol(protocol)
  520. self._start_init(loop)
  521. if server is not None:
  522. self._set_server(server)
  523. if waiter is not None:
  524. self._set_waiter(waiter)
  525. cdef inline _on_connect(self, object exc):
  526. # Called from __tcp_connect_callback (tcp.pyx) and
  527. # __pipe_connect_callback (pipe.pyx).
  528. if exc is None:
  529. self._init_protocol()
  530. else:
  531. if self._waiter is None:
  532. self._fatal_error(exc, False, "connect failed")
  533. elif self._waiter.cancelled():
  534. # Connect call was cancelled; just close the transport
  535. # silently.
  536. self._close()
  537. elif self._waiter.done():
  538. self._fatal_error(exc, False, "connect failed")
  539. else:
  540. self._waiter.set_exception(exc)
  541. self._close()
  542. # === Public API ===
  543. def __repr__(self):
  544. return '<{} closed={} reading={} {:#x}>'.format(
  545. self.__class__.__name__,
  546. self._closed,
  547. self.__reading,
  548. id(self))
  549. def write(self, object buf):
  550. self._ensure_alive()
  551. if self._eof:
  552. raise RuntimeError('Cannot call write() after write_eof()')
  553. if not buf:
  554. return
  555. if self._conn_lost:
  556. self._conn_lost += 1
  557. return
  558. self._buffer_write(buf)
  559. self._initiate_write()
  560. def writelines(self, bufs):
  561. self._ensure_alive()
  562. if self._eof:
  563. raise RuntimeError('Cannot call writelines() after write_eof()')
  564. if self._conn_lost:
  565. self._conn_lost += 1
  566. return
  567. for buf in bufs:
  568. self._buffer_write(buf)
  569. self._initiate_write()
  570. def write_eof(self):
  571. self._ensure_alive()
  572. if self._eof:
  573. return
  574. self._eof = 1
  575. if not self._get_write_buffer_size():
  576. self._shutdown()
  577. def can_write_eof(self):
  578. return True
  579. def is_reading(self):
  580. return self._is_reading()
  581. def pause_reading(self):
  582. if self._closing or not self._is_reading():
  583. return
  584. self._stop_reading()
  585. def resume_reading(self):
  586. if self._is_reading() or self._closing:
  587. return
  588. self._start_reading()
  589. cdef void __uv_stream_on_shutdown(uv.uv_shutdown_t* req,
  590. int status) noexcept with gil:
  591. # callback for uv_shutdown
  592. if req.data is NULL:
  593. aio_logger.error(
  594. 'UVStream.shutdown callback called with NULL req.data, status=%r',
  595. status)
  596. return
  597. cdef UVStream stream = <UVStream> req.data
  598. if status < 0 and status != uv.UV_ECANCELED:
  599. # From libuv source code:
  600. # The ECANCELED error code is a lie, the shutdown(2) syscall is a
  601. # fait accompli at this point. Maybe we should revisit this in
  602. # v0.11. A possible reason for leaving it unchanged is that it
  603. # informs the callee that the handle has been destroyed.
  604. if UVLOOP_DEBUG:
  605. stream._loop._debug_stream_shutdown_errors_total += 1
  606. exc = convert_error(status)
  607. stream._fatal_error(
  608. exc, False, "error status in uv_stream_t.shutdown callback")
  609. return
  610. cdef inline bint __uv_stream_on_read_common(
  611. UVStream sc,
  612. Loop loop,
  613. ssize_t nread,
  614. ):
  615. if sc._closed:
  616. # The stream was closed, there is no reason to
  617. # do any work now.
  618. sc.__reading_stopped() # Just in case.
  619. return True
  620. if nread == uv.UV_EOF:
  621. # From libuv docs:
  622. # The callee is responsible for stopping closing the stream
  623. # when an error happens by calling uv_read_stop() or uv_close().
  624. # Trying to read from the stream again is undefined.
  625. try:
  626. if UVLOOP_DEBUG:
  627. loop._debug_stream_read_eof_total += 1
  628. sc._stop_reading()
  629. sc._on_eof()
  630. except BaseException as ex:
  631. if UVLOOP_DEBUG:
  632. loop._debug_stream_read_eof_cb_errors_total += 1
  633. sc._fatal_error(ex, False)
  634. finally:
  635. return True
  636. if nread == 0:
  637. # From libuv docs:
  638. # nread might be 0, which does not indicate an error or EOF.
  639. # This is equivalent to EAGAIN or EWOULDBLOCK under read(2).
  640. return True
  641. if nread < 0:
  642. # From libuv docs:
  643. # The callee is responsible for stopping closing the stream
  644. # when an error happens by calling uv_read_stop() or uv_close().
  645. # Trying to read from the stream again is undefined.
  646. #
  647. # Therefore, we're closing the stream. Since "UVHandle._close()"
  648. # doesn't raise exceptions unless uvloop is built with DEBUG=1,
  649. # we don't need try...finally here.
  650. if UVLOOP_DEBUG:
  651. loop._debug_stream_read_errors_total += 1
  652. if sc.__read_error_close:
  653. # Used for getting notified when a pipe is closed.
  654. # See WriteUnixTransport for the explanation.
  655. sc._on_eof()
  656. return True
  657. exc = convert_error(nread)
  658. sc._fatal_error(
  659. exc, False, "error status in uv_stream_t.read callback")
  660. return True
  661. return False
  662. cdef inline void __uv_stream_on_read_impl(
  663. uv.uv_stream_t* stream,
  664. ssize_t nread,
  665. const uv.uv_buf_t* buf,
  666. ):
  667. cdef:
  668. UVStream sc = <UVStream>stream.data
  669. Loop loop = sc._loop
  670. # It's OK to free the buffer early, since nothing will
  671. # be able to touch it until this method is done.
  672. __loop_free_buffer(loop)
  673. if __uv_stream_on_read_common(sc, loop, nread):
  674. return
  675. try:
  676. if UVLOOP_DEBUG:
  677. loop._debug_stream_read_cb_total += 1
  678. run_in_context1(
  679. sc.context,
  680. sc._protocol_data_received,
  681. loop._recv_buffer[:nread],
  682. )
  683. except BaseException as exc:
  684. if UVLOOP_DEBUG:
  685. loop._debug_stream_read_cb_errors_total += 1
  686. sc._fatal_error(exc, False)
  687. cdef inline void __uv_stream_on_write_impl(
  688. uv.uv_write_t* req,
  689. int status,
  690. ):
  691. cdef:
  692. _StreamWriteContext ctx = <_StreamWriteContext> req.data
  693. UVStream stream = <UVStream>ctx.stream
  694. ctx.close()
  695. if stream._closed:
  696. # The stream was closed, there is nothing to do.
  697. # Even if there is an error, like EPIPE, there
  698. # is no reason to report it.
  699. return
  700. if status < 0:
  701. if UVLOOP_DEBUG:
  702. stream._loop._debug_stream_write_errors_total += 1
  703. exc = convert_error(status)
  704. stream._fatal_error(
  705. exc, False, "error status in uv_stream_t.write callback")
  706. return
  707. try:
  708. stream._on_write()
  709. except BaseException as exc:
  710. if UVLOOP_DEBUG:
  711. stream._loop._debug_stream_write_cb_errors_total += 1
  712. stream._fatal_error(exc, False)
  713. cdef void __uv_stream_on_read(
  714. uv.uv_stream_t* stream,
  715. ssize_t nread,
  716. const uv.uv_buf_t* buf,
  717. ) noexcept with gil:
  718. if __ensure_handle_data(<uv.uv_handle_t*>stream,
  719. "UVStream read callback") == 0:
  720. return
  721. # Don't need try-finally, __uv_stream_on_read_impl is void
  722. __uv_stream_on_read_impl(stream, nread, buf)
  723. cdef void __uv_stream_on_write(
  724. uv.uv_write_t* req,
  725. int status,
  726. ) noexcept with gil:
  727. if UVLOOP_DEBUG:
  728. if req.data is NULL:
  729. aio_logger.error(
  730. 'UVStream.write callback called with NULL req.data, status=%r',
  731. status)
  732. return
  733. # Don't need try-finally, __uv_stream_on_write_impl is void
  734. __uv_stream_on_write_impl(req, status)
  735. cdef void __uv_stream_buffered_alloc(
  736. uv.uv_handle_t* stream,
  737. size_t suggested_size,
  738. uv.uv_buf_t* uvbuf,
  739. ) noexcept with gil:
  740. if __ensure_handle_data(<uv.uv_handle_t*>stream,
  741. "UVStream alloc buffer callback") == 0:
  742. return
  743. cdef:
  744. UVStream sc = <UVStream>stream.data
  745. Loop loop = sc._loop
  746. Py_buffer* pybuf = &sc._read_pybuf
  747. int got_buf = 0
  748. if sc._read_pybuf_acquired:
  749. uvbuf.len = 0
  750. uvbuf.base = NULL
  751. return
  752. sc._read_pybuf_acquired = 0
  753. try:
  754. buf = run_in_context1(
  755. sc.context,
  756. sc._protocol_get_buffer,
  757. suggested_size,
  758. )
  759. PyObject_GetBuffer(buf, pybuf, PyBUF_WRITABLE)
  760. got_buf = 1
  761. except BaseException as exc:
  762. # Can't call 'sc._fatal_error' or 'sc._close', libuv will SF.
  763. # We'll do it later in __uv_stream_buffered_on_read when we
  764. # receive UV_ENOBUFS.
  765. uvbuf.len = 0
  766. uvbuf.base = NULL
  767. return
  768. if not pybuf.len:
  769. uvbuf.len = 0
  770. uvbuf.base = NULL
  771. if got_buf:
  772. PyBuffer_Release(pybuf)
  773. return
  774. sc._read_pybuf_acquired = 1
  775. uvbuf.base = <char*>pybuf.buf
  776. uvbuf.len = pybuf.len
  777. cdef void __uv_stream_buffered_on_read(
  778. uv.uv_stream_t* stream,
  779. ssize_t nread,
  780. const uv.uv_buf_t* buf,
  781. ) noexcept with gil:
  782. if __ensure_handle_data(<uv.uv_handle_t*>stream,
  783. "UVStream buffered read callback") == 0:
  784. return
  785. cdef:
  786. UVStream sc = <UVStream>stream.data
  787. Loop loop = sc._loop
  788. Py_buffer* pybuf = &sc._read_pybuf
  789. if nread == uv.UV_ENOBUFS:
  790. sc._fatal_error(
  791. RuntimeError(
  792. 'unhandled error (or an empty buffer) in get_buffer()'),
  793. False)
  794. return
  795. try:
  796. if nread > 0 and not sc._read_pybuf_acquired:
  797. # From libuv docs:
  798. # nread is > 0 if there is data available or < 0 on error. When
  799. # we’ve reached EOF, nread will be set to UV_EOF. When
  800. # nread < 0, the buf parameter might not point to a valid
  801. # buffer; in that case buf.len and buf.base are both set to 0.
  802. raise RuntimeError(
  803. f'no python buffer is allocated in on_read; nread={nread}')
  804. if nread == 0:
  805. # From libuv docs:
  806. # nread might be 0, which does not indicate an error or EOF.
  807. # This is equivalent to EAGAIN or EWOULDBLOCK under read(2).
  808. return
  809. if __uv_stream_on_read_common(sc, loop, nread):
  810. return
  811. if UVLOOP_DEBUG:
  812. loop._debug_stream_read_cb_total += 1
  813. run_in_context1(sc.context, sc._protocol_buffer_updated, nread)
  814. except BaseException as exc:
  815. if UVLOOP_DEBUG:
  816. loop._debug_stream_read_cb_errors_total += 1
  817. sc._fatal_error(exc, False)
  818. finally:
  819. sc._read_pybuf_acquired = 0
  820. PyBuffer_Release(pybuf)