Repository: pytorch/pytorch
Pull request: Dataloader issues
1. Fixes `signal.signal` being called on child threads.

2. Sometimes error in dataloader methods causes the trace to be polluted as in #4507 . After investigating this with @ezyang , we realize the reason is that when an exception is thrown, 

    1. sometimes worker processes are terminated before the `Dataloader` is deleted (likely because that the trace contains a reference to the `Dataloader`)

    2. then the workers got SIGTERM from loader before the clean up methods are called in `Dataloader.__del__`

    3. then loader's SIGCHLD handler is triggered, and it sees that workers fail, so it reports a RuntimeError with polluted trace as shown in #4507 .

    As a solution, this PR suppresses SIGTERM from loader in worker SIGTERM handler, and `exit(0)` in such case.

3. Improves some dataloder tests so they use a subclass of `multiprocessing.Process` that tracks the last error, and checks the error message.

4. Prevents unnecessary cuda context initialization in `Dataloader` by `torch.cuda.current_device()` in case `pin_memory=False`.

5. Always call `_remove_worker_pids` in `Dataloader.shutdown` so the set on cpp side in `Dataloder.cpp` is always updated correctly even when `Dataloader.shutdown` errors.

Fixes #4507 .

Verified that windows tests pass.

Problem statement:
RuntimeError: DataLoader worker (pid 23616) is killed by signal: Terminated.
Here is the error info
```PYTHON
Traceback (most recent call last):
  File "train_multi_task.py", line 192, in <module>
    experiment(config)
  File "train_multi_task.py", line 82, in experiment
    data_list.append(iter_.next())
  File "/usr/local/lib/python2.7/dist-packages/torch/utils/data/dataloader.py", line 267, in __next__
    idx, batch = self._get_batch()
  File "/usr/local/lib/python2.7/dist-packages/torch/utils/data/dataloader.py", line 246, in _get_batch
    return self.data_queue.get()
  File "/usr/lib/python2.7/multiprocessing/queues.py", line 376, in get
    return recv()
  File "/usr/local/lib/python2.7/dist-packages/torch/multiprocessing/queue.py", line 22, in recv
    return pickle.loads(buf)
  File "/usr/lib/python2.7/pickle.py", line 1382, in loads
    return Unpickler(file).load()
  File "/usr/lib/python2.7/pickle.py", line 858, in load
    dispatch[key](self)
  File "/usr/lib/python2.7/pickle.py", line 1133, in load_reduce
    value = func(*args)
  File "/usr/local/lib/python2.7/dist-packages/torch/multiprocessing/reductions.py", line 68, in rebuild_storage_fd
    fd = multiprocessing.reduction.rebuild_handle(df)
  File "/usr/lib/python2.7/multiprocessing/reduction.py", line 157, in rebuild_handle
    new_handle = recv_handle(conn)
  File "/usr/lib/python2.7/multiprocessing/reduction.py", line 83, in recv_handle
    return _multiprocessing.recvfd(conn.fileno())
OSError: [Errno 4] Interrupted system call
Error in atexit._run_exitfuncs:
Traceback (most recent call last):
  File "/usr/lib/python2.7/atexit.py", line 24, in _run_exitfuncs
    func(*targs, **kargs)
  File "/usr/lib/python2.7/multiprocessing/util.py", line 325, in _exit_function
Error in sys.exitfunc:
Exception RuntimeError: RuntimeError('DataLoader worker (pid 23604) is killed by signal: Terminated.',) in <bound method DataLoaderIter.__del__ of <torch.utils.data.dataloader.DataLoaderIter object at 0x7f73ba2c8850>> ignored
Exception IOError: IOError(104, 'Connection reset by peer') in <bound method DataLoaderIter.__del__ of <torch.utils.data.dataloader.DataLoaderIter object at 0x7f73ba2c1890>> ignored
Traceback (most recent call last):
  File "/usr/lib/python2.7/atexit.py", line 30, in _run_exitfuncs
    traceback.print_exc()
  File "/usr/lib/python2.7/traceback.py", line 233, in print_exc
    print_exception(etype, value, tb, limit, file)
  File "/usr/lib/python2.7/traceback.py", line 125, in print_exception
    print_tb(tb, limit, file)
  File "/usr/lib/python2.7/traceback.py", line 69, in print_tb
    line = linecache.getline(filename, lineno, f.f_globals)
  File "/usr/lib/python2.7/linecache.py", line 14, in getline
    lines = getlines(filename, module_globals)
  File "/usr/lib/python2.7/linecache.py", line 40, in getlines
    return updatecache(filename, module_globals)
  File "/usr/lib/python2.7/linecache.py", line 133, in updatecache
    lines = fp.readlines()
  File "/usr/local/lib/python2.7/dist-packages/torch/utils/data/dataloader.py", line 172, in handler
    _error_if_any_worker_fails()
RuntimeError: DataLoader worker (pid 23616) is killed by signal: Terminated.
```
I'm lost...

Hints from issue discussion:
One solution might be moving your codes to Python 3.5+, where [syscalls failing with EINTR are retried automatically](https://www.python.org/dev/peps/pep-0475/).
+1
Installed the latest build from the source, now getting similar one:
`RuntimeError: DataLoader worker (pid 28124) is killed by signal: Illegal instruction.`
What’s your CPU?
@Jensen-Su Sorry about the tracing being polluted by the final RunTimeError. Your real issue is that EINTR interrupts `queue.get`. Python doesn't handle this until 3.5. I'll send out a PR to address these two things.

@lysuhin Your issue is unrelated to the above one. What did you do in your dataset's `__getitem__`? Also, as asked by @apaszke , what's your CPU?

Commit message: addressed @apaszke 's comments

Patch to review:
diff --git a/test/common.py b/test/common.py
index 1f8c27e023a0..9aa77af01356 100644
--- a/test/common.py
+++ b/test/common.py
@@ -378,6 +378,8 @@ def accept_output(update_type):
                 self.assertEqual(s, expected)
 
     if sys.version_info < (3, 2):
+        # assertRegexpMatches renamed assertRegex in 3.2
+        assertRegex = unittest.TestCase.assertRegexpMatches
         # assertRaisesRegexp renamed assertRaisesRegex in 3.2
         assertRaisesRegex = unittest.TestCase.assertRaisesRegexp
 
diff --git a/test/test_dataloader.py b/test/test_dataloader.py
index 07ac7ebe20cf..264acccc6586 100644
--- a/test/test_dataloader.py
+++ b/test/test_dataloader.py
@@ -1,19 +1,30 @@
 import math
 import sys
+import errno
 import os
 import ctypes
+import signal
 import torch
 import time
 import traceback
 import unittest
+import socket
 from torch import multiprocessing
 from torch.utils.data import Dataset, TensorDataset, DataLoader, ConcatDataset
 from torch.utils.data.dataset import random_split
 from torch.utils.data.dataloader import default_collate
+from torch.utils.data.utils import ExceptionWrapper, QueueWrapper
 from common import TestCase, run_tests, TEST_NUMPY, IS_WINDOWS
 from common_nn import TEST_CUDA
 
-JOIN_TIMEOUT = 14.0 if IS_WINDOWS else 1.5
+
+if sys.version_info[0] == 2:
+    import Queue as queue
+else:
+    import queue
+
+
+JOIN_TIMEOUT = 17.0 if IS_WINDOWS else 4.5
 
 
 class TestDatasetRandomSplit(TestCase):
@@ -103,6 +114,46 @@ def test_add_dataset(self):
         self.assertEqual(0, (d3[0][0] - result[14][0]).abs().sum())
 
 
+# Stores the first encountered exception in .exception.
+# Inspired by https://stackoverflow.com/a/33599967
+class ErrorTrackingProcess(multiprocessing.Process):
+
+    def __init__(self, *args, **kwargs):
+        super(ErrorTrackingProcess, self).__init__(*args, **kwargs)
+        self._pconn, self._cconn = multiprocessing.Pipe()
+        self._exception = None
+
+    def run(self):
+        # Disable stderr printing from os level, and make workers not printing
+        # to stderr.
+        # Can't use sys.stderr.close, otherwise Python `raise` will error with
+        # ValueError: I/O operation on closed file.
+        os.close(sys.stderr.fileno())
+        try:
+            super(ErrorTrackingProcess, self).run()
+            self._cconn.send(None)
+        except Exception as e:
+            self._cconn.send(ExceptionWrapper(sys.exc_info()))
+            raise
+
+    @property
+    def exception(self):
+        if self._pconn.poll():
+            self._exception = self._pconn.recv()
+        if self._exception is None:
+            return None
+        else:
+            return self._exception.exc_type(self._exception.exc_msg)
+
+    # ESRCH means that os.kill can't finds alive proc
+    def send_signal(self, signum, ignore_ESRCH=False):
+        try:
+            os.kill(self.pid, signum)
+        except OSError as e:
+            if not ignore_ESRCH or e.errno != errno.ESRCH:
+                raise
+
+
 class ErrorDataset(Dataset):
 
     def __init__(self, size):
@@ -175,21 +226,73 @@ def __len__(self):
 
 
 def _test_timeout():
-    os.close(sys.stderr.fileno())
-    sys.stderr.close()
     dataset = SleepDataset(10, 10)
     dataloader = DataLoader(dataset, batch_size=2, num_workers=2, timeout=1)
     _ = next(iter(dataloader))
 
 
 def _test_segfault():
-    os.close(sys.stderr.fileno())
-    sys.stderr.close()
     dataset = SegfaultDataset(10)
     dataloader = DataLoader(dataset, batch_size=2, num_workers=2)
     _ = next(iter(dataloader))
 
 
+def _test_interrupt_retry(timeout=0):
+    dataset = TensorDataset(torch.randn(1, 1), torch.randn(1, 1))
+    dataloader = DataLoader(dataset, batch_size=1, num_workers=1, timeout=timeout)
+    dataloaderiter = iter(dataloader)
+
+    # make SIGUSR1 interrupt
+    def handler(signum, frame):
+        pass
+    signal.signal(signal.SIGUSR1, handler)
+
+    # Replace iterator getter with a wrapper that reliably calls an
+    # interruptable blocking recv syscall to simulate interruption during recv
+    # in queue.get.
+    # The used socket.recv call below in the replacing function is quite
+    # dangerous because it blocks everything on Python side (likely by holding
+    # GIL), including any Python signal handlers and the cleaning up in
+    # dataloder.__del__ when this process exits. To prevent orphan worker child,
+    # we manually terminate worker process here.
+    # Conveniently, the worker has SIGTERM handler installed so SIGTERM from
+    # loader process won't cause loader error.
+
+    data_queue = dataloaderiter.data_queue
+    data = data_queue.get()  # ensure that worker handlers are installed
+    for w in dataloaderiter.workers:
+        w.terminate()
+
+    def interruptable_get(*args, **kwargs):
+        if dataloaderiter.shutdown:
+            return data
+        # get and config timeout if the argument is present
+        if timeout >= 0:
+            if 'timeout' in kwargs:
+                timeout_val = kwargs['timeout']
+            elif len(args) > 1:
+                timeout_val = args[1]  # first arg is `block`
+            else:
+                timeout_val = None
+            socket.setdefaulttimeout(timeout_val)
+        s = socket.socket(socket.AF_INET, type=socket.SOCK_DGRAM)
+        s.bind(("127.0.0.1", 0))
+        try:
+            return s.recv(1024)
+        except socket.timeout:
+            raise queue.Empty
+        finally:
+            s.close()
+
+    try:
+        if isinstance(data_queue, QueueWrapper):
+            data_queue.queue.get = interruptable_get
+    except Exception:
+        # in Python >= 3.5 or on Windows QueueWrapper is not a class but a no-op
+        pass
+    _ = next(dataloaderiter)
+
+
 # test custom init function
 def init_fn(worker_id):
     torch.manual_seed(12345)
@@ -272,22 +375,30 @@ def test_multiple_dataloaders(self):
         next(loader2_it)
 
     def test_segfault(self):
-        p = multiprocessing.Process(target=_test_segfault)
+        p = ErrorTrackingProcess(target=_test_segfault)
         p.start()
         p.join(JOIN_TIMEOUT)
         try:
             self.assertFalse(p.is_alive())
             self.assertNotEqual(p.exitcode, 0)
+            if IS_WINDOWS:
+                self.assertIsInstance(p.exception, OSError)
+                self.assertRegex(str(p.exception), r'access violation reading ')
+            else:
+                self.assertIsInstance(p.exception, RuntimeError)
+                self.assertRegex(str(p.exception), r'DataLoader worker \(pid \d+\) is killed by signal: ')
         finally:
             p.terminate()
 
     def test_timeout(self):
-        p = multiprocessing.Process(target=_test_timeout)
+        p = ErrorTrackingProcess(target=_test_timeout)
         p.start()
-        p.join(3.0 + JOIN_TIMEOUT)
+        p.join(JOIN_TIMEOUT)
         try:
             self.assertFalse(p.is_alive())
             self.assertNotEqual(p.exitcode, 0)
+            self.assertIsInstance(p.exception, RuntimeError)
+            self.assertRegex(str(p.exception), r'DataLoader timed out after \d+ seconds')
         finally:
             p.terminate()
 
@@ -308,6 +419,53 @@ def test_worker_init_fn(self):
             self.assertEqual(12345, batch[0])
             self.assertEqual(12345, batch[1])
 
+    @unittest.skipIf(IS_WINDOWS, "TODO: need to fix this test case for Windows")
+    def test_interrupt_retry(self):
+        # time.sleep doesn't seem to work on main process when running unittest
+        # for all tests together. Use Process.join as a sleep function.
+        p = ErrorTrackingProcess(target=_test_interrupt_retry)
+        p.start()
+        p.join(1)  # give it some time to start
+        try:
+            self.assertTrue(p.is_alive())
+            for i in range(3):
+                p.send_signal(signal.SIGUSR1)
+                p.join(0.5)
+                self.assertTrue(p.is_alive())
+        except OSError as e:
+            self.fail("DataLoader shouldn't fail due to interrupted syscall")
+        try:
+            self.assertTrue(p.is_alive())
+            self.assertIsNone(p.exception)
+        finally:
+            p.terminate()
+
+    @unittest.skipIf(IS_WINDOWS, "TODO: need to fix this test case for Windows")
+    def test_interrupt_retry_with_timeout(self):
+        # time.sleep doesn't seem to work on main process when running unittest
+        # for all tests together. Use Process.join as a sleep function.
+        timeout = 2
+        p = ErrorTrackingProcess(target=lambda: _test_interrupt_retry(timeout))
+        p.start()
+        p.join(1)  # give it some time to start
+        try:
+            for _ in range(5):
+                if p.is_alive():
+                    p.send_signal(signal.SIGUSR1)
+                    p.join(0.5)
+                else:
+                    break
+            p.join(JOIN_TIMEOUT)
+        except OSError as e:
+            self.fail("DataLoader shouldn't fail due to interrupted syscall")
+        try:
+            self.assertFalse(p.is_alive())
+            self.assertNotEqual(p.exitcode, 0)
+            self.assertIsInstance(p.exception, RuntimeError)
+            self.assertRegex(str(p.exception), r'DataLoader timed out after {} seconds'.format(timeout))
+        finally:
+            p.terminate()
+
     def test_shuffle(self):
         self._test_shuffle(DataLoader(self.dataset, shuffle=True))
 
diff --git a/torch/csrc/DataLoader.cpp b/torch/csrc/DataLoader.cpp
index f534ad262db8..9678c95300bf 100644
--- a/torch/csrc/DataLoader.cpp
+++ b/torch/csrc/DataLoader.cpp
@@ -43,7 +43,7 @@ static void HANDLER_NAME(int sig, siginfo_t *info, void *ctx)                 \
 
 // signal(2) is really not portable. So use sigaction.
 // http://man7.org/linux/man-pages/man2/signal.2.html
-static void setSignalHandler(int signal, void(*handler)(int, siginfo_t *, void *), struct sigaction *old_sa_ptr)
+static inline void setSignalHandler(int signal, void(*handler)(int, siginfo_t *, void *), struct sigaction *old_sa_ptr)
 {
   struct sigaction sa;
   sa.sa_sigaction = handler;
@@ -59,10 +59,34 @@ SIGNAL_HANDLER(SIGBUS, handler_SIGBUS, "ERROR: Unexpected bus error encountered
   "This might be caused by insufficient shared memory (shm).\n");
 SIGNAL_HANDLER(SIGSEGV, handler_SIGSEGV, "ERROR: Unexpected segmentation fault encountered in worker.\n");
 
+// When an error happend in DataLoader methods and Python starts to exit, the
+// error trace will keep the loader alive, and Python may kill the children
+// processes first before deleting the loader object. Then the cleaning up
+// methods in DataLoader.__del__ are not yet called, and SIGCHILD will print an
+// error saying a worker is killed by SIGTERM. So we suppress SIGTERM from main
+// loader process here to avoid this by _exit(EXIT_SUCCESS). Note that if we
+// exit with nonzero code, the loader SIGCHLD handler may report RuntimeError
+// again, and then it defeats the whole purpose.
+static void handler_SIGTERM(int sig, siginfo_t *info, void *ctx)
+{
+  if (info->si_pid == getppid()) {
+    _exit(EXIT_SUCCESS);
+  }
+  struct sigaction sa;
+  sa.sa_handler = SIG_DFL;
+  sa.sa_flags = 0;
+  if (sigemptyset(&sa.sa_mask) != 0 || sigaction(SIGTERM, &sa, NULL) != 0) {
+    _exit(EXIT_FAILURE);
+  } else {
+    raise(SIGTERM);
+  }
+}
+
 PyObject *THPModule_setWorkerSignalHandlers(PyObject *module, PyObject *arg) {
   HANDLE_TH_ERRORS
   setSignalHandler(SIGBUS, &handler_SIGBUS, NULL);
   setSignalHandler(SIGSEGV, &handler_SIGSEGV, NULL);
+  setSignalHandler(SIGTERM, &handler_SIGTERM, NULL);
   Py_RETURN_TRUE;
   END_HANDLE_TH_ERRORS
 }
@@ -73,33 +97,33 @@ PyObject *THPModule_errorIfAnyWorkerFails(PyObject *module) {
   HANDLE_TH_ERRORS
   int error;
   std::set<pid_t> *pid_set;
-  pid_t pid;
+  pid_t worker_pid;
   siginfo_t infop;
 
   // Only check the pids we care about
   for (auto it = worker_pids.begin(); it != worker_pids.end(); ++it) {
     pid_set = &(it->second);
     for (auto pid_it = pid_set->begin(); pid_it != pid_set->end(); ++pid_it) {
-      pid = *pid_it;
+      worker_pid = *pid_it;
       // Use waitid rather than waitpid so that we can set NOWAIT, and that Python
       // and other handlers can get whatever info they want about the child.
       infop.si_pid = 0;
-      error = waitid(P_PID, pid, &infop, WEXITED|WNOHANG|WNOWAIT);
+      error = waitid(P_PID, worker_pid, &infop, WEXITED|WNOHANG|WNOWAIT);
       // ignore errors and case with no waitable child
       if (error < 0 || infop.si_pid == 0)
         continue;
-      if (infop.si_code == CLD_EXITED && infop.si_status != 0) {  // exit with error
+      if (infop.si_code == CLD_EXITED && infop.si_status != EXIT_SUCCESS) {  // exit with error
         std::ostringstream oss;
-        oss << "DataLoader worker (pid " << pid << ") exited unexpectedly "
-            << "with exit code " << infop.si_status << ".";
+        oss << "DataLoader worker (pid " << worker_pid << ") exited "
+            << "unexpectedly with exit code " << infop.si_status << ".";
         // This is necessary. Otherwise, the runtime error will kill the other
         // workers, and trigger this again.
         pid_set->clear();
         throw std::runtime_error(oss.str());
       }  else if (infop.si_code == CLD_KILLED || infop.si_code == CLD_DUMPED) {  // killed by signal
         std::ostringstream oss;
-        oss << "DataLoader worker (pid " << pid << ") is killed by signal: "
-            << strsignal(infop.si_status) << ".";
+        oss << "DataLoader worker (pid " << worker_pid << ") is killed "
+            << "by signal: " << strsignal(infop.si_status) << ".";
         // This is necessary. Otherwise, the runtime error will kill the other
         // workers, and trigger this again.
         pid_set->clear();
diff --git a/torch/utils/data/dataloader.py b/torch/utils/data/dataloader.py
index b72d4d9abf43..49f949cd10ff 100644
--- a/torch/utils/data/dataloader.py
+++ b/torch/utils/data/dataloader.py
@@ -3,12 +3,12 @@
 from torch._C import _set_worker_signal_handlers, _update_worker_pids, \
     _remove_worker_pids, _error_if_any_worker_fails
 from .sampler import SequentialSampler, RandomSampler, BatchSampler
+from .utils import ExceptionWrapper, QueueWrapper
 import signal
 import functools
 import collections
 import re
 import sys
-import traceback
 import threading
 from torch._six import string_classes, int_classes
 
@@ -23,14 +23,6 @@
 """Whether to use shared memory in default_collate"""
 
 
-class ExceptionWrapper(object):
-    "Wraps an exception plus traceback to communicate across threads"
-
-    def __init__(self, exc_info):
-        self.exc_type = exc_info[0]
-        self.exc_msg = "".join(traceback.format_exception(*exc_info))
-
-
 def _worker_loop(dataset, index_queue, data_queue, collate_fn, seed, init_fn, worker_id):
     global _use_shared_memory
     _use_shared_memory = True
@@ -193,8 +185,8 @@ def __init__(self, loader):
 
         if self.num_workers > 0:
             self.worker_init_fn = loader.worker_init_fn
-            self.index_queue = multiprocessing.SimpleQueue()
-            self.worker_result_queue = multiprocessing.SimpleQueue()
+            self.index_queue = QueueWrapper(multiprocessing.SimpleQueue())
+            self.worker_result_queue = QueueWrapper(multiprocessing.SimpleQueue())
             self.batches_outstanding = 0
             self.worker_pids_set = False
             self.shutdown = False
@@ -211,11 +203,16 @@ def __init__(self, loader):
                 for i in range(self.num_workers)]
 
             if self.pin_memory or self.timeout > 0:
-                self.data_queue = queue.Queue()
+                self.data_queue = QueueWrapper(queue.Queue())
+                if self.pin_memory:
+                    maybe_device_id = torch.cuda.current_device()
+                else:
+                    # do not initialize cuda context if not necessary
+                    maybe_device_id = None
                 self.worker_manager_thread = threading.Thread(
                     target=_worker_manager_loop,
                     args=(self.worker_result_queue, self.data_queue, self.done_event, self.pin_memory,
-                          torch.cuda.current_device()))
+                          maybe_device_id))
                 self.worker_manager_thread.daemon = True
                 self.worker_manager_thread.start()
             else:
@@ -239,7 +236,7 @@ def __len__(self):
     def _get_batch(self):
         if self.timeout > 0:
             try:
-                return self.data_queue.get(True, self.timeout)
+                return self.data_queue.get(timeout=self.timeout)
             except queue.Empty:
                 raise RuntimeError('DataLoader timed out after {} seconds'.format(self.timeout))
         else:
@@ -302,17 +299,20 @@ def __getstate__(self):
         raise NotImplementedError("DataLoaderIterator cannot be pickled")
 
     def _shutdown_workers(self):
-        if not self.shutdown:
-            self.shutdown = True
-            self.done_event.set()
-            # if worker_manager_thread is waiting to put
-            while not self.data_queue.empty():
-                self.data_queue.get()
-            for _ in self.workers:
-                self.index_queue.put(None)
-            # done_event should be sufficient to exit worker_manager_thread, but
-            # be safe here and put another None
-            self.worker_result_queue.put(None)
+        try:
+            if not self.shutdown:
+                self.shutdown = True
+                self.done_event.set()
+                # if worker_manager_thread is waiting to put
+                while not self.data_queue.empty():
+                    self.data_queue.get()
+                for _ in self.workers:
+                    self.index_queue.put(None)
+                # done_event should be sufficient to exit worker_manager_thread,
+                # but be safe here and put another None
+                self.worker_result_queue.put(None)
+        finally:
+            # removes pids no matter what
             if self.worker_pids_set:
                 _remove_worker_pids(id(self))
                 self.worker_pids_set = False
diff --git a/torch/utils/data/utils.py b/torch/utils/data/utils.py
new file mode 100644
index 000000000000..ee66a9e0bd84
--- /dev/null
+++ b/torch/utils/data/utils.py
@@ -0,0 +1,71 @@
+import sys
+import traceback
+
+if sys.version_info[0] == 2:
+    import Queue as queue
+else:
+    import queue
+
+
+class ExceptionWrapper(object):
+    r"Wraps an exception plus traceback to communicate across threads"
+
+    def __init__(self, exc_info):
+        self.exc_type = exc_info[0]
+        self.exc_msg = "".join(traceback.format_exception(*exc_info))
+
+
+# Queue get and put methods may fail if the underlying syscalls are interrupted
+# (EINTR). So we need to wrap them to retry automatically if needed.
+#
+# Syscalls are automatically retried upon encountering EINTR since Python 3.5
+# https://www.python.org/dev/peps/pep-0475/
+# EINTR is not available on Windows.
+if sys.platform == 'win32' or sys.version_info >= (3, 5):
+    def QueueWrapper(queue):
+        return queue
+else:
+    import time
+    import errno
+
+    if sys.version_info >= (3, 3):
+        time_fn = time.perf_counter
+    else:
+        time_fn = time.time
+
+    class QueueWrapper(object):
+        r"""Wraps a queue object that conforms to following interface:
+
+            .get()  or  .get(timeout=timeout)
+            .put(value)
+        """
+
+        def __init__(self, queue):
+            self.queue = queue
+
+        def get(self, timeout=None):
+            while True:
+                try:
+                    if timeout is None:
+                        return self.queue.get()
+                    else:
+                        t0 = time_fn()
+                        return self.queue.get(timeout=timeout)
+                except IOError as e:
+                    if e.errno != errno.EINTR:
+                        raise
+                    if timeout is not None:
+                        timeout -= time_fn() - t0
+                        if timeout <= 0:
+                            raise queue.Empty
+
+        def put(self, val):
+            while True:
+                try:
+                    return self.queue.put(val)
+                except IOError as e:
+                    if e.errno != errno.EINTR:
+                        raise
+
+        def __getattr__(self, attr):
+            return getattr(self.queue, attr)


Code review comments:
diff_hunk: @@ -0,0 +1,71 @@
+import sys
+import traceback
+
+if sys.version_info[0] == 2:
+    import Queue as queue
+else:
+    import queue
+
+
+class ExceptionWrapper(object):
+    r"Wraps an exception plus traceback to communicate across threads"
+
+    def __init__(self, exc_info):
+        self.exc_type = exc_info[0]
+        self.exc_msg = "".join(traceback.format_exception(*exc_info))
+
+
+# Queue get and put methods may fail if the underlying syscalls are interrupted
+# (EINTR). So we need to wrap them to retry automatically if needed.
+#
+# Syscalls are automatically retried upon encountering EINTR since Python 3.5
+# https://www.python.org/dev/peps/pep-0475/
+# EINTR is not available on Windows.
+if sys.platform == 'win32' or sys.version_info >= (3, 5):
+    def QueueWrapper(queue):
+        return queue
+else:
+    import time
+    import errno
+
+    if sys.version_info >= (3, 3):
+        time_fn = time.perf_counter
+    else:
+        time_fn = time.time
+
+    class QueueWrapper(object):
+        r"""Wraps a queue object that conforms to following interface:
+
+            .get()  or  .get(timeout=timeout)
+            .put(value)
+        """
+
+        def __init__(self, queue):
+            self.queue = queue
+
+        def get(self, timeout=None):
+            while True:
+                try:
+                    if timeout is None:
+                        return self.queue.get()
+                    else:
+                        t0 = time_fn()
+                        return self.queue.get(timeout=timeout)
+                except IOError as e:
+                    if e.errno != errno.EINTR:
+                        raise
+                    if timeout is not None:
+                        timeout -= time_fn() - t0
+                        if timeout <= 0:
+                            raise queue.Empty
+
+        def put(self, val):
+            while True:
+                try:
+                    return self.queue.put(val)
path: torch/utils/data/utils.py
text: @user1:
One thing the comes mind is what happens if EINTR comes in the middle of writing to a queue? This will mean that the data has been corrupted and we can't really do anything about it, right? Similarly with `get()` (what if we get interrupted in the middle of reading!)

@author:
According to manual, `recv` can get EINTR when it is interrupted before reading any data, and `send` can get EINTR before any data is transmitted.

@user1:
Aren't queues implemented around pipes? If so they are using `read` and `write`. What I'm concerned about is that if I were writing such a queue I would do it like this:
```cpp
std::size_t remaining_bytes = ...;
while (remaining_bytes > 0) {
  ssize_t read_bytes = read(fd, &buffer, remaining_bytes);
  if (read_bytes == -1) throw some_error();
  remaining_bytes -= read_bytes;
  buffer += read_bytes;
}
```
This is because `read` is not guaranteed to return you exactly `remaining_bytes`, but *at most* `remaining_bytes` so you might need to loop. Now, if Python implements read in this way, but breaks out of the loop if it sees EINTR, then by the time we call `get()` again, some of the data could have still been left in the pipe, and it will be corrupted.

@author:
You are right. I checked python implementation. Queues are implemented using`read` and `write`, and this is indeed problematic. In this case, I can't think of a clean solution currently. Do you have any ideas?

@user1:
Well if EINTR makes the data inside the queue corrupted, then I don't think we have a better choice than crashing 😕 

@author:
Yeah, I'll add a comment about it then.

@user1:
Well I'm just not sure if we still want to merge this patch, since it's likely to crash on the next `get()` anyway

@author:
Hmm yeah. I'll remove the EINTR part then :) Other parts of the patch should still be reasonable.

There is another issue that I can add a quick fix in that patch. Some people have been using dataloder in a separate thread. Then because the first `DataLoaderIter` object is init'd in a child thread, setting SIGCHLD handler will error at `signal.signal` complaining that it should only be used in main thread. There are two solutions: 1) setting handler at import time or 2) try catch the `signal.signal` (or test if it is main thread using `isinstance(threading.current_thread(), threading._MainThread)`). I am inclined towards the second because the first will still fail if import is done in the multithreading function. Does it sound reasonable to you?

diff_hunk: @@ -175,21 +226,73 @@ def __len__(self):
 
 
 def _test_timeout():
-    os.close(sys.stderr.fileno())
-    sys.stderr.close()
     dataset = SleepDataset(10, 10)
     dataloader = DataLoader(dataset, batch_size=2, num_workers=2, timeout=1)
     _ = next(iter(dataloader))
 
 
 def _test_segfault():
-    os.close(sys.stderr.fileno())
-    sys.stderr.close()
     dataset = SegfaultDataset(10)
     dataloader = DataLoader(dataset, batch_size=2, num_workers=2)
     _ = next(iter(dataloader))
 
 
+def _test_interrupt_retry(timeout=0):
+    dataset = TensorDataset(torch.randn(1, 1), torch.randn(1, 1))
+    dataloader = DataLoader(dataset, batch_size=1, num_workers=1, timeout=timeout)
+    dataloaderiter = iter(dataloader)
+
+    # make SIGUSR1 interrupt
+    def handler(signum, frame):
+        pass
+    signal.signal(signal.SIGUSR1, handler)
+
+    # Replace iterator getter with a wrapper that reliably calls an
+    # interruptable blocking recv syscall to simulate interruption during recv
+    # in queue.get.
+    # The used socket.recv call below in the replacing function is quite
+    # dangerous because it blocks everything on Python side (likely by holding
+    # GIL), including any Python signal handlers and the cleaning up in
+    # dataloder.__del__ when this process exits. To prevent orphan worker child,
+    # we manually terminate worker process here.
+    # Conveniently, the worker has SIGTERM handler installed so SIGTERM from
+    # loader process won't cause loader error.
+
+    data_queue = dataloaderiter.data_queue
+    data = data_queue.get()  # ensure that worker handlers are installed
+    for w in dataloaderiter.workers:
+        w.terminate()
+
+    def interruptable_get(*args, **kwargs):
+        if dataloaderiter.shutdown:
+            return data
+        # get and config timeout if the argument is present
+        if timeout >= 0:
+            if 'timeout' in kwargs:
+                timeout_val = kwargs['timeout']
+            elif len(args) > 1:
+                timeout_val = args[1]  # first arg is `block`
+            else:
+                timeout_val = None
+            socket.setdefaulttimeout(timeout_val)
+        s = socket.socket(socket.AF_INET, type=socket.SOCK_DGRAM)
+        s.bind(("127.0.0.1", 0))
+        try:
+            return s.recv(1024)
+        except socket.timeout:
+            raise queue.Empty
+        finally:
+            s.close()
+
+    try:
+        if isinstance(data_queue, QueueWrapper):
+            data_queue.queue.get = interruptable_get
+    except Exception:
+        # in Python >= 3.5 or on Windows QueueWrapper is not a class but a no-op
+        pass
path: test/test_dataloader.py
text: @user1:
It's not really an exception, can you just use an `if` with `sys.version_info`? Also, why do we need to replace `get` with our custom implementation? I thought it is interruptible in its default implementation and we're actually trying to see if we can improve this

@author:
The default get is interruptible, but only during its `recv` call. And it doesn't call `recv` unless the queue has data. So in reality there is only a tiny timeframe where it is interruptible, so it is really not testable. The custom implementation can be interrupted at any time.

@user1:
Makes sense. Can you still change the code to use an `if` instead of raising exceptions?

@user1:
Also, why can't we patch `get` in Python 3.5? Should we just skip this test in this version, since it doesn't suffer from the same problems

@author:
`get` is already patched by Python in 3.5. But yeah, I'll add an `if` here.

diff_hunk: @@ -308,6 +419,53 @@ def test_worker_init_fn(self):
             self.assertEqual(12345, batch[0])
             self.assertEqual(12345, batch[1])
 
+    @unittest.skipIf(IS_WINDOWS, "TODO: need to fix this test case for Windows")
+    def test_interrupt_retry(self):
+        # time.sleep doesn't seem to work on main process when running unittest
+        # for all tests together. Use Process.join as a sleep function.
+        p = ErrorTrackingProcess(target=_test_interrupt_retry)
+        p.start()
+        p.join(1)  # give it some time to start
+        try:
+            self.assertTrue(p.is_alive())
+            for i in range(3):
+                p.send_signal(signal.SIGUSR1)
+                p.join(0.5)
+                self.assertTrue(p.is_alive())
+        except OSError as e:
+            self.fail("DataLoader shouldn't fail due to interrupted syscall")
path: test/test_dataloader.py
text: @user1:
Will this ever throw an `OSError`? That error could have been thrown in the subprocess, but it will be visible as `p.exception` in here, right? Same applies to the test below

@author:
signal.signal can throw OSError when the signaled process has already terminated.

@user1:
Can you check that it's not a random `OSError`, but one that says that the process has died. This is both more robust, and will be a sufficient form of documentation for the code

@author:
Will do!

diff_hunk: @@ -0,0 +1,71 @@
+import sys
+import traceback
+
+if sys.version_info[0] == 2:
+    import Queue as queue
+else:
+    import queue
+
+
+class ExceptionWrapper(object):
+    r"Wraps an exception plus traceback to communicate across threads"
+
+    def __init__(self, exc_info):
+        self.exc_type = exc_info[0]
+        self.exc_msg = "".join(traceback.format_exception(*exc_info))
+
+
+# Queue get and put methods may fail if the underlying syscalls are interrupted
+# (EINTR). So we need to wrap them to retry automatically if needed.
+#
+# Syscalls are automatically retried upon encountering EINTR since Python 3.5
+# https://www.python.org/dev/peps/pep-0475/
+# EINTR is not available on Windows.
+if sys.platform == 'win32' or sys.version_info >= (3, 5):
+    def QueueWrapper(queue):
+        return queue
+else:
+    import time
+    import errno
+
+    if sys.version_info >= (3, 3):
+        time_fn = time.perf_counter
+    else:
+        time_fn = time.time
+
+    class QueueWrapper(object):
+        r"""Wraps a queue object that conforms to following interface:
+
+            .get()  or  .get(timeout=timeout)
+            .put(value)
+        """
+
+        def __init__(self, queue):
+            self.queue = queue
+
+        def get(self, timeout=None):
+            while True:
+                try:
+                    if timeout is None:
+                        return self.queue.get()
+                    else:
+                        t0 = time_fn()
+                        return self.queue.get(timeout=timeout)
+                except IOError as e:
+                    if e.errno != errno.EINTR:
+                        raise
+                    if timeout is not None:
+                        timeout -= time_fn() - t0
+                        if timeout <= 0:
+                            raise queue.Empty
path: torch/utils/data/utils.py
text: @user1:
I think it should be `raise queue.Empty()` to raise an instance instead of a class