prompt: 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()
  
  ```
  
golden_diff: 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
  
verification_info: {"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"}]}