We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Support lists of `BaseModel` for outputs
The following model fails on Replicate.com with a cryptic `Can't pickle <class 'predict.Output'>: it's not the same object as predict.Output` error:

```python
class Output(BaseModel):
    foo: str
    bar: str

class Predictor(BasePredictor):
    def predict(self) -> List[Output]:
        return [Output(foo="foo", bar="bar")]
```

This is [documented deep in Cog's Python layer](https://github.com/replicate/cog/blob/main/python/cog/server/runner.py#L238).

We should support this type of output.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `python/cog/server/runner.py`
Content:
```
1 import multiprocessing
2 import types
3 from enum import Enum
4 from multiprocessing.connection import Connection
5 from typing import Any, Dict, List, Optional
6 
7 from pydantic import BaseModel
8 
9 from ..predictor import load_config, load_predictor
10 from .log_capture import capture_log
11 
12 
13 class PredictionRunner:
14     PROCESSING_DONE = 1
15 
16     class OutputType(Enum):
17         NOT_STARTED = 0
18         SINGLE = 1
19         GENERATOR = 2
20 
21     def __init__(self) -> None:
22         self.logs_pipe_reader, self.logs_pipe_writer = multiprocessing.Pipe(
23             duplex=False
24         )
25         (
26             self.prediction_input_pipe_reader,
27             self.prediction_input_pipe_writer,
28         ) = multiprocessing.Pipe(duplex=False)
29         self.predictor_pipe_reader, self.predictor_pipe_writer = multiprocessing.Pipe(
30             duplex=False
31         )
32         self.error_pipe_reader, self.error_pipe_writer = multiprocessing.Pipe(
33             duplex=False
34         )
35         self.done_pipe_reader, self.done_pipe_writer = multiprocessing.Pipe(
36             duplex=False
37         )
38 
39     def setup(self) -> None:
40         """
41         Sets up the predictor in a subprocess. Blocks until the predictor has
42         finished setup. To start a prediction after setup call `run()`.
43         """
44         # `multiprocessing.get_context("spawn")` returns the same API as
45         # `multiprocessing`, but will use the spawn method when creating any
46         # subprocess. Using the spawn method for the predictor subprocess is
47         # useful for compatibility with CUDA, which cannot run in a process
48         # that gets forked. If we can guarantee that all initialization happens
49         # within the subprocess, we could probably get away with using fork
50         # here instead.
51         self.predictor_process = multiprocessing.get_context("spawn").Process(
52             target=self._start_predictor_process
53         )
54 
55         self._is_processing = True
56         self.predictor_process.start()
57 
58         # poll with an infinite timeout to avoid burning resources in the loop
59         while self.done_pipe_reader.poll(timeout=None) and self.is_processing():
60             pass
61 
62     def _start_predictor_process(self) -> None:
63         config = load_config()
64         self.predictor = load_predictor(config)
65         self.predictor.setup()
66 
67         # tell the main process we've finished setup
68         self.done_pipe_writer.send(self.PROCESSING_DONE)
69 
70         while True:
71             try:
72                 prediction_input = self.prediction_input_pipe_reader.recv()
73                 self._run_prediction(prediction_input)
74             except EOFError:
75                 continue
76 
77     def run(self, **prediction_input: Dict[str, Any]) -> None:
78         """
79         Starts running a prediction in the predictor subprocess, using the
80         inputs provided in `prediction_input`.
81 
82         The subprocess will send prediction output and logs to pipes as soon as
83         they're available. You can check if the pipes have any data using
84         `has_output_waiting()` and `has_logs_waiting()`. You can read data from
85         the pipes using `read_output()` and `read_logs()`.
86 
87         Use `is_processing()` to check whether more data is expected in the
88         pipe for prediction output.
89         """
90         # We're starting processing!
91         self._is_processing = True
92 
93         # We don't know whether or not we've got a generator (progressive
94         # output) until we start getting output from the model
95         self._is_output_generator = self.OutputType.NOT_STARTED
96 
97         # We haven't encountered an error yet
98         self._error = None
99 
100         # Send prediction input through the pipe to the predictor subprocess
101         self.prediction_input_pipe_writer.send(prediction_input)
102 
103     def is_processing(self) -> bool:
104         """
105         Returns True if the subprocess running the prediction is still
106         processing.
107         """
108         if self.done_pipe_reader.poll():
109             try:
110                 if self.done_pipe_reader.recv() == self.PROCESSING_DONE:
111                     self._is_processing = False
112             except EOFError:
113                 pass
114 
115         return self._is_processing
116 
117     def has_output_waiting(self) -> bool:
118         return self.predictor_pipe_reader.poll()
119 
120     def read_output(self) -> List[Any]:
121         if self._is_output_generator is self.OutputType.NOT_STARTED:
122             return []
123 
124         output = []
125         while self.has_output_waiting():
126             try:
127                 output.append(self.predictor_pipe_reader.recv())
128             except EOFError:
129                 break
130         return output
131 
132     def has_logs_waiting(self) -> bool:
133         return self.logs_pipe_reader.poll()
134 
135     def read_logs(self) -> List[str]:
136         logs = []
137         while self.has_logs_waiting():
138             try:
139                 logs.append(self.logs_pipe_reader.recv())
140             except EOFError:
141                 break
142         return logs
143 
144     def is_output_generator(self) -> Optional[bool]:
145         """
146         Returns `True` if the output is a generator, `False` if it's not, and
147         `None` if we don't know yet.
148         """
149         if self._is_output_generator is self.OutputType.NOT_STARTED:
150             if self.has_output_waiting():
151                 # if there's output waiting use the first one to set whether
152                 # we've got a generator, with a safety check
153                 self._is_output_generator = self.predictor_pipe_reader.recv()
154                 assert isinstance(self._is_output_generator, self.OutputType)
155 
156         if self._is_output_generator is self.OutputType.NOT_STARTED:
157             return None
158         elif self._is_output_generator is self.OutputType.SINGLE:
159             return False
160         elif self._is_output_generator is self.OutputType.GENERATOR:
161             return True
162 
163     def _run_prediction(self, prediction_input: Dict[str, Any]) -> None:
164         """
165         Sends a boolean first, to indicate whether the output is a generator.
166         After that it sends the output(s).
167 
168         If the predictor raises an exception it'll send it to the error pipe
169         writer and then exit.
170 
171         When the prediction is finished it'll send a token to the done pipe.
172         """
173         # Empty all the pipes before we start sending more messages to them
174         drain_pipe(self.logs_pipe_reader)
175         drain_pipe(self.predictor_pipe_reader)
176         drain_pipe(self.error_pipe_reader)
177         drain_pipe(self.done_pipe_reader)
178 
179         with capture_log(self.logs_pipe_writer):
180             try:
181                 output = self.predictor.predict(**prediction_input)
182 
183                 if isinstance(output, types.GeneratorType):
184                     self.predictor_pipe_writer.send(self.OutputType.GENERATOR)
185                     while True:
186                         try:
187                             self.predictor_pipe_writer.send(
188                                 next(make_pickleable(output))
189                             )
190                         except StopIteration:
191                             break
192                 else:
193                     self.predictor_pipe_writer.send(self.OutputType.SINGLE)
194                     self.predictor_pipe_writer.send(make_pickleable(output))
195             except Exception as e:
196                 self.error_pipe_writer.send(e)
197 
198         self.done_pipe_writer.send(self.PROCESSING_DONE)
199 
200     def error(self) -> Optional[str]:
201         """
202         Returns the error encountered by the predictor, if one exists.
203         """
204         if self._error is None and self.error_pipe_reader.poll():
205             try:
206                 self._error = self.error_pipe_reader.recv()
207             except EOFError:
208                 # I don't know how this is reachable ¯\_(ツ)_/¯
209                 pass
210 
211         return self._error
212 
213 
214 def drain_pipe(pipe_reader: Connection) -> None:
215     """
216     Reads all available messages from a pipe and discards them. This serves to
217     clear the pipe for future usage.
218     """
219     while pipe_reader.poll():
220         try:
221             pipe_reader.recv()
222         except EOFError:
223             break
224 
225 
226 def make_pickleable(obj: Any) -> Any:
227     """
228     Returns a version of `obj` which can be pickled and therefore sent through
229     the pipe to the main process.
230 
231     If the predictor uses a custom output like:
232 
233         class Output(BaseModel):
234             text: str
235 
236     then the output can't be sent through the pipe because:
237 
238     > Can't pickle <class 'predict.Output'>: it's not the same object as
239     > 'predict.Output'
240 
241     The way we're getting around this here will only work for singly-nested
242     outputs. If there's a complex object inside a complex object, it's likely
243     to fall over.
244 
245     A better fix for this would be to work out why the pickling process is
246     getting a different class when loading `Output`, so the pickling Just
247     Works.
248     """
249     if isinstance(obj, BaseModel):
250         return obj.dict(exclude_unset=True)
251     else:
252         return obj
253
```
--- 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/python/cog/server/runner.py b/python/cog/server/runner.py
--- a/python/cog/server/runner.py
+++ b/python/cog/server/runner.py
@@ -248,5 +248,7 @@
     """
     if isinstance(obj, BaseModel):
         return obj.dict(exclude_unset=True)
+    elif isinstance(obj, List):
+        return [make_pickleable(item) for item in obj]
     else:
         return obj

{"golden_diff": "diff --git a/python/cog/server/runner.py b/python/cog/server/runner.py\n--- a/python/cog/server/runner.py\n+++ b/python/cog/server/runner.py\n@@ -248,5 +248,7 @@\n     \"\"\"\n     if isinstance(obj, BaseModel):\n         return obj.dict(exclude_unset=True)\n+    elif isinstance(obj, List):\n+        return [make_pickleable(item) for item in obj]\n     else:\n         return obj\n", "issue": "Support lists of `BaseModel` for outputs\nThe following model fails on Replicate.com with a cryptic `Can't pickle <class 'predict.Output'>: it's not the same object as predict.Output` error:\r\n\r\n```python\r\nclass Output(BaseModel):\r\n    foo: str\r\n    bar: str\r\n\r\nclass Predictor(BasePredictor):\r\n    def predict(self) -> List[Output]:\r\n        return [Output(foo=\"foo\", bar=\"bar\")]\r\n```\r\n\r\nThis is [documented deep in Cog's Python layer](https://github.com/replicate/cog/blob/main/python/cog/server/runner.py#L238).\r\n\r\nWe should support this type of output.\n", "before_files": [{"content": "import multiprocessing\nimport types\nfrom enum import Enum\nfrom multiprocessing.connection import Connection\nfrom typing import Any, Dict, List, Optional\n\nfrom pydantic import BaseModel\n\nfrom ..predictor import load_config, load_predictor\nfrom .log_capture import capture_log\n\n\nclass PredictionRunner:\n    PROCESSING_DONE = 1\n\n    class OutputType(Enum):\n        NOT_STARTED = 0\n        SINGLE = 1\n        GENERATOR = 2\n\n    def __init__(self) -> None:\n        self.logs_pipe_reader, self.logs_pipe_writer = multiprocessing.Pipe(\n            duplex=False\n        )\n        (\n            self.prediction_input_pipe_reader,\n            self.prediction_input_pipe_writer,\n        ) = multiprocessing.Pipe(duplex=False)\n        self.predictor_pipe_reader, self.predictor_pipe_writer = multiprocessing.Pipe(\n            duplex=False\n        )\n        self.error_pipe_reader, self.error_pipe_writer = multiprocessing.Pipe(\n            duplex=False\n        )\n        self.done_pipe_reader, self.done_pipe_writer = multiprocessing.Pipe(\n            duplex=False\n        )\n\n    def setup(self) -> None:\n        \"\"\"\n        Sets up the predictor in a subprocess. Blocks until the predictor has\n        finished setup. To start a prediction after setup call `run()`.\n        \"\"\"\n        # `multiprocessing.get_context(\"spawn\")` returns the same API as\n        # `multiprocessing`, but will use the spawn method when creating any\n        # subprocess. Using the spawn method for the predictor subprocess is\n        # useful for compatibility with CUDA, which cannot run in a process\n        # that gets forked. If we can guarantee that all initialization happens\n        # within the subprocess, we could probably get away with using fork\n        # here instead.\n        self.predictor_process = multiprocessing.get_context(\"spawn\").Process(\n            target=self._start_predictor_process\n        )\n\n        self._is_processing = True\n        self.predictor_process.start()\n\n        # poll with an infinite timeout to avoid burning resources in the loop\n        while self.done_pipe_reader.poll(timeout=None) and self.is_processing():\n            pass\n\n    def _start_predictor_process(self) -> None:\n        config = load_config()\n        self.predictor = load_predictor(config)\n        self.predictor.setup()\n\n        # tell the main process we've finished setup\n        self.done_pipe_writer.send(self.PROCESSING_DONE)\n\n        while True:\n            try:\n                prediction_input = self.prediction_input_pipe_reader.recv()\n                self._run_prediction(prediction_input)\n            except EOFError:\n                continue\n\n    def run(self, **prediction_input: Dict[str, Any]) -> None:\n        \"\"\"\n        Starts running a prediction in the predictor subprocess, using the\n        inputs provided in `prediction_input`.\n\n        The subprocess will send prediction output and logs to pipes as soon as\n        they're available. You can check if the pipes have any data using\n        `has_output_waiting()` and `has_logs_waiting()`. You can read data from\n        the pipes using `read_output()` and `read_logs()`.\n\n        Use `is_processing()` to check whether more data is expected in the\n        pipe for prediction output.\n        \"\"\"\n        # We're starting processing!\n        self._is_processing = True\n\n        # We don't know whether or not we've got a generator (progressive\n        # output) until we start getting output from the model\n        self._is_output_generator = self.OutputType.NOT_STARTED\n\n        # We haven't encountered an error yet\n        self._error = None\n\n        # Send prediction input through the pipe to the predictor subprocess\n        self.prediction_input_pipe_writer.send(prediction_input)\n\n    def is_processing(self) -> bool:\n        \"\"\"\n        Returns True if the subprocess running the prediction is still\n        processing.\n        \"\"\"\n        if self.done_pipe_reader.poll():\n            try:\n                if self.done_pipe_reader.recv() == self.PROCESSING_DONE:\n                    self._is_processing = False\n            except EOFError:\n                pass\n\n        return self._is_processing\n\n    def has_output_waiting(self) -> bool:\n        return self.predictor_pipe_reader.poll()\n\n    def read_output(self) -> List[Any]:\n        if self._is_output_generator is self.OutputType.NOT_STARTED:\n            return []\n\n        output = []\n        while self.has_output_waiting():\n            try:\n                output.append(self.predictor_pipe_reader.recv())\n            except EOFError:\n                break\n        return output\n\n    def has_logs_waiting(self) -> bool:\n        return self.logs_pipe_reader.poll()\n\n    def read_logs(self) -> List[str]:\n        logs = []\n        while self.has_logs_waiting():\n            try:\n                logs.append(self.logs_pipe_reader.recv())\n            except EOFError:\n                break\n        return logs\n\n    def is_output_generator(self) -> Optional[bool]:\n        \"\"\"\n        Returns `True` if the output is a generator, `False` if it's not, and\n        `None` if we don't know yet.\n        \"\"\"\n        if self._is_output_generator is self.OutputType.NOT_STARTED:\n            if self.has_output_waiting():\n                # if there's output waiting use the first one to set whether\n                # we've got a generator, with a safety check\n                self._is_output_generator = self.predictor_pipe_reader.recv()\n                assert isinstance(self._is_output_generator, self.OutputType)\n\n        if self._is_output_generator is self.OutputType.NOT_STARTED:\n            return None\n        elif self._is_output_generator is self.OutputType.SINGLE:\n            return False\n        elif self._is_output_generator is self.OutputType.GENERATOR:\n            return True\n\n    def _run_prediction(self, prediction_input: Dict[str, Any]) -> None:\n        \"\"\"\n        Sends a boolean first, to indicate whether the output is a generator.\n        After that it sends the output(s).\n\n        If the predictor raises an exception it'll send it to the error pipe\n        writer and then exit.\n\n        When the prediction is finished it'll send a token to the done pipe.\n        \"\"\"\n        # Empty all the pipes before we start sending more messages to them\n        drain_pipe(self.logs_pipe_reader)\n        drain_pipe(self.predictor_pipe_reader)\n        drain_pipe(self.error_pipe_reader)\n        drain_pipe(self.done_pipe_reader)\n\n        with capture_log(self.logs_pipe_writer):\n            try:\n                output = self.predictor.predict(**prediction_input)\n\n                if isinstance(output, types.GeneratorType):\n                    self.predictor_pipe_writer.send(self.OutputType.GENERATOR)\n                    while True:\n                        try:\n                            self.predictor_pipe_writer.send(\n                                next(make_pickleable(output))\n                            )\n                        except StopIteration:\n                            break\n                else:\n                    self.predictor_pipe_writer.send(self.OutputType.SINGLE)\n                    self.predictor_pipe_writer.send(make_pickleable(output))\n            except Exception as e:\n                self.error_pipe_writer.send(e)\n\n        self.done_pipe_writer.send(self.PROCESSING_DONE)\n\n    def error(self) -> Optional[str]:\n        \"\"\"\n        Returns the error encountered by the predictor, if one exists.\n        \"\"\"\n        if self._error is None and self.error_pipe_reader.poll():\n            try:\n                self._error = self.error_pipe_reader.recv()\n            except EOFError:\n                # I don't know how this is reachable \u00af\\_(\u30c4)_/\u00af\n                pass\n\n        return self._error\n\n\ndef drain_pipe(pipe_reader: Connection) -> None:\n    \"\"\"\n    Reads all available messages from a pipe and discards them. This serves to\n    clear the pipe for future usage.\n    \"\"\"\n    while pipe_reader.poll():\n        try:\n            pipe_reader.recv()\n        except EOFError:\n            break\n\n\ndef make_pickleable(obj: Any) -> Any:\n    \"\"\"\n    Returns a version of `obj` which can be pickled and therefore sent through\n    the pipe to the main process.\n\n    If the predictor uses a custom output like:\n\n        class Output(BaseModel):\n            text: str\n\n    then the output can't be sent through the pipe because:\n\n    > Can't pickle <class 'predict.Output'>: it's not the same object as\n    > 'predict.Output'\n\n    The way we're getting around this here will only work for singly-nested\n    outputs. If there's a complex object inside a complex object, it's likely\n    to fall over.\n\n    A better fix for this would be to work out why the pickling process is\n    getting a different class when loading `Output`, so the pickling Just\n    Works.\n    \"\"\"\n    if isinstance(obj, BaseModel):\n        return obj.dict(exclude_unset=True)\n    else:\n        return obj\n", "path": "python/cog/server/runner.py"}], "after_files": [{"content": "import multiprocessing\nimport types\nfrom enum import Enum\nfrom multiprocessing.connection import Connection\nfrom typing import Any, Dict, List, Optional\n\nfrom pydantic import BaseModel\n\nfrom ..predictor import load_config, load_predictor\nfrom .log_capture import capture_log\n\n\nclass PredictionRunner:\n    PROCESSING_DONE = 1\n\n    class OutputType(Enum):\n        NOT_STARTED = 0\n        SINGLE = 1\n        GENERATOR = 2\n\n    def __init__(self) -> None:\n        self.logs_pipe_reader, self.logs_pipe_writer = multiprocessing.Pipe(\n            duplex=False\n        )\n        (\n            self.prediction_input_pipe_reader,\n            self.prediction_input_pipe_writer,\n        ) = multiprocessing.Pipe(duplex=False)\n        self.predictor_pipe_reader, self.predictor_pipe_writer = multiprocessing.Pipe(\n            duplex=False\n        )\n        self.error_pipe_reader, self.error_pipe_writer = multiprocessing.Pipe(\n            duplex=False\n        )\n        self.done_pipe_reader, self.done_pipe_writer = multiprocessing.Pipe(\n            duplex=False\n        )\n\n    def setup(self) -> None:\n        \"\"\"\n        Sets up the predictor in a subprocess. Blocks until the predictor has\n        finished setup. To start a prediction after setup call `run()`.\n        \"\"\"\n        # `multiprocessing.get_context(\"spawn\")` returns the same API as\n        # `multiprocessing`, but will use the spawn method when creating any\n        # subprocess. Using the spawn method for the predictor subprocess is\n        # useful for compatibility with CUDA, which cannot run in a process\n        # that gets forked. If we can guarantee that all initialization happens\n        # within the subprocess, we could probably get away with using fork\n        # here instead.\n        self.predictor_process = multiprocessing.get_context(\"spawn\").Process(\n            target=self._start_predictor_process\n        )\n\n        self._is_processing = True\n        self.predictor_process.start()\n\n        # poll with an infinite timeout to avoid burning resources in the loop\n        while self.done_pipe_reader.poll(timeout=None) and self.is_processing():\n            pass\n\n    def _start_predictor_process(self) -> None:\n        config = load_config()\n        self.predictor = load_predictor(config)\n        self.predictor.setup()\n\n        # tell the main process we've finished setup\n        self.done_pipe_writer.send(self.PROCESSING_DONE)\n\n        while True:\n            try:\n                prediction_input = self.prediction_input_pipe_reader.recv()\n                self._run_prediction(prediction_input)\n            except EOFError:\n                continue\n\n    def run(self, **prediction_input: Dict[str, Any]) -> None:\n        \"\"\"\n        Starts running a prediction in the predictor subprocess, using the\n        inputs provided in `prediction_input`.\n\n        The subprocess will send prediction output and logs to pipes as soon as\n        they're available. You can check if the pipes have any data using\n        `has_output_waiting()` and `has_logs_waiting()`. You can read data from\n        the pipes using `read_output()` and `read_logs()`.\n\n        Use `is_processing()` to check whether more data is expected in the\n        pipe for prediction output.\n        \"\"\"\n        # We're starting processing!\n        self._is_processing = True\n\n        # We don't know whether or not we've got a generator (progressive\n        # output) until we start getting output from the model\n        self._is_output_generator = self.OutputType.NOT_STARTED\n\n        # We haven't encountered an error yet\n        self._error = None\n\n        # Send prediction input through the pipe to the predictor subprocess\n        self.prediction_input_pipe_writer.send(prediction_input)\n\n    def is_processing(self) -> bool:\n        \"\"\"\n        Returns True if the subprocess running the prediction is still\n        processing.\n        \"\"\"\n        if self.done_pipe_reader.poll():\n            try:\n                if self.done_pipe_reader.recv() == self.PROCESSING_DONE:\n                    self._is_processing = False\n            except EOFError:\n                pass\n\n        return self._is_processing\n\n    def has_output_waiting(self) -> bool:\n        return self.predictor_pipe_reader.poll()\n\n    def read_output(self) -> List[Any]:\n        if self._is_output_generator is self.OutputType.NOT_STARTED:\n            return []\n\n        output = []\n        while self.has_output_waiting():\n            try:\n                output.append(self.predictor_pipe_reader.recv())\n            except EOFError:\n                break\n        return output\n\n    def has_logs_waiting(self) -> bool:\n        return self.logs_pipe_reader.poll()\n\n    def read_logs(self) -> List[str]:\n        logs = []\n        while self.has_logs_waiting():\n            try:\n                logs.append(self.logs_pipe_reader.recv())\n            except EOFError:\n                break\n        return logs\n\n    def is_output_generator(self) -> Optional[bool]:\n        \"\"\"\n        Returns `True` if the output is a generator, `False` if it's not, and\n        `None` if we don't know yet.\n        \"\"\"\n        if self._is_output_generator is self.OutputType.NOT_STARTED:\n            if self.has_output_waiting():\n                # if there's output waiting use the first one to set whether\n                # we've got a generator, with a safety check\n                self._is_output_generator = self.predictor_pipe_reader.recv()\n                assert isinstance(self._is_output_generator, self.OutputType)\n\n        if self._is_output_generator is self.OutputType.NOT_STARTED:\n            return None\n        elif self._is_output_generator is self.OutputType.SINGLE:\n            return False\n        elif self._is_output_generator is self.OutputType.GENERATOR:\n            return True\n\n    def _run_prediction(self, prediction_input: Dict[str, Any]) -> None:\n        \"\"\"\n        Sends a boolean first, to indicate whether the output is a generator.\n        After that it sends the output(s).\n\n        If the predictor raises an exception it'll send it to the error pipe\n        writer and then exit.\n\n        When the prediction is finished it'll send a token to the done pipe.\n        \"\"\"\n        # Empty all the pipes before we start sending more messages to them\n        drain_pipe(self.logs_pipe_reader)\n        drain_pipe(self.predictor_pipe_reader)\n        drain_pipe(self.error_pipe_reader)\n        drain_pipe(self.done_pipe_reader)\n\n        with capture_log(self.logs_pipe_writer):\n            try:\n                output = self.predictor.predict(**prediction_input)\n\n                if isinstance(output, types.GeneratorType):\n                    self.predictor_pipe_writer.send(self.OutputType.GENERATOR)\n                    while True:\n                        try:\n                            self.predictor_pipe_writer.send(\n                                next(make_pickleable(output))\n                            )\n                        except StopIteration:\n                            break\n                else:\n                    self.predictor_pipe_writer.send(self.OutputType.SINGLE)\n                    self.predictor_pipe_writer.send(make_pickleable(output))\n            except Exception as e:\n                self.error_pipe_writer.send(e)\n\n        self.done_pipe_writer.send(self.PROCESSING_DONE)\n\n    def error(self) -> Optional[str]:\n        \"\"\"\n        Returns the error encountered by the predictor, if one exists.\n        \"\"\"\n        if self._error is None and self.error_pipe_reader.poll():\n            try:\n                self._error = self.error_pipe_reader.recv()\n            except EOFError:\n                # I don't know how this is reachable \u00af\\_(\u30c4)_/\u00af\n                pass\n\n        return self._error\n\n\ndef drain_pipe(pipe_reader: Connection) -> None:\n    \"\"\"\n    Reads all available messages from a pipe and discards them. This serves to\n    clear the pipe for future usage.\n    \"\"\"\n    while pipe_reader.poll():\n        try:\n            pipe_reader.recv()\n        except EOFError:\n            break\n\n\ndef make_pickleable(obj: Any) -> Any:\n    \"\"\"\n    Returns a version of `obj` which can be pickled and therefore sent through\n    the pipe to the main process.\n\n    If the predictor uses a custom output like:\n\n        class Output(BaseModel):\n            text: str\n\n    then the output can't be sent through the pipe because:\n\n    > Can't pickle <class 'predict.Output'>: it's not the same object as\n    > 'predict.Output'\n\n    The way we're getting around this here will only work for singly-nested\n    outputs. If there's a complex object inside a complex object, it's likely\n    to fall over.\n\n    A better fix for this would be to work out why the pickling process is\n    getting a different class when loading `Output`, so the pickling Just\n    Works.\n    \"\"\"\n    if isinstance(obj, BaseModel):\n        return obj.dict(exclude_unset=True)\n    elif isinstance(obj, List):\n        return [make_pickleable(item) for item in obj]\n    else:\n        return obj\n", "path": "python/cog/server/runner.py"}]}