Stop HistorySavingThread before fork (ipython#15115)

Python 3.12+ issues a DeprecationWarning if `os.fork()` is called while
there are multiple threads, and in fact this is not strictly safe on any
version of Python, although on Linux with glibc it mostly works well
enough that we don't notice. The man page for `fork()` says:

> After a fork() in a multithreaded program, the child can safely call
only async-signal-safe functions (see signal-safety(7)) until such time
as it calls execve(2).

This is incompatible with executing any Python code, hence the warning.
IPython is always multi-threaded because of HistorySavingThread, so it's
never safe to call `os.fork()` or use multiprocessing's fork context.
However, I think we can easily resolve this by stopping the history
saving thread before fork, and starting it again after, so at the moment
of fork it's a single-threaded process. In principle, I think it's safe
to start it again in the `after_in_parent` handler, but Python's check
for threads to issue the warning runs after that callback, so to avoid
triggering the warning, it's better to start it again when there's
something for it to do.

This unfortunately only solves the issue for IPython in the terminal,
since IPykernel uses several more threads. But fixing it in the terminal
is already useful, and this would also be one small step towards fixing
it in IPykernel.

-----

Testing:

```python
import os
import time

def func_with_fork():
    if (pid := os.fork()) > 0:
        print("Parent")
        t = os.waitpid(pid, 0)
        print("Parent: child finished", t)
    else:
        print("Child")
        time.sleep(0.5)
        print("Child finishing")
        os._exit(0)

if __name__ == "__main__":
    func_with_fork()
```

```shell
PYTHONWARNINGS=default ipython
```

Author: Carreau

IPython/core/history.py

@@ -8,6 +8,7 @@
 
 import atexit
 import datetime
+import os
 import re
 
 
@@ -689,6 +690,7 @@ def __init__(
             )
             self.hist_file = ":memory:"
 
+        self.using_thread = False
         if self.enabled and self.hist_file != ":memory:":
             self.save_thread = HistorySavingThread(self)
             try:
@@ -699,6 +701,8 @@ def __init__(
                     exc_info=True,
                 )
                 self.hist_file = ":memory:"
+            else:
+                self.using_thread = True
         self._instances.add(self)
         assert len(HistoryManager._instances) <= HistoryManager._max_inst, (
             len(HistoryManager._instances),
@@ -709,6 +713,20 @@ def __del__(self) -> None:
         if self.save_thread is not None:
             self.save_thread.stop()
 
+    @classmethod
+    def _stop_thread(cls) -> None:
+        # Used before forking so the thread isn't running at fork
+        for inst in cls._instances:
+            if inst.save_thread is not None:
+                inst.save_thread.stop()
+                inst.save_thread = None
+
+    def _restart_thread_if_stopped(self) -> None:
+        # Start the thread again after it was stopped for forking
+        if self.save_thread is None and self.using_thread:
+            self.save_thread = HistorySavingThread(self)
+            self.save_thread.start()
+
     def _get_hist_file_name(self, profile: Optional[str] = None) -> Path:
         """Get default history file name based on the Shell's profile.
 
@@ -970,8 +988,10 @@ def store_inputs(
             self.db_input_cache.append((line_num, source, source_raw))
             # Trigger to flush cache and write to DB.
             if len(self.db_input_cache) >= self.db_cache_size:
-                if self.save_flag:
-                    self.save_flag.set()
+                if self.using_thread:
+                    self._restart_thread_if_stopped()
+                    if self.save_flag is not None:
+                        self.save_flag.set()
 
         # update the auto _i variables
         self._iii = self._ii
@@ -1003,8 +1023,10 @@ def store_output(self, line_num: int) -> None:
 
         with self.db_output_cache_lock:
             self.db_output_cache.append((line_num, output))
-        if self.db_cache_size <= 1 and self.save_flag is not None:
-            self.save_flag.set()
+        if self.db_cache_size <= 1 and self.using_thread:
+            self._restart_thread_if_stopped()
+            if self.save_flag is not None:
+                self.save_flag.set()
 
     def _writeout_input_cache(self, conn: sqlite3.Connection) -> None:
         with conn:
@@ -1059,6 +1081,10 @@ def writeout_cache(self, conn: Optional[sqlite3.Connection] = None) -> None:
                 self.db_output_cache = []
 
 
+if hasattr(os, "register_at_fork"):
+    os.register_at_fork(before=HistoryManager._stop_thread)
+
+
 from collections.abc import Callable, Iterator
 from weakref import ReferenceType