diff --git a/Lib/test/script_helper.py b/Lib/test/script_helper.py
index 8c599d1b04b..b31fc40013e 100644
--- a/Lib/test/script_helper.py
+++ b/Lib/test/script_helper.py
@@ -1,6 +1,7 @@
 # Common utility functions used by various script execution tests
 #  e.g. test_cmd_line, test_cmd_line_script and test_runpy
 
+import collections
 import importlib
 import sys
 import os
@@ -50,8 +51,12 @@ def _interpreter_requires_environment():
     return __cached_interp_requires_environment
 
 
+_PythonRunResult = collections.namedtuple("_PythonRunResult",
+                                          ("rc", "out", "err"))
+
+
 # Executing the interpreter in a subprocess
-def _assert_python(expected_success, *args, **env_vars):
+def run_python_until_end(*args, **env_vars):
     env_required = _interpreter_requires_environment()
     if '__isolated' in env_vars:
         isolated = env_vars.pop('__isolated')
@@ -85,12 +90,16 @@ def _assert_python(expected_success, *args, **env_vars):
         p.stderr.close()
     rc = p.returncode
     err = strip_python_stderr(err)
-    if (rc and expected_success) or (not rc and not expected_success):
+    return _PythonRunResult(rc, out, err), cmd_line
+
+def _assert_python(expected_success, *args, **env_vars):
+    res, cmd_line = run_python_until_end(*args, **env_vars)
+    if (res.rc and expected_success) or (not res.rc and not expected_success):
         raise AssertionError(
             "Process return code is %d, command line was: %r, "
-            "stderr follows:\n%s" % (rc, cmd_line,
-                                     err.decode('ascii', 'ignore')))
-    return rc, out, err
+            "stderr follows:\n%s" % (res.rc, cmd_line,
+                                     res.err.decode('ascii', 'ignore')))
+    return res
 
 def assert_python_ok(*args, **env_vars):
     """
diff --git a/Lib/test/test_io.py b/Lib/test/test_io.py
index 95277d9c476..dfa3d771f42 100644
--- a/Lib/test/test_io.py
+++ b/Lib/test/test_io.py
@@ -35,7 +35,7 @@ import weakref
 from collections import deque, UserList
 from itertools import cycle, count
 from test import support
-from test.script_helper import assert_python_ok
+from test.script_helper import assert_python_ok, run_python_until_end
 
 import codecs
 import io  # C implementation of io
@@ -3350,6 +3350,49 @@ class CMiscIOTest(MiscIOTest):
         b = bytearray(2)
         self.assertRaises(ValueError, bufio.readinto, b)
 
+    @unittest.skipUnless(threading, 'Threading required for this test.')
+    def check_daemon_threads_shutdown_deadlock(self, stream_name):
+        # Issue #23309: deadlocks at shutdown should be avoided when a
+        # daemon thread and the main thread both write to a file.
+        code = """if 1:
+            import sys
+            import time
+            import threading
+
+            file = sys.{stream_name}
+
+            def run():
+                while True:
+                    file.write('.')
+                    file.flush()
+
+            thread = threading.Thread(target=run)
+            thread.daemon = True
+            thread.start()
+
+            time.sleep(0.5)
+            file.write('!')
+            file.flush()
+            """.format_map(locals())
+        res, _ = run_python_until_end("-c", code)
+        err = res.err.decode()
+        if res.rc != 0:
+            # Failure: should be a fatal error
+            self.assertIn("Fatal Python error: could not acquire lock "
+                          "for <_io.BufferedWriter name='<{stream_name}>'> "
+                          "at interpreter shutdown, possibly due to "
+                          "daemon threads".format_map(locals()),
+                          err)
+        else:
+            self.assertFalse(err.strip('.!'))
+
+    def test_daemon_threads_shutdown_stdout_deadlock(self):
+        self.check_daemon_threads_shutdown_deadlock('stdout')
+
+    def test_daemon_threads_shutdown_stderr_deadlock(self):
+        self.check_daemon_threads_shutdown_deadlock('stderr')
+
+
 class PyMiscIOTest(MiscIOTest):
     io = pyio
 
diff --git a/Lib/test/test_script_helper.py b/Lib/test/test_script_helper.py
index 7540e2e99bf..372d6a79ae9 100755
--- a/Lib/test/test_script_helper.py
+++ b/Lib/test/test_script_helper.py
@@ -8,26 +8,27 @@ from unittest import mock
 
 
 class TestScriptHelper(unittest.TestCase):
-    def test_assert_python_expect_success(self):
-        t = script_helper._assert_python(True, '-c', 'import sys; sys.exit(0)')
+
+    def test_assert_python_ok(self):
+        t = script_helper.assert_python_ok('-c', 'import sys; sys.exit(0)')
         self.assertEqual(0, t[0], 'return code was not 0')
 
-    def test_assert_python_expect_failure(self):
+    def test_assert_python_failure(self):
         # I didn't import the sys module so this child will fail.
-        rc, out, err = script_helper._assert_python(False, '-c', 'sys.exit(0)')
+        rc, out, err = script_helper.assert_python_failure('-c', 'sys.exit(0)')
         self.assertNotEqual(0, rc, 'return code should not be 0')
 
-    def test_assert_python_raises_expect_success(self):
+    def test_assert_python_ok_raises(self):
         # I didn't import the sys module so this child will fail.
         with self.assertRaises(AssertionError) as error_context:
-            script_helper._assert_python(True, '-c', 'sys.exit(0)')
+            script_helper.assert_python_ok('-c', 'sys.exit(0)')
         error_msg = str(error_context.exception)
         self.assertIn('command line was:', error_msg)
         self.assertIn('sys.exit(0)', error_msg, msg='unexpected command line')
 
-    def test_assert_python_raises_expect_failure(self):
+    def test_assert_python_failure_raises(self):
         with self.assertRaises(AssertionError) as error_context:
-            script_helper._assert_python(False, '-c', 'import sys; sys.exit(0)')
+            script_helper.assert_python_failure('-c', 'import sys; sys.exit(0)')
         error_msg = str(error_context.exception)
         self.assertIn('Process return code is 0,', error_msg)
         self.assertIn('import sys; sys.exit(0)', error_msg,
diff --git a/Misc/NEWS b/Misc/NEWS
index 539252e6823..fe91ae2c1a1 100644
--- a/Misc/NEWS
+++ b/Misc/NEWS
@@ -10,6 +10,11 @@ Release date: tba
 Core and Builtins
 -----------------
 
+- Issue #23309: Avoid a deadlock at shutdown if a daemon thread is aborted
+  while it is holding a lock to a buffered I/O object, and the main thread
+  tries to use the same I/O object (typically stdout or stderr).  A fatal
+  error is emitted instead.
+
 - Issue #22977: Fixed formatting Windows error messages on Wine.
   Patch by Martin Panter.
 
diff --git a/Modules/_io/bufferedio.c b/Modules/_io/bufferedio.c
index 445c8708ad7..3606cc8fe09 100644
--- a/Modules/_io/bufferedio.c
+++ b/Modules/_io/bufferedio.c
@@ -300,14 +300,35 @@ typedef struct {
 static int
 _enter_buffered_busy(buffered *self)
 {
+    int relax_locking;
+    PyLockStatus st;
     if (self->owner == PyThread_get_thread_ident()) {
         PyErr_Format(PyExc_RuntimeError,
                      "reentrant call inside %R", self);
         return 0;
     }
+    relax_locking = (_Py_Finalizing != NULL);
     Py_BEGIN_ALLOW_THREADS
-    PyThread_acquire_lock(self->lock, 1);
+    if (!relax_locking)
+        st = PyThread_acquire_lock(self->lock, 1);
+    else {
+        /* When finalizing, we don't want a deadlock to happen with daemon
+         * threads abruptly shut down while they owned the lock.
+         * Therefore, only wait for a grace period (1 s.).
+         * Note that non-daemon threads have already exited here, so this
+         * shouldn't affect carefully written threaded I/O code.
+         */
+        st = PyThread_acquire_lock_timed(self->lock, 1e6, 0);
+    }
     Py_END_ALLOW_THREADS
+    if (relax_locking && st != PY_LOCK_ACQUIRED) {
+        PyObject *msgobj = PyUnicode_FromFormat(
+            "could not acquire lock for %A at interpreter "
+            "shutdown, possibly due to daemon threads",
+            (PyObject *) self);
+        char *msg = PyUnicode_AsUTF8(msgobj);
+        Py_FatalError(msg);
+    }
     return 1;
 }