We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Tempo example fails when parallel inference is enabled
When parallel inference is enabled, the [outlier example using the Tempo runtime](https://tempo.readthedocs.io/en/latest/examples/outlier/README.html)seems to fail. In particular, it seems that either the `cifar10-service` or the `outlier` containers block the request path and never return a 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: `mlserver/parallel.py`
Content:
```
1 import asyncio
2 
3 from functools import wraps
4 from concurrent.futures import ProcessPoolExecutor
5 from typing import Any, Coroutine, Callable, Optional
6 
7 from .errors import MLServerError
8 from .settings import ModelSettings
9 from .model import MLModel
10 from .types import InferenceRequest, InferenceResponse
11 
12 _InferencePoolAttr = "__inference_pool__"
13 
14 # NOTE: Workaround for mypy
15 _mp_model: MLModel
16 
17 
18 class InvalidParallelMethod(MLServerError):
19     def __init__(self, method_name: str, reason: Optional[str] = None):
20         msg = f"Method {method_name} can't be parallelised"
21         if reason:
22             msg += f": {reason}"
23 
24         super().__init__(msg)
25 
26 
27 def _mp_load(model_settings: ModelSettings):
28     """
29     This method is meant to run internally in the multiprocessing workers.
30     The loading needs to run synchronously, since the initializer argument
31     doesn't support coroutines.
32     """
33     # NOTE: The global `_mp_model` variable is shared with the `_mp_predict`
34     # method.
35     # This global variable should only be used within the inference
36     # multiprocessing workers.
37     global _mp_model
38 
39     model_class = model_settings.implementation
40     _mp_model = model_class(model_settings)  # type: ignore
41     return asyncio.run(_mp_model.load())
42 
43 
44 def _mp_predict(payload: InferenceRequest) -> InferenceResponse:
45     """
46     This method is meant to run internally in the multiprocessing workers.
47     The prediction needs to run synchronously, since multiprocessing
48     doesn't know how to serialise coroutines.
49     """
50     # NOTE: `_mp_model` is a global variable initialised in the `_mp_load`
51     # method.
52     # This global variable is only to be used within the inference worker
53     # context.
54     global _mp_model
55 
56     return asyncio.run(_mp_model.predict(payload))
57 
58 
59 class InferencePool:
60     """
61     The InferencePool class represents a pool of workers where we can run
62     inference on.
63 
64     Under the hood, it's responsible for managing a pool of multiprocessing
65     workers, where the model is loaded.
66     This approach lets MLServer work around the GIL to make sure that inference
67     can occur in parallel across multiple models or instances of a model.
68     """
69 
70     def __init__(self, model: MLModel):
71         parallel_workers = model.settings.parallel_workers
72         self._executor = ProcessPoolExecutor(
73             max_workers=parallel_workers,
74             initializer=_mp_load,
75             initargs=(model.settings,),
76         )
77 
78     async def predict(self, payload: InferenceRequest) -> InferenceResponse:
79         # What if we serialise payload?
80         loop = asyncio.get_running_loop()
81         return await loop.run_in_executor(self._executor, _mp_predict, payload)
82 
83     def __del__(self):
84         self._executor.shutdown(wait=True)
85 
86 
87 def parallel(f: Callable[[InferenceRequest], Coroutine[Any, Any, InferenceResponse]]):
88     """
89     Decorator to attach to model's methods so that they run in parallel.
90     By default, this will get attached to every model's "inference" method.
91 
92     NOTE: At the moment, this method only works with `predict()`.
93     """
94     # TODO: Extend to multiple methods
95     @wraps(f)
96     async def _inner(payload: InferenceRequest) -> InferenceResponse:
97         if not hasattr(f, "__self__"):
98             raise InvalidParallelMethod(f.__name__, reason="method is not bound")
99 
100         model = getattr(f, "__self__")
101         if not hasattr(model, _InferencePoolAttr):
102             raise InvalidParallelMethod(
103                 f.__name__, reason="inference pool has not been loaded"
104             )
105 
106         pool = getattr(model, _InferencePoolAttr)
107         return await pool.predict(payload)
108 
109     return _inner
110 
111 
112 async def load_inference_pool(model: MLModel):
113     if model.settings.parallel_workers == 0:
114         # When parallel workers is set to 0, disable parallel inference
115         return model
116 
117     pool = InferencePool(model)
118     setattr(model, _InferencePoolAttr, pool)
119 
120     # Decorate predict method
121     setattr(model, "predict", parallel(model.predict))
122 
123     return model
124 
125 
126 async def unload_inference_pool(model: MLModel):
127     pool = getattr(model, _InferencePoolAttr)
128     if not pool:
129         return
130 
131     pool.__del__()
132     delattr(model, _InferencePoolAttr)
133
```
--- 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/mlserver/parallel.py b/mlserver/parallel.py
--- a/mlserver/parallel.py
+++ b/mlserver/parallel.py
@@ -1,4 +1,5 @@
 import asyncio
+import multiprocessing as mp
 
 from functools import wraps
 from concurrent.futures import ProcessPoolExecutor
@@ -52,7 +53,6 @@
     # This global variable is only to be used within the inference worker
     # context.
     global _mp_model
-
     return asyncio.run(_mp_model.predict(payload))
 
 
@@ -69,8 +69,14 @@
 
     def __init__(self, model: MLModel):
         parallel_workers = model.settings.parallel_workers
+
+        # Use 'spawn' instead of 'fork' to ensure that models are loaded in a
+        # clean environment (e.g. to avoid issues like
+        # https://github.com/tensorflow/tensorflow/issues/8220)
+        ctx = mp.get_context("spawn")
         self._executor = ProcessPoolExecutor(
             max_workers=parallel_workers,
+            mp_context=ctx,
             initializer=_mp_load,
             initargs=(model.settings,),
         )

{"golden_diff": "diff --git a/mlserver/parallel.py b/mlserver/parallel.py\n--- a/mlserver/parallel.py\n+++ b/mlserver/parallel.py\n@@ -1,4 +1,5 @@\n import asyncio\n+import multiprocessing as mp\n \n from functools import wraps\n from concurrent.futures import ProcessPoolExecutor\n@@ -52,7 +53,6 @@\n     # This global variable is only to be used within the inference worker\n     # context.\n     global _mp_model\n-\n     return asyncio.run(_mp_model.predict(payload))\n \n \n@@ -69,8 +69,14 @@\n \n     def __init__(self, model: MLModel):\n         parallel_workers = model.settings.parallel_workers\n+\n+        # Use 'spawn' instead of 'fork' to ensure that models are loaded in a\n+        # clean environment (e.g. to avoid issues like\n+        # https://github.com/tensorflow/tensorflow/issues/8220)\n+        ctx = mp.get_context(\"spawn\")\n         self._executor = ProcessPoolExecutor(\n             max_workers=parallel_workers,\n+            mp_context=ctx,\n             initializer=_mp_load,\n             initargs=(model.settings,),\n         )\n", "issue": "Tempo example fails when parallel inference is enabled\nWhen parallel inference is enabled, the [outlier example using the Tempo runtime](https://tempo.readthedocs.io/en/latest/examples/outlier/README.html)seems to fail. In particular, it seems that either the `cifar10-service` or the `outlier` containers block the request path and never return a response.\n", "before_files": [{"content": "import asyncio\n\nfrom functools import wraps\nfrom concurrent.futures import ProcessPoolExecutor\nfrom typing import Any, Coroutine, Callable, Optional\n\nfrom .errors import MLServerError\nfrom .settings import ModelSettings\nfrom .model import MLModel\nfrom .types import InferenceRequest, InferenceResponse\n\n_InferencePoolAttr = \"__inference_pool__\"\n\n# NOTE: Workaround for mypy\n_mp_model: MLModel\n\n\nclass InvalidParallelMethod(MLServerError):\n    def __init__(self, method_name: str, reason: Optional[str] = None):\n        msg = f\"Method {method_name} can't be parallelised\"\n        if reason:\n            msg += f\": {reason}\"\n\n        super().__init__(msg)\n\n\ndef _mp_load(model_settings: ModelSettings):\n    \"\"\"\n    This method is meant to run internally in the multiprocessing workers.\n    The loading needs to run synchronously, since the initializer argument\n    doesn't support coroutines.\n    \"\"\"\n    # NOTE: The global `_mp_model` variable is shared with the `_mp_predict`\n    # method.\n    # This global variable should only be used within the inference\n    # multiprocessing workers.\n    global _mp_model\n\n    model_class = model_settings.implementation\n    _mp_model = model_class(model_settings)  # type: ignore\n    return asyncio.run(_mp_model.load())\n\n\ndef _mp_predict(payload: InferenceRequest) -> InferenceResponse:\n    \"\"\"\n    This method is meant to run internally in the multiprocessing workers.\n    The prediction needs to run synchronously, since multiprocessing\n    doesn't know how to serialise coroutines.\n    \"\"\"\n    # NOTE: `_mp_model` is a global variable initialised in the `_mp_load`\n    # method.\n    # This global variable is only to be used within the inference worker\n    # context.\n    global _mp_model\n\n    return asyncio.run(_mp_model.predict(payload))\n\n\nclass InferencePool:\n    \"\"\"\n    The InferencePool class represents a pool of workers where we can run\n    inference on.\n\n    Under the hood, it's responsible for managing a pool of multiprocessing\n    workers, where the model is loaded.\n    This approach lets MLServer work around the GIL to make sure that inference\n    can occur in parallel across multiple models or instances of a model.\n    \"\"\"\n\n    def __init__(self, model: MLModel):\n        parallel_workers = model.settings.parallel_workers\n        self._executor = ProcessPoolExecutor(\n            max_workers=parallel_workers,\n            initializer=_mp_load,\n            initargs=(model.settings,),\n        )\n\n    async def predict(self, payload: InferenceRequest) -> InferenceResponse:\n        # What if we serialise payload?\n        loop = asyncio.get_running_loop()\n        return await loop.run_in_executor(self._executor, _mp_predict, payload)\n\n    def __del__(self):\n        self._executor.shutdown(wait=True)\n\n\ndef parallel(f: Callable[[InferenceRequest], Coroutine[Any, Any, InferenceResponse]]):\n    \"\"\"\n    Decorator to attach to model's methods so that they run in parallel.\n    By default, this will get attached to every model's \"inference\" method.\n\n    NOTE: At the moment, this method only works with `predict()`.\n    \"\"\"\n    # TODO: Extend to multiple methods\n    @wraps(f)\n    async def _inner(payload: InferenceRequest) -> InferenceResponse:\n        if not hasattr(f, \"__self__\"):\n            raise InvalidParallelMethod(f.__name__, reason=\"method is not bound\")\n\n        model = getattr(f, \"__self__\")\n        if not hasattr(model, _InferencePoolAttr):\n            raise InvalidParallelMethod(\n                f.__name__, reason=\"inference pool has not been loaded\"\n            )\n\n        pool = getattr(model, _InferencePoolAttr)\n        return await pool.predict(payload)\n\n    return _inner\n\n\nasync def load_inference_pool(model: MLModel):\n    if model.settings.parallel_workers == 0:\n        # When parallel workers is set to 0, disable parallel inference\n        return model\n\n    pool = InferencePool(model)\n    setattr(model, _InferencePoolAttr, pool)\n\n    # Decorate predict method\n    setattr(model, \"predict\", parallel(model.predict))\n\n    return model\n\n\nasync def unload_inference_pool(model: MLModel):\n    pool = getattr(model, _InferencePoolAttr)\n    if not pool:\n        return\n\n    pool.__del__()\n    delattr(model, _InferencePoolAttr)\n", "path": "mlserver/parallel.py"}], "after_files": [{"content": "import asyncio\nimport multiprocessing as mp\n\nfrom functools import wraps\nfrom concurrent.futures import ProcessPoolExecutor\nfrom typing import Any, Coroutine, Callable, Optional\n\nfrom .errors import MLServerError\nfrom .settings import ModelSettings\nfrom .model import MLModel\nfrom .types import InferenceRequest, InferenceResponse\n\n_InferencePoolAttr = \"__inference_pool__\"\n\n# NOTE: Workaround for mypy\n_mp_model: MLModel\n\n\nclass InvalidParallelMethod(MLServerError):\n    def __init__(self, method_name: str, reason: Optional[str] = None):\n        msg = f\"Method {method_name} can't be parallelised\"\n        if reason:\n            msg += f\": {reason}\"\n\n        super().__init__(msg)\n\n\ndef _mp_load(model_settings: ModelSettings):\n    \"\"\"\n    This method is meant to run internally in the multiprocessing workers.\n    The loading needs to run synchronously, since the initializer argument\n    doesn't support coroutines.\n    \"\"\"\n    # NOTE: The global `_mp_model` variable is shared with the `_mp_predict`\n    # method.\n    # This global variable should only be used within the inference\n    # multiprocessing workers.\n    global _mp_model\n\n    model_class = model_settings.implementation\n    _mp_model = model_class(model_settings)  # type: ignore\n    return asyncio.run(_mp_model.load())\n\n\ndef _mp_predict(payload: InferenceRequest) -> InferenceResponse:\n    \"\"\"\n    This method is meant to run internally in the multiprocessing workers.\n    The prediction needs to run synchronously, since multiprocessing\n    doesn't know how to serialise coroutines.\n    \"\"\"\n    # NOTE: `_mp_model` is a global variable initialised in the `_mp_load`\n    # method.\n    # This global variable is only to be used within the inference worker\n    # context.\n    global _mp_model\n    return asyncio.run(_mp_model.predict(payload))\n\n\nclass InferencePool:\n    \"\"\"\n    The InferencePool class represents a pool of workers where we can run\n    inference on.\n\n    Under the hood, it's responsible for managing a pool of multiprocessing\n    workers, where the model is loaded.\n    This approach lets MLServer work around the GIL to make sure that inference\n    can occur in parallel across multiple models or instances of a model.\n    \"\"\"\n\n    def __init__(self, model: MLModel):\n        parallel_workers = model.settings.parallel_workers\n\n        # Use 'spawn' instead of 'fork' to ensure that models are loaded in a\n        # clean environment (e.g. to avoid issues like\n        # https://github.com/tensorflow/tensorflow/issues/8220)\n        ctx = mp.get_context(\"spawn\")\n        self._executor = ProcessPoolExecutor(\n            max_workers=parallel_workers,\n            mp_context=ctx,\n            initializer=_mp_load,\n            initargs=(model.settings,),\n        )\n\n    async def predict(self, payload: InferenceRequest) -> InferenceResponse:\n        # What if we serialise payload?\n        loop = asyncio.get_running_loop()\n        return await loop.run_in_executor(self._executor, _mp_predict, payload)\n\n    def __del__(self):\n        self._executor.shutdown(wait=True)\n\n\ndef parallel(f: Callable[[InferenceRequest], Coroutine[Any, Any, InferenceResponse]]):\n    \"\"\"\n    Decorator to attach to model's methods so that they run in parallel.\n    By default, this will get attached to every model's \"inference\" method.\n\n    NOTE: At the moment, this method only works with `predict()`.\n    \"\"\"\n    # TODO: Extend to multiple methods\n    @wraps(f)\n    async def _inner(payload: InferenceRequest) -> InferenceResponse:\n        if not hasattr(f, \"__self__\"):\n            raise InvalidParallelMethod(f.__name__, reason=\"method is not bound\")\n\n        model = getattr(f, \"__self__\")\n        if not hasattr(model, _InferencePoolAttr):\n            raise InvalidParallelMethod(\n                f.__name__, reason=\"inference pool has not been loaded\"\n            )\n\n        pool = getattr(model, _InferencePoolAttr)\n        return await pool.predict(payload)\n\n    return _inner\n\n\nasync def load_inference_pool(model: MLModel):\n    if model.settings.parallel_workers == 0:\n        # When parallel workers is set to 0, disable parallel inference\n        return model\n\n    pool = InferencePool(model)\n    setattr(model, _InferencePoolAttr, pool)\n\n    # Decorate predict method\n    setattr(model, \"predict\", parallel(model.predict))\n\n    return model\n\n\nasync def unload_inference_pool(model: MLModel):\n    pool = getattr(model, _InferencePoolAttr)\n    if not pool:\n        return\n\n    pool.__del__()\n    delattr(model, _InferencePoolAttr)\n", "path": "mlserver/parallel.py"}]}