diff --git a/Lib/multiprocessing/synchronize.py b/Lib/multiprocessing/synchronize.py
index 3ccbfe311c71f37..4772e95e485b1eb 100644
--- a/Lib/multiprocessing/synchronize.py
+++ b/Lib/multiprocessing/synchronize.py
@@ -9,7 +9,7 @@
 
 __all__ = [
     'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore', 'Condition', 'Event'
-    ]
+]
 
 import threading
 import sys
@@ -157,7 +157,7 @@ def __repr__(self):
         except Exception:
             value = 'unknown'
         return '<%s(value=%s, maxvalue=%s)>' % \
-               (self.__class__.__name__, value, self._semlock.maxvalue)
+            (self.__class__.__name__, value, self._semlock.maxvalue)
 
 #
 # Non-recursive lock
@@ -253,7 +253,7 @@ def __repr__(self):
 
     def wait(self, timeout=None):
         assert self._lock._semlock._is_mine(), \
-               'must acquire() condition before using wait()'
+            'must acquire() condition before using wait()'
 
         # indicate that this thread is going to sleep
         self._sleeping_count.release()
diff --git a/Lib/signal.py b/Lib/signal.py
index c8cd3d4f597ca50..2b2381b73c6c8a6 100644
--- a/Lib/signal.py
+++ b/Lib/signal.py
@@ -1,6 +1,10 @@
 import _signal
+import sys
 from _signal import *
 from enum import IntEnum as _IntEnum
+import threading
+import queue
+import traceback
 
 _globals = globals()
 
@@ -42,6 +46,74 @@ def _enum_to_int(value):
     except (ValueError, TypeError):
         return value
 
+_signal_queue = queue.SimpleQueue() # SimpleQueue has reentrant put, so it can safely be called from signal handlers. https://github.com/python/cpython/issues/59181
+_sys_exit_queue = queue.SimpleQueue()
+_signal_thread = None
+_signo_to_handler = {}
+
+def _init_signal_thread():
+    assert threading.current_thread() is threading.main_thread()
+    global _signal_thread
+    if _signal_thread is None:
+        _signal_thread = threading.Thread(target=_signal_queue_handler, daemon=True)
+        _signal_thread.name = 'SignalHandlerThread'
+        _signal_thread.start()
+
+def _push_signal_to_queue_handler(signo, _stack_frame):
+    assert threading.current_thread() is threading.main_thread()
+    global _signal_queue, _sys_exit_queue
+    try:
+        exit_code = _sys_exit_queue.get(block=False)
+        sys.exit(exit_code)
+    except queue.Empty:
+        _signal_queue.put(signo)
+
+def _sigint_to_str(signo):
+    for x in valid_signals():
+        if x == signo:
+            return x.name
+    raise RuntimeError('Could not find signal name')
+
+def _log_missing_signal_handler(signo):
+    import logging
+    logger = logging.getLogger(__name__)
+    str_name = ''
+    for x in valid_signals():
+        if x == signo:
+            str_name = x.name
+    logger.warning('Handler for signal.%s (%d) was not found.', str_name, signo)
+
+def stop_signal_thread():
+    global _signal_thread, _signal_queue
+    if _signal_thread is not None:
+        _signal_queue.put('STOP_SIGNAL_HANDLER')
+        _signal_thread.join()
+        _signal_thread = None
+
+def _signal_queue_handler():
+    assert threading.current_thread() is not threading.main_thread()
+    global _signal_queue, _signo_to_handler
+    while True:
+        signo = _signal_queue.get()
+        if signo == 'STOP_SIGNAL_HANDLER':
+            break
+        raise_systemexit = False
+        exitcode = 'NOTSET'
+        try:
+            handler = _signo_to_handler.get(signo, None)
+            if handler is not None:
+                handler(signo, None)
+            else:
+                _log_missing_signal_handler(signo)
+        except SystemExit as se:
+            exitcode = se.code
+            raise_systemexit = True
+        except Exception:
+            traceback.print_exc()
+        if raise_systemexit:
+            global _sys_exit_queue
+            _sys_exit_queue.put(exitcode)
+            raise_signal(signo)
 
 # Similar to functools.wraps(), but only assign __doc__.
 # __module__ should be preserved,
@@ -53,16 +125,37 @@ def decorator(wrapper):
         return wrapper
     return decorator
 
-@_wraps(_signal.signal)
-def signal(signalnum, handler):
-    handler = _signal.signal(_enum_to_int(signalnum), _enum_to_int(handler))
-    return _int_to_enum(handler, Handlers)
+def signal(signalnum, handler, use_dedicated_thread=True):
+    if use_dedicated_thread:
+        assert threading.current_thread() is threading.main_thread()
+    global _signo_to_handler
+    signal_int = _enum_to_int(signalnum)
+    old_handler = _signo_to_handler.get(signal_int, None)
+    if use_dedicated_thread and callable(handler):
+        assert callable(handler)
+        global _signal_thread
+        if _signal_thread is None:
+            _init_signal_thread()
+        _signo_to_handler[signal_int] = handler
+        handler = _signal.signal(signal_int, _enum_to_int(_push_signal_to_queue_handler))
+        return old_handler or _int_to_enum(handler, Handlers)
+    else:
+        if signal_int in _signo_to_handler:
+            del _signo_to_handler[signal_int]
+        if 0 == len(_signo_to_handler):
+            stop_signal_thread()
+        handler = _signal.signal(signal_int, _enum_to_int(handler))
+        return old_handler or _int_to_enum(handler, Handlers)
 
 
 @_wraps(_signal.getsignal)
 def getsignal(signalnum):
-    handler = _signal.getsignal(signalnum)
-    return _int_to_enum(handler, Handlers)
+    global _signo_to_handler
+    if signalnum in _signo_to_handler:
+        return _signo_to_handler[signalnum]
+    else:
+        handler = _signal.getsignal(signalnum)
+        return _int_to_enum(handler, Handlers)
 
 
 if 'pthread_sigmask' in _globals:
diff --git a/Lib/test/multiprocessingdata/is_set_set.py b/Lib/test/multiprocessingdata/is_set_set.py
new file mode 100644
index 000000000000000..31d6d18e0c6c7a1
--- /dev/null
+++ b/Lib/test/multiprocessingdata/is_set_set.py
@@ -0,0 +1,29 @@
+import multiprocessing
+import os
+import signal
+import concurrent.futures
+import time
+
+
+def send_sigint(pid):
+    time.sleep(1)
+    os.kill(pid, signal.SIGINT)
+
+
+def run_signal_handler_set_is_set_test():
+    shutdown_event = multiprocessing.Event()
+
+    def sigterm_handler(_signo, _stack_frame):
+        shutdown_event.set()
+
+    signal.signal(signal.SIGINT, sigterm_handler)
+
+    with concurrent.futures.ProcessPoolExecutor() as executor:
+        f = executor.submit(send_sigint, os.getpid())
+        while not shutdown_event.is_set():
+            pass
+        f.result()
+
+
+if __name__ == '__main__':
+    run_signal_handler_set_is_set_test()
diff --git a/Lib/test/multiprocessingdata/set_clear_race.py b/Lib/test/multiprocessingdata/set_clear_race.py
new file mode 100644
index 000000000000000..47ce99b3a36dcb0
--- /dev/null
+++ b/Lib/test/multiprocessingdata/set_clear_race.py
@@ -0,0 +1,36 @@
+import multiprocessing
+import sys
+
+
+# Reproduction code copied and modified from https://github.com/python/cpython/issues/95826
+# Fixes the issue above
+
+class SimpleRepro:
+    def __init__(self):
+        self.heartbeat_event = multiprocessing.Event()
+        self.shutdown_event = multiprocessing.Event()
+        self.child_proc = multiprocessing.Process(target=self.child_process, daemon=True)
+        self.child_proc.start()
+
+    def child_process(self):
+        while True:
+            if self.shutdown_event.is_set():
+                return
+            self.heartbeat_event.set()
+            self.heartbeat_event.clear()
+
+    def test_heartbeat(self):
+        exit_code = 0
+        for i in range(2000):
+            success = self.heartbeat_event.wait(100)
+            if not success:
+                exit_code = 1
+                break
+        self.shutdown_event.set()
+        self.child_proc.join()
+        sys.exit(exit_code)
+
+
+if __name__ == '__main__':
+    foo = SimpleRepro()
+    foo.test_heartbeat()
diff --git a/Lib/test/multiprocessingdata/wait_set_no_deadlock.py b/Lib/test/multiprocessingdata/wait_set_no_deadlock.py
new file mode 100644
index 000000000000000..26160a3f2278059
--- /dev/null
+++ b/Lib/test/multiprocessingdata/wait_set_no_deadlock.py
@@ -0,0 +1,30 @@
+import multiprocessing
+import signal
+import concurrent.futures
+import time
+import os
+
+
+# Shows that https://github.com/python/cpython/issues/85772 is fixed
+
+def send_sigint(pid):
+    time.sleep(1) # Make sure shutdown_event.wait() is called
+    os.kill(pid, signal.SIGINT)
+
+
+def run_signal_handler_wait_set_test():
+    shutdown_event = multiprocessing.Event()
+
+    def sigterm_handler(_signo, _stack_frame):
+        shutdown_event.set()
+
+    signal.signal(signal.SIGINT, sigterm_handler)
+
+    with concurrent.futures.ProcessPoolExecutor() as executor:
+        f = executor.submit(send_sigint, os.getpid())
+        shutdown_event.wait()
+        f.result()
+
+
+if __name__ == '__main__':
+    run_signal_handler_wait_set_test()
diff --git a/Lib/test/support/threading_helper.py b/Lib/test/support/threading_helper.py
index afa25a76f638295..ab8608df93f139e 100644
--- a/Lib/test/support/threading_helper.py
+++ b/Lib/test/support/threading_helper.py
@@ -5,6 +5,7 @@
 import threading
 import time
 import unittest
+import signal
 
 from test import support
 
@@ -28,6 +29,8 @@ def threading_setup():
 def threading_cleanup(*original_values):
     orig_count, orig_ndangling = original_values
 
+    signal.stop_signal_thread()
+
     timeout = 1.0
     for _ in support.sleeping_retry(timeout, error=False):
         # Copy the thread list to get a consistent output. threading._dangling
diff --git a/Lib/test/test_asyncio/test_base_events.py b/Lib/test/test_asyncio/test_base_events.py
index c14a0bb180d79bd..6bd4df41c553f45 100644
--- a/Lib/test/test_asyncio/test_base_events.py
+++ b/Lib/test/test_asyncio/test_base_events.py
@@ -4,6 +4,7 @@
 import errno
 import math
 import platform
+import signal
 import socket
 import sys
 import threading
@@ -26,6 +27,7 @@
 
 def tearDownModule():
     asyncio.set_event_loop_policy(None)
+    signal.stop_signal_thread()
 
 
 def mock_socket_module():
diff --git a/Lib/test/test_asyncio/test_context.py b/Lib/test/test_asyncio/test_context.py
index 6b80721873d95cb..54770f926a8fe9e 100644
--- a/Lib/test/test_asyncio/test_context.py
+++ b/Lib/test/test_asyncio/test_context.py
@@ -1,10 +1,12 @@
 import asyncio
 import decimal
 import unittest
+import signal
 
 
 def tearDownModule():
     asyncio.set_event_loop_policy(None)
+    signal.stop_signal_thread()
 
 
 @unittest.skipUnless(decimal.HAVE_CONTEXTVAR, "decimal is built with a thread-local context")
diff --git a/Lib/test/test_asyncio/test_futures2.py b/Lib/test/test_asyncio/test_futures2.py
index b7cfffb76bd8f17..1f1ef25c049c1a1 100644
--- a/Lib/test/test_asyncio/test_futures2.py
+++ b/Lib/test/test_asyncio/test_futures2.py
@@ -3,11 +3,13 @@
 import contextvars
 import traceback
 import unittest
+import signal
 from asyncio import tasks
 
 
 def tearDownModule():
     asyncio.set_event_loop_policy(None)
+    signal.stop_signal_thread()
 
 
 class FutureTests:
diff --git a/Lib/test/test_asyncio/test_locks.py b/Lib/test/test_asyncio/test_locks.py
index c3bff760f7307e3..2b16bc52a2aab0f 100644
--- a/Lib/test/test_asyncio/test_locks.py
+++ b/Lib/test/test_asyncio/test_locks.py
@@ -1,5 +1,5 @@
 """Tests for locks.py"""
-
+import signal
 import unittest
 from unittest import mock
 import re
@@ -21,6 +21,7 @@
 
 def tearDownModule():
     asyncio.set_event_loop_policy(None)
+    signal.stop_signal_thread()
 
 
 class LockTests(unittest.IsolatedAsyncioTestCase):
diff --git a/Lib/test/test_asyncio/test_queues.py b/Lib/test/test_asyncio/test_queues.py
index 5019e9a293525d1..f5e5586c3a8abcf 100644
--- a/Lib/test/test_asyncio/test_queues.py
+++ b/Lib/test/test_asyncio/test_queues.py
@@ -1,13 +1,14 @@
 """Tests for queues.py"""
 
 import asyncio
+import signal
 import unittest
 from types import GenericAlias
 
 
 def tearDownModule():
     asyncio.set_event_loop_policy(None)
-
+    signal.stop_signal_thread()
 
 class QueueBasicTests(unittest.IsolatedAsyncioTestCase):
 
diff --git a/Lib/test/test_asyncio/test_runners.py b/Lib/test/test_asyncio/test_runners.py
index 45f70d09a2083a0..476aa05e2fbd88b 100644
--- a/Lib/test/test_asyncio/test_runners.py
+++ b/Lib/test/test_asyncio/test_runners.py
@@ -13,6 +13,7 @@
 
 def tearDownModule():
     asyncio.set_event_loop_policy(None)
+    signal.stop_signal_thread()
 
 
 def interrupt_self():
diff --git a/Lib/test/test_asyncio/test_server.py b/Lib/test/test_asyncio/test_server.py
index 60a40cc8349fed5..04d86ce9dfef611 100644
--- a/Lib/test/test_asyncio/test_server.py
+++ b/Lib/test/test_asyncio/test_server.py
@@ -1,5 +1,6 @@
 import asyncio
 import os
+import signal
 import socket
 import time
 import threading
@@ -12,6 +13,7 @@
 
 def tearDownModule():
     asyncio.set_event_loop_policy(None)
+    signal.stop_signal_thread()
 
 
 class BaseStartServer(func_tests.FunctionalTestCaseMixin):
diff --git a/Lib/test/test_asyncio/test_staggered.py b/Lib/test/test_asyncio/test_staggered.py
index 74941f704c48901..ef1394d1be53782 100644
--- a/Lib/test/test_asyncio/test_staggered.py
+++ b/Lib/test/test_asyncio/test_staggered.py
@@ -1,4 +1,5 @@
 import asyncio
+import signal
 import unittest
 from asyncio.staggered import staggered_race
 
@@ -9,6 +10,7 @@
 
 def tearDownModule():
     asyncio.set_event_loop_policy(None)
+    signal.stop_signal_thread()
 
 
 class StaggeredTests(unittest.IsolatedAsyncioTestCase):
diff --git a/Lib/test/test_asyncio/test_taskgroups.py b/Lib/test/test_asyncio/test_taskgroups.py
index 1b4de96a572fb94..7c2697190b17464 100644
--- a/Lib/test/test_asyncio/test_taskgroups.py
+++ b/Lib/test/test_asyncio/test_taskgroups.py
@@ -1,6 +1,6 @@
 # Adapted with permission from the EdgeDB project;
 # license: PSFL.
-
+import signal
 import sys
 import gc
 import asyncio
@@ -15,7 +15,7 @@
 # To prevent a warning "test altered the execution environment"
 def tearDownModule():
     asyncio.set_event_loop_policy(None)
-
+    signal.stop_signal_thread()
 
 class MyExc(Exception):
     pass
diff --git a/Lib/test/test_asyncio/test_threads.py b/Lib/test/test_asyncio/test_threads.py
index 774380270a7d70a..39c5d4a49296ee8 100644
--- a/Lib/test/test_asyncio/test_threads.py
+++ b/Lib/test/test_asyncio/test_threads.py
@@ -1,6 +1,7 @@
 """Tests for asyncio/threads.py"""
 
 import asyncio
+import signal
 import unittest
 
 from contextvars import ContextVar
@@ -9,7 +10,7 @@
 
 def tearDownModule():
     asyncio.set_event_loop_policy(None)
-
+    signal.stop_signal_thread()
 
 class ToThreadTests(unittest.IsolatedAsyncioTestCase):
     async def test_to_thread(self):
diff --git a/Lib/test/test_asyncio/test_timeouts.py b/Lib/test/test_asyncio/test_timeouts.py
index f5543e191d07ff9..283391daa29482b 100644
--- a/Lib/test/test_asyncio/test_timeouts.py
+++ b/Lib/test/test_asyncio/test_timeouts.py
@@ -1,5 +1,5 @@
 """Tests for asyncio/timeouts.py"""
-
+import signal
 import unittest
 import time
 
@@ -10,6 +10,7 @@
 
 def tearDownModule():
     asyncio.set_event_loop_policy(None)
+    signal.stop_signal_thread()
 
 class TimeoutTests(unittest.IsolatedAsyncioTestCase):
 
diff --git a/Lib/test/test_asyncio/test_unix_events.py b/Lib/test/test_asyncio/test_unix_events.py
index 021f45478d6f484..e09a196ddc971e4 100644
--- a/Lib/test/test_asyncio/test_unix_events.py
+++ b/Lib/test/test_asyncio/test_unix_events.py
@@ -34,6 +34,7 @@
 
 def tearDownModule():
     asyncio.set_event_loop_policy(None)
+    signal.stop_signal_thread()
 
 
 MOCK_ANY = mock.ANY
diff --git a/Makefile.pre.in b/Makefile.pre.in
index c650ecaf7be1373..c7ff502ec7c7b67 100644
--- a/Makefile.pre.in
+++ b/Makefile.pre.in
@@ -2449,6 +2449,7 @@ TESTSUBDIRS=	idlelib/idle_test \
 		test/leakers \
 		test/libregrtest \
 		test/mathdata \
+		test/multiprocessingdata \
 		test/regrtestdata \
 		test/regrtestdata/import_from_tests \
 		test/regrtestdata/import_from_tests/test_regrtest_b \
diff --git a/Misc/NEWS.d/next/Library/2024-11-05-11-11-29.gh-issue-126434.Hw3wdI.rst b/Misc/NEWS.d/next/Library/2024-11-05-11-11-29.gh-issue-126434.Hw3wdI.rst
new file mode 100644
index 000000000000000..2333aab7eea7b75
--- /dev/null
+++ b/Misc/NEWS.d/next/Library/2024-11-05-11-11-29.gh-issue-126434.Hw3wdI.rst
@@ -0,0 +1 @@
+All of :mod:`multiprocessing` ``Event`` is now reentrant and thread safe, and can thus be used from signal handlers.
diff --git a/bug.py b/bug.py
new file mode 100755
index 000000000000000..317fd50f02b087d
--- /dev/null
+++ b/bug.py
@@ -0,0 +1,56 @@
+import os
+import signal
+import threading
+import time
+import multiprocessing
+import sys
+
+def sigint_self():
+    time.sleep(1)
+    print(f'{threading.current_thread().name}: Stopping PID {os.getpid()} with SIGINT')
+    os.kill(os.getpid(), signal.SIGINT)
+
+def sigkill_self():
+    time.sleep(5)
+    print(f'{threading.current_thread().name}: Killing PID {os.getpid()} with SIGKILL')
+    os.kill(os.getpid(), signal.SIGKILL)
+
+def run_signal_handler_dedicated_thread():
+    event = multiprocessing.Event()
+    def sigint_handler(_signo, _stack_frame):
+        try:
+            # print(f'{threading.current_thread().name}: sigint_handler raising SIGUSR1 ...')
+            # signal.raise_signal(signal.SIGUSR1)
+            # print(f'{threading.current_thread().name}: sigint_handler raising SIGUSR1 ... OK')
+            sys.exit()
+        finally:
+            print(f'{threading.current_thread().name}: sigint_handler exiting ...')
+
+    def sigusr1_handler(_signo, _):
+        print(f'{threading.current_thread().name}: USR1 running')
+
+    def sigterm_handler(_signo, _stack_frame):
+        print(f'{threading.current_thread().name}: sigterm_handler is running')
+        pass
+
+    signal.signal(signal.SIGINT, sigint_handler, True)
+    signal.signal(signal.SIGUSR1, sigusr1_handler, True)
+
+    # signal.raise_signal(signal.SIGINT)
+
+    threading.Thread(target=sigint_self, daemon=True).start()
+    threading.Thread(target=sigkill_self, daemon=True).start() # Used for debugging only.
+
+    print(f'{threading.current_thread().name}: Waiting on event. PID = {os.getpid()}')
+    while True:
+        if event.is_set():
+            break
+        else:
+            time.sleep(0.1)
+    print(f'{threading.current_thread().name}: Waiting is done')
+
+if __name__ == '__main__':
+    try:
+        run_signal_handler_dedicated_thread()
+    finally:
+        print(f'{threading.current_thread().name}: Exiting')