prompt: You will be provided with a partial code base and an issue statement explaining a problem to resolve.
  <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.
  
  </issue>
  <code>
  [start of mlserver/parallel.py]
  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 of mlserver/parallel.py]
  </code>
  
  I need you to solve this issue by generating a single patch file that I can apply directly to this repository using git apply. Please respond with a single patch file in the following format.
  <patch>
  diff --git a/file.py b/file.py
  --- a/file.py
  +++ b/file.py
  @@ -1,27 +1,35 @@
   def euclidean(a, b):
  -    while b:
  -        a, b = b, a % b
  -    return a
  +    if b == 0:
  +        return a
  +    return euclidean(b, a % b)
   
   
   def bresenham(x0, y0, x1, y1):
       points = []
       dx = abs(x1 - x0)
       dy = abs(y1 - y0)
  -    sx = 1 if x0 < x1 else -1
  -    sy = 1 if y0 < y1 else -1
  -    err = dx - dy
  +    x, y = x0, y0
  +    sx = -1 if x0 > x1 else 1
  +    sy = -1 if y0 > y1 else 1
   
  -    while True:
  -        points.append((x0, y0))
  -        if x0 == x1 and y0 == y1:
  -            break
  -        e2 = 2 * err
  -        if e2 > -dy:
  -            err -= dy
  -            x0 += sx
  -        if e2 < dx:
  -            err += dx
  -            y0 += sy
  +    if dx > dy:
  +        err = dx / 2.0
  +        while x != x1:
  +            points.append((x, y))
  +            err -= dy
  +            if err < 0:
  +                y += sy
  +                err += dx
  +            x += sx
  +    else:
  +        err = dy / 2.0
  +        while y != y1:
  +            points.append((x, y))
  +            err -= dx
  +            if err < 0:
  +                x += sx
  +                err += dy
  +            y += sy
  +
  +    points.append((x, y))
       return points
  
  </patch>
golden_diff: 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,),
           )
  
verification_info: {"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"}]}