diff --git a/Lib/multiprocessing/connection.py b/Lib/multiprocessing/connection.py
index 40f7a9e594b..55f6ba99f24 100644
--- a/Lib/multiprocessing/connection.py
+++ b/Lib/multiprocessing/connection.py
@@ -363,7 +363,10 @@ class Connection(_ConnectionBase):
     def _send(self, buf, write=_write):
         remaining = len(buf)
         while True:
-            n = write(self._handle, buf)
+            try:
+                n = write(self._handle, buf)
+            except InterruptedError:
+                continue
             remaining -= n
             if remaining == 0:
                 break
@@ -374,7 +377,10 @@ class Connection(_ConnectionBase):
         handle = self._handle
         remaining = size
         while remaining > 0:
-            chunk = read(handle, remaining)
+            try:
+                chunk = read(handle, remaining)
+            except InterruptedError:
+                continue
             n = len(chunk)
             if n == 0:
                 if remaining == size:
@@ -578,7 +584,13 @@ class SocketListener(object):
             self._unlink = None
 
     def accept(self):
-        s, self._last_accepted = self._socket.accept()
+        while True:
+            try:
+                s, self._last_accepted = self._socket.accept()
+            except InterruptedError:
+                pass
+            else:
+                break
         s.setblocking(True)
         return Connection(s.detach())
 
diff --git a/Lib/test/test_multiprocessing.py b/Lib/test/test_multiprocessing.py
index 3be72c4ad1a..23556fea814 100644
--- a/Lib/test/test_multiprocessing.py
+++ b/Lib/test/test_multiprocessing.py
@@ -3513,6 +3513,74 @@ class TestForkAwareThreadLock(unittest.TestCase):
         p.join()
         self.assertLessEqual(new_size, old_size)
 
+#
+# Issue #17097: EINTR should be ignored by recv(), send(), accept() etc
+#
+
+class TestIgnoreEINTR(unittest.TestCase):
+
+    @classmethod
+    def _test_ignore(cls, conn):
+        def handler(signum, frame):
+            pass
+        signal.signal(signal.SIGUSR1, handler)
+        conn.send('ready')
+        x = conn.recv()
+        conn.send(x)
+        conn.send_bytes(b'x'*(1024*1024))   # sending 1 MB should block
+
+    @unittest.skipUnless(hasattr(signal, 'SIGUSR1'), 'requires SIGUSR1')
+    def test_ignore(self):
+        conn, child_conn = multiprocessing.Pipe()
+        try:
+            p = multiprocessing.Process(target=self._test_ignore,
+                                        args=(child_conn,))
+            p.daemon = True
+            p.start()
+            child_conn.close()
+            self.assertEqual(conn.recv(), 'ready')
+            time.sleep(0.1)
+            os.kill(p.pid, signal.SIGUSR1)
+            time.sleep(0.1)
+            conn.send(1234)
+            self.assertEqual(conn.recv(), 1234)
+            time.sleep(0.1)
+            os.kill(p.pid, signal.SIGUSR1)
+            self.assertEqual(conn.recv_bytes(), b'x'*(1024*1024))
+            time.sleep(0.1)
+            p.join()
+        finally:
+            conn.close()
+
+    @classmethod
+    def _test_ignore_listener(cls, conn):
+        def handler(signum, frame):
+            pass
+        signal.signal(signal.SIGUSR1, handler)
+        l = multiprocessing.connection.Listener()
+        conn.send(l.address)
+        a = l.accept()
+        a.send('welcome')
+
+    @unittest.skipUnless(hasattr(signal, 'SIGUSR1'), 'requires SIGUSR1')
+    def test_ignore_listener(self):
+        conn, child_conn = multiprocessing.Pipe()
+        try:
+            p = multiprocessing.Process(target=self._test_ignore_listener,
+                                        args=(child_conn,))
+            p.daemon = True
+            p.start()
+            child_conn.close()
+            address = conn.recv()
+            time.sleep(0.1)
+            os.kill(p.pid, signal.SIGUSR1)
+            time.sleep(0.1)
+            client = multiprocessing.connection.Client(address)
+            self.assertEqual(client.recv(), 'welcome')
+            p.join()
+        finally:
+            conn.close()
+
 #
 #
 #
@@ -3520,7 +3588,7 @@ class TestForkAwareThreadLock(unittest.TestCase):
 testcases_other = [OtherTest, TestInvalidHandle, TestInitializers,
                    TestStdinBadfiledescriptor, TestWait, TestInvalidFamily,
                    TestFlags, TestTimeouts, TestNoForkBomb,
-                   TestForkAwareThreadLock]
+                   TestForkAwareThreadLock, TestIgnoreEINTR]
 
 #
 #
diff --git a/Misc/NEWS b/Misc/NEWS
index ed51b38f1fc..6f0b54e7eed 100644
--- a/Misc/NEWS
+++ b/Misc/NEWS
@@ -135,6 +135,8 @@ Core and Builtins
 Library
 -------
 
+- Issue #17097: Make multiprocessing ignore EINTR.
+
 - Issue #18339: Negative ints keys in unpickler.memo dict no longer cause a
   segfault inside the _pickle C extension.