|
7 | 7 | import atexit |
8 | 8 | import contextvars |
9 | 9 | import io |
| 10 | +import logging |
10 | 11 | import os |
11 | 12 | import sys |
12 | 13 | import threading |
|
33 | 34 |
|
34 | 35 | PIPE_BUFFER_SIZE = 1000 |
35 | 36 |
|
| 37 | +logger = logging.getLogger(__name__) |
| 38 | + |
36 | 39 | # ----------------------------------------------------------------------------- |
37 | 40 | # IO classes |
38 | 41 | # ----------------------------------------------------------------------------- |
@@ -355,8 +358,20 @@ def _really_send(self, msg, *args, **kwargs): |
355 | 358 | mp_mode = self._check_mp_mode() |
356 | 359 |
|
357 | 360 | if mp_mode != CHILD: |
358 | | - # we are master, do a regular send |
359 | | - self.socket.send_multipart(msg, *args, **kwargs) |
| 361 | + # we are master, do a regular send. |
| 362 | + # The closed check above is racy: once the IO thread has been |
| 363 | + # joined, _really_send runs on the caller's thread while close() |
| 364 | + # may run concurrently on another, so the socket can be closed |
| 365 | + # (or become None) between the check and the send. Swallow that |
| 366 | + # specific case rather than logging a noisy traceback during |
| 367 | + # shutdown. |
| 368 | + try: |
| 369 | + self.socket.send_multipart(msg, *args, **kwargs) |
| 370 | + except (AttributeError, zmq.error.ZMQError) as e: |
| 371 | + if isinstance(e, AttributeError) or e.errno == zmq.ENOTSOCK: |
| 372 | + logger.debug("IOPub socket closed during send (likely shutdown): %s", e) |
| 373 | + return |
| 374 | + raise |
360 | 375 | else: |
361 | 376 | # we are a child, pipe to master |
362 | 377 | # new context/socket for every pipe-out |
|
0 commit comments