We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Exception raised in `TCPStore.__del__` upon process termination
### Description

`__del__` should not perform syncronization when called during process termination.

```
Exception ignored in: <function TCPStore.__del__ at 0x7fb939be23a0>
Traceback (most recent call last):
  File "/home/maehashi/Development/cupy/cupyx/distributed/_store.py", line 49, in __del__
  File "/home/maehashi/Development/cupy/cupyx/distributed/_store.py", line 97, in stop
  File "/home/maehashi/Development/cupy/cupyx/distributed/_store.py", line 31, in join
  File "/home/maehashi/.pyenv/versions/3.8.1/lib/python3.8/multiprocessing/connection.py", line 251, in recv
ModuleNotFoundError: import of builtins halted; None in sys.modules
```

### To Reproduce

_No response_

### Installation

_No response_

### Environment

_No response_

### Additional Information

_No response_

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `cupyx/distributed/_store.py`
Content:
```
1 from ctypes import sizeof
2 import multiprocessing
3 import threading
4 import socket
5 import time
6 
7 from cupyx.distributed import _klv_utils
8 from cupyx.distributed import _store_actions
9 
10 
11 _DEFAULT_HOST = '127.0.0.1'
12 _DEFAULT_PORT = 13333
13 
14 
15 class ExceptionAwareProcess(multiprocessing.Process):
16     def __init__(self, *args, **kwargs):
17         super().__init__(*args, **kwargs)
18         self._exception = None
19         self._parent_p, self._child_p = multiprocessing.Pipe()
20 
21     def run(self):
22         try:
23             super().run()
24             self._child_p.send(None)
25         except Exception as e:
26             self._child_p.send(e)
27 
28     def join(self):
29         super().join()
30         if self._parent_p.poll():
31             exception = self._parent_p.recv()
32             if exception is not None:
33                 raise exception
34 
35 
36 class TCPStore:
37     # This is only used for initialization of nccl so we don't care
38     # too much about peformance
39     def __init__(self, world_size):
40         self.storage = {}
41         self._process = None
42         self._world_size = world_size
43         self._run = multiprocessing.Value('b', 1)
44         # For implementing a barrier
45         self._lock = threading.Lock()
46         self._current_barrier = None
47 
48     def __del__(self):
49         self.stop()
50 
51     def _set_process(self, process):
52         self._process = process
53 
54     def _process_request(self, c_socket):
55         with c_socket:
56             # Receive in KLV format
57             action_bytes = c_socket.recv(sizeof(_klv_utils.action_t))
58             if len(action_bytes) > 0:
59                 action_m = _klv_utils.action_t.from_buffer_copy(action_bytes)
60                 if action_m.length > 256:
61                     raise ValueError('Invalid length for message')
62                 value = bytearray(action_m.value)[:action_m.length]
63                 r = _store_actions.execute_action(action_m.action, value, self)
64                 if r is not None:
65                     c_socket.sendall(r.klv())
66 
67     def _server_loop(self, host, port):
68         # This is for minimum info exchange during initialization
69         # a single connection allows to implement locking mechanics easily
70         with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
71             s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
72             s.bind((host, port))
73             s.listen()
74             s.settimeout(0.5)
75             while self._run.value == 1:
76                 try:
77                     c_socket, addr = s.accept()
78                 except socket.timeout:
79                     continue
80 
81                 t = threading.Thread(
82                     target=self._process_request,
83                     args=(c_socket,), daemon=True)
84                 t.start()
85 
86     def run(self, host=_DEFAULT_HOST, port=_DEFAULT_PORT):
87         # Run the TCP store in a different process
88         p = ExceptionAwareProcess(
89             target=self._server_loop, args=(host, port))
90         p.start()
91         self._process = p
92 
93     def stop(self):
94         if self._process is not None:
95             with self._run.get_lock():
96                 self._run.value = 0
97             self._process.join()
98 
99 
100 class TCPStoreProxy:
101 
102     MAX_NUM_RETRIES = 50
103     DELAY_FOR_RETRY = 0.5
104 
105     def __init__(self, host=_DEFAULT_HOST, port=_DEFAULT_PORT):
106         self.host = host
107         self.port = port
108 
109     def _send_recv(self, action):
110         # Retry several times in case the rank 0 has not established the
111         # main store yet
112         for i in range(TCPStoreProxy.MAX_NUM_RETRIES):
113             try:
114                 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
115                     # TODO retry connects
116                     s.connect((self.host, self.port))
117                     s.sendall(action.klv())
118                     result_bytes = s.recv(sizeof(
119                         _klv_utils.result_action_t))
120                     if len(result_bytes) > 0:
121                         result = _klv_utils.result_action_t.from_buffer_copy(
122                             result_bytes)
123                         value = bytearray(result.value)[:result.length]
124                         if result.status == 0:
125                             return action.decode_result(value)
126                         else:
127                             raise RuntimeError(value.decode('utf-8'))
128             except ConnectionRefusedError:
129                 time.sleep(TCPStoreProxy.DELAY_FOR_RETRY)
130         raise RuntimeError('TCPStore is not available')
131 
132     def __getitem__(self, key):
133         return self._send_recv(_store_actions.Get(key))
134 
135     def __setitem__(self, key, value):
136         self._send_recv(_store_actions.Set(key, value))
137 
138     def barrier(self):
139         # Barrier has special semantics
140         self._send_recv(_store_actions.Barrier())
141
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/cupyx/distributed/_store.py b/cupyx/distributed/_store.py
--- a/cupyx/distributed/_store.py
+++ b/cupyx/distributed/_store.py
@@ -1,3 +1,4 @@
+import atexit
 from ctypes import sizeof
 import multiprocessing
 import threading
@@ -11,6 +12,14 @@
 _DEFAULT_HOST = '127.0.0.1'
 _DEFAULT_PORT = 13333
 
+_exit_mode = False
+
+
+@atexit.register
+def _exit():
+    global _exit_mode
+    _exit_mode = True
+
 
 class ExceptionAwareProcess(multiprocessing.Process):
     def __init__(self, *args, **kwargs):
@@ -46,7 +55,8 @@
         self._current_barrier = None
 
     def __del__(self):
-        self.stop()
+        if not _exit_mode:
+            self.stop()
 
     def _set_process(self, process):
         self._process = process
@@ -91,6 +101,8 @@
         self._process = p
 
     def stop(self):
+        if _exit_mode:
+            return  # Prevent shutdown errors
         if self._process is not None:
             with self._run.get_lock():
                 self._run.value = 0

{"golden_diff": "diff --git a/cupyx/distributed/_store.py b/cupyx/distributed/_store.py\n--- a/cupyx/distributed/_store.py\n+++ b/cupyx/distributed/_store.py\n@@ -1,3 +1,4 @@\n+import atexit\n from ctypes import sizeof\n import multiprocessing\n import threading\n@@ -11,6 +12,14 @@\n _DEFAULT_HOST = '127.0.0.1'\n _DEFAULT_PORT = 13333\n \n+_exit_mode = False\n+\n+\n+@atexit.register\n+def _exit():\n+    global _exit_mode\n+    _exit_mode = True\n+\n \n class ExceptionAwareProcess(multiprocessing.Process):\n     def __init__(self, *args, **kwargs):\n@@ -46,7 +55,8 @@\n         self._current_barrier = None\n \n     def __del__(self):\n-        self.stop()\n+        if not _exit_mode:\n+            self.stop()\n \n     def _set_process(self, process):\n         self._process = process\n@@ -91,6 +101,8 @@\n         self._process = p\n \n     def stop(self):\n+        if _exit_mode:\n+            return  # Prevent shutdown errors\n         if self._process is not None:\n             with self._run.get_lock():\n                 self._run.value = 0\n", "issue": "Exception raised in `TCPStore.__del__` upon process termination\n### Description\n\n`__del__` should not perform syncronization when called during process termination.\r\n\r\n```\r\nException ignored in: <function TCPStore.__del__ at 0x7fb939be23a0>\r\nTraceback (most recent call last):\r\n  File \"/home/maehashi/Development/cupy/cupyx/distributed/_store.py\", line 49, in __del__\r\n  File \"/home/maehashi/Development/cupy/cupyx/distributed/_store.py\", line 97, in stop\r\n  File \"/home/maehashi/Development/cupy/cupyx/distributed/_store.py\", line 31, in join\r\n  File \"/home/maehashi/.pyenv/versions/3.8.1/lib/python3.8/multiprocessing/connection.py\", line 251, in recv\r\nModuleNotFoundError: import of builtins halted; None in sys.modules\r\n```\n\n### To Reproduce\n\n_No response_\n\n### Installation\n\n_No response_\n\n### Environment\n\n_No response_\n\n### Additional Information\n\n_No response_\n", "before_files": [{"content": "from ctypes import sizeof\nimport multiprocessing\nimport threading\nimport socket\nimport time\n\nfrom cupyx.distributed import _klv_utils\nfrom cupyx.distributed import _store_actions\n\n\n_DEFAULT_HOST = '127.0.0.1'\n_DEFAULT_PORT = 13333\n\n\nclass ExceptionAwareProcess(multiprocessing.Process):\n    def __init__(self, *args, **kwargs):\n        super().__init__(*args, **kwargs)\n        self._exception = None\n        self._parent_p, self._child_p = multiprocessing.Pipe()\n\n    def run(self):\n        try:\n            super().run()\n            self._child_p.send(None)\n        except Exception as e:\n            self._child_p.send(e)\n\n    def join(self):\n        super().join()\n        if self._parent_p.poll():\n            exception = self._parent_p.recv()\n            if exception is not None:\n                raise exception\n\n\nclass TCPStore:\n    # This is only used for initialization of nccl so we don't care\n    # too much about peformance\n    def __init__(self, world_size):\n        self.storage = {}\n        self._process = None\n        self._world_size = world_size\n        self._run = multiprocessing.Value('b', 1)\n        # For implementing a barrier\n        self._lock = threading.Lock()\n        self._current_barrier = None\n\n    def __del__(self):\n        self.stop()\n\n    def _set_process(self, process):\n        self._process = process\n\n    def _process_request(self, c_socket):\n        with c_socket:\n            # Receive in KLV format\n            action_bytes = c_socket.recv(sizeof(_klv_utils.action_t))\n            if len(action_bytes) > 0:\n                action_m = _klv_utils.action_t.from_buffer_copy(action_bytes)\n                if action_m.length > 256:\n                    raise ValueError('Invalid length for message')\n                value = bytearray(action_m.value)[:action_m.length]\n                r = _store_actions.execute_action(action_m.action, value, self)\n                if r is not None:\n                    c_socket.sendall(r.klv())\n\n    def _server_loop(self, host, port):\n        # This is for minimum info exchange during initialization\n        # a single connection allows to implement locking mechanics easily\n        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:\n            s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)\n            s.bind((host, port))\n            s.listen()\n            s.settimeout(0.5)\n            while self._run.value == 1:\n                try:\n                    c_socket, addr = s.accept()\n                except socket.timeout:\n                    continue\n\n                t = threading.Thread(\n                    target=self._process_request,\n                    args=(c_socket,), daemon=True)\n                t.start()\n\n    def run(self, host=_DEFAULT_HOST, port=_DEFAULT_PORT):\n        # Run the TCP store in a different process\n        p = ExceptionAwareProcess(\n            target=self._server_loop, args=(host, port))\n        p.start()\n        self._process = p\n\n    def stop(self):\n        if self._process is not None:\n            with self._run.get_lock():\n                self._run.value = 0\n            self._process.join()\n\n\nclass TCPStoreProxy:\n\n    MAX_NUM_RETRIES = 50\n    DELAY_FOR_RETRY = 0.5\n\n    def __init__(self, host=_DEFAULT_HOST, port=_DEFAULT_PORT):\n        self.host = host\n        self.port = port\n\n    def _send_recv(self, action):\n        # Retry several times in case the rank 0 has not established the\n        # main store yet\n        for i in range(TCPStoreProxy.MAX_NUM_RETRIES):\n            try:\n                with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:\n                    # TODO retry connects\n                    s.connect((self.host, self.port))\n                    s.sendall(action.klv())\n                    result_bytes = s.recv(sizeof(\n                        _klv_utils.result_action_t))\n                    if len(result_bytes) > 0:\n                        result = _klv_utils.result_action_t.from_buffer_copy(\n                            result_bytes)\n                        value = bytearray(result.value)[:result.length]\n                        if result.status == 0:\n                            return action.decode_result(value)\n                        else:\n                            raise RuntimeError(value.decode('utf-8'))\n            except ConnectionRefusedError:\n                time.sleep(TCPStoreProxy.DELAY_FOR_RETRY)\n        raise RuntimeError('TCPStore is not available')\n\n    def __getitem__(self, key):\n        return self._send_recv(_store_actions.Get(key))\n\n    def __setitem__(self, key, value):\n        self._send_recv(_store_actions.Set(key, value))\n\n    def barrier(self):\n        # Barrier has special semantics\n        self._send_recv(_store_actions.Barrier())\n", "path": "cupyx/distributed/_store.py"}], "after_files": [{"content": "import atexit\nfrom ctypes import sizeof\nimport multiprocessing\nimport threading\nimport socket\nimport time\n\nfrom cupyx.distributed import _klv_utils\nfrom cupyx.distributed import _store_actions\n\n\n_DEFAULT_HOST = '127.0.0.1'\n_DEFAULT_PORT = 13333\n\n_exit_mode = False\n\n\n@atexit.register\ndef _exit():\n    global _exit_mode\n    _exit_mode = True\n\n\nclass ExceptionAwareProcess(multiprocessing.Process):\n    def __init__(self, *args, **kwargs):\n        super().__init__(*args, **kwargs)\n        self._exception = None\n        self._parent_p, self._child_p = multiprocessing.Pipe()\n\n    def run(self):\n        try:\n            super().run()\n            self._child_p.send(None)\n        except Exception as e:\n            self._child_p.send(e)\n\n    def join(self):\n        super().join()\n        if self._parent_p.poll():\n            exception = self._parent_p.recv()\n            if exception is not None:\n                raise exception\n\n\nclass TCPStore:\n    # This is only used for initialization of nccl so we don't care\n    # too much about peformance\n    def __init__(self, world_size):\n        self.storage = {}\n        self._process = None\n        self._world_size = world_size\n        self._run = multiprocessing.Value('b', 1)\n        # For implementing a barrier\n        self._lock = threading.Lock()\n        self._current_barrier = None\n\n    def __del__(self):\n        if not _exit_mode:\n            self.stop()\n\n    def _set_process(self, process):\n        self._process = process\n\n    def _process_request(self, c_socket):\n        with c_socket:\n            # Receive in KLV format\n            action_bytes = c_socket.recv(sizeof(_klv_utils.action_t))\n            if len(action_bytes) > 0:\n                action_m = _klv_utils.action_t.from_buffer_copy(action_bytes)\n                if action_m.length > 256:\n                    raise ValueError('Invalid length for message')\n                value = bytearray(action_m.value)[:action_m.length]\n                r = _store_actions.execute_action(action_m.action, value, self)\n                if r is not None:\n                    c_socket.sendall(r.klv())\n\n    def _server_loop(self, host, port):\n        # This is for minimum info exchange during initialization\n        # a single connection allows to implement locking mechanics easily\n        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:\n            s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)\n            s.bind((host, port))\n            s.listen()\n            s.settimeout(0.5)\n            while self._run.value == 1:\n                try:\n                    c_socket, addr = s.accept()\n                except socket.timeout:\n                    continue\n\n                t = threading.Thread(\n                    target=self._process_request,\n                    args=(c_socket,), daemon=True)\n                t.start()\n\n    def run(self, host=_DEFAULT_HOST, port=_DEFAULT_PORT):\n        # Run the TCP store in a different process\n        p = ExceptionAwareProcess(\n            target=self._server_loop, args=(host, port))\n        p.start()\n        self._process = p\n\n    def stop(self):\n        if _exit_mode:\n            return  # Prevent shutdown errors\n        if self._process is not None:\n            with self._run.get_lock():\n                self._run.value = 0\n            self._process.join()\n\n\nclass TCPStoreProxy:\n\n    MAX_NUM_RETRIES = 50\n    DELAY_FOR_RETRY = 0.5\n\n    def __init__(self, host=_DEFAULT_HOST, port=_DEFAULT_PORT):\n        self.host = host\n        self.port = port\n\n    def _send_recv(self, action):\n        # Retry several times in case the rank 0 has not established the\n        # main store yet\n        for i in range(TCPStoreProxy.MAX_NUM_RETRIES):\n            try:\n                with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:\n                    # TODO retry connects\n                    s.connect((self.host, self.port))\n                    s.sendall(action.klv())\n                    result_bytes = s.recv(sizeof(\n                        _klv_utils.result_action_t))\n                    if len(result_bytes) > 0:\n                        result = _klv_utils.result_action_t.from_buffer_copy(\n                            result_bytes)\n                        value = bytearray(result.value)[:result.length]\n                        if result.status == 0:\n                            return action.decode_result(value)\n                        else:\n                            raise RuntimeError(value.decode('utf-8'))\n            except ConnectionRefusedError:\n                time.sleep(TCPStoreProxy.DELAY_FOR_RETRY)\n        raise RuntimeError('TCPStore is not available')\n\n    def __getitem__(self, key):\n        return self._send_recv(_store_actions.Get(key))\n\n    def __setitem__(self, key, value):\n        self._send_recv(_store_actions.Set(key, value))\n\n    def barrier(self):\n        # Barrier has special semantics\n        self._send_recv(_store_actions.Barrier())\n", "path": "cupyx/distributed/_store.py"}]}