We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Reduce memory usage of the pipeline
Author: @bmbouter (bmbouter)


Redmine Issue: 9635, https://pulp.plan.io/issues/9635

---

## Motivation

It would be nice if users could specify a desired maximum amount of RAM to be used during sync. For example, a user can say I only want 1500 MB of RAM to be used max.

## What is already in place

The stages pipeline restricts memory usage by only allowing 1000 declarative content objects between each stage (so for 8-9 stages that's 8000-9000 declarative content objects. This happens [here](https://github.com/pulp/pulpcore/blob/main/pulpcore/plugin/stages/api.py#L217).

Interestingly the docstring says this defaults to 100, but it seems to actually be 1000!

Also the stages perform batching, so they will only taking in a limited number of items (the batch size). That happens [with minsize](https://github.com/pulp/pulpcore/blob/main/pulpcore/plugin/stages/api.py#L84).

## Why this isn't enough

These are count-based mechnisms and don't correspond to actual MB or GB of memory used. Some content units vary a lot in how much memory each DeclarativeContent objects take up.

Another lesser problem is that it doesn't help plugin writers restrict their usage of memory in FirstStage.

## Idea

Add a new param called `max_mb` to base Remote, which defaults to None. If specified, the user will be specifying the desired maximum MB used by process syncing.

Have the queues between the stages, and the bather implementation, both check the total memory the current process is using and asyncio.sleep() polling until it goes down. This should keep the maximum amount used by all objects roughly to that number.

## Details

Introduce a new `MBSizeQueue` which is a wrapper around `asyncio.Queue` used today. It will have the same `put()` call, only wait if the amount of memory in use is greater than the remote is configured for.

Then introduce the same memory checking feature in the batcher. I'm not completely sure this second part is needed though.

We have to be very careful not to deadlock with this feature. For example, we have to account for the base case where even a single item is larger than the memory desired. Repos in pulp_rpm have had a single unit use more than 1.2G if I remember right, so if someone was syncing with 800 MB and we weren't careful to allow that unit to still flow through the pipeline we'd deadlock.....




--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `pulpcore/plugin/stages/api.py`
Content:
```
1 import asyncio
2 import logging
3 
4 from gettext import gettext as _
5 
6 from django.conf import settings
7 
8 from .profiler import ProfilingQueue
9 
10 
11 log = logging.getLogger(__name__)
12 
13 
14 class Stage:
15     """
16     The base class for all Stages API stages.
17 
18     To make a stage, inherit from this class and implement :meth:`run` on the subclass.
19     """
20 
21     def __init__(self):
22         self._in_q = None
23         self._out_q = None
24 
25     def _connect(self, in_q, out_q):
26         """
27         Connect to queues within a pipeline.
28 
29         Args:
30             in_q (asyncio.Queue): The stage input queue.
31             out_q (asyncio.Queue): The stage output queue.
32         """
33         self._in_q = in_q
34         self._out_q = out_q
35 
36     async def __call__(self):
37         """
38         This coroutine makes the stage callable.
39 
40         It calls :meth:`run` and signals the next stage that its work is finished.
41         """
42         log.debug(_("%(name)s - begin."), {"name": self})
43         await self.run()
44         await self._out_q.put(None)
45         log.debug(_("%(name)s - put end-marker."), {"name": self})
46 
47     async def run(self):
48         """
49         The coroutine that is run as part of this stage.
50 
51         Returns:
52             The coroutine that runs this stage.
53 
54         """
55         raise NotImplementedError(_("A plugin writer must implement this method"))
56 
57     async def items(self):
58         """
59         Asynchronous iterator yielding items of :class:`DeclarativeContent` from `self._in_q`.
60 
61         The iterator will get instances of :class:`DeclarativeContent` one by one as they get
62         available.
63 
64         Yields:
65             An instance of :class:`DeclarativeContent`
66 
67         Examples:
68             Used in stages to get d_content instances one by one from `self._in_q`::
69 
70                 class MyStage(Stage):
71                     async def run(self):
72                         async for d_content in self.items():
73                             # process declarative content
74                             await self.put(d_content)
75 
76         """
77         while True:
78             content = await self._in_q.get()
79             if content is None:
80                 break
81             log.debug("%(name)s - next: %(content)s.", {"name": self, "content": content})
82             yield content
83 
84     async def batches(self, minsize=500):
85         """
86         Asynchronous iterator yielding batches of :class:`DeclarativeContent` from `self._in_q`.
87 
88         The iterator will try to get as many instances of
89         :class:`DeclarativeContent` as possible without blocking, but
90         at least `minsize` instances.
91 
92         Args:
93             minsize (int): The minimum batch size to yield (unless it is the final batch)
94 
95         Yields:
96             A list of :class:`DeclarativeContent` instances
97 
98         Examples:
99             Used in stages to get large chunks of d_content instances from `self._in_q`::
100 
101                 class MyStage(Stage):
102                     async def run(self):
103                         async for batch in self.batches():
104                             for d_content in batch:
105                                 # process declarative content
106                                 await self.put(d_content)
107 
108         """
109         batch = []
110         shutdown = False
111         no_block = False
112         thaw_queue_event = asyncio.Event()
113 
114         def add_to_batch(content):
115             nonlocal batch
116             nonlocal shutdown
117             nonlocal no_block
118             nonlocal thaw_queue_event
119 
120             if content is None:
121                 shutdown = True
122                 log.debug(_("%(name)s - shutdown."), {"name": self})
123             else:
124                 if not content.does_batch:
125                     no_block = True
126                 content._thaw_queue_event = thaw_queue_event
127                 batch.append(content)
128 
129         get_listener = asyncio.ensure_future(self._in_q.get())
130         thaw_event_listener = asyncio.ensure_future(thaw_queue_event.wait())
131         while not shutdown:
132             done, pending = await asyncio.wait(
133                 [thaw_event_listener, get_listener], return_when=asyncio.FIRST_COMPLETED
134             )
135             if thaw_event_listener in done:
136                 thaw_event_listener = asyncio.ensure_future(thaw_queue_event.wait())
137                 no_block = True
138             if get_listener in done:
139                 content = await get_listener
140                 add_to_batch(content)
141                 get_listener = asyncio.ensure_future(self._in_q.get())
142             while not shutdown:
143                 try:
144                     content = self._in_q.get_nowait()
145                 except asyncio.QueueEmpty:
146                     break
147                 else:
148                     add_to_batch(content)
149 
150             if batch and (len(batch) >= minsize or shutdown or no_block):
151                 log.debug(
152                     _("%(name)s - next batch[%(length)d]."), {"name": self, "length": len(batch)}
153                 )
154                 for content in batch:
155                     content._thaw_queue_event = None
156                 thaw_queue_event.clear()
157                 yield batch
158                 batch = []
159                 no_block = False
160         thaw_event_listener.cancel()
161         get_listener.cancel()
162 
163     async def put(self, item):
164         """
165         Coroutine to pass items to the next stage.
166 
167         Args:
168             item: A handled instance of :class:`pulpcore.plugin.stages.DeclarativeContent`
169 
170         Raises:
171             ValueError: When `item` is None.
172         """
173         if item is None:
174             raise ValueError(_("(None) not permitted."))
175         await self._out_q.put(item)
176         log.debug("{name} - put: {content}".format(name=self, content=item))
177 
178     def __str__(self):
179         return "[{id}] {name}".format(id=id(self), name=self.__class__.__name__)
180 
181 
182 async def create_pipeline(stages, maxsize=1000):
183     """
184     A coroutine that builds a Stages API linear pipeline from the list `stages` and runs it.
185 
186     Each stage is an instance of a class derived from :class:`pulpcore.plugin.stages.Stage` that
187     implements the :meth:`run` coroutine. This coroutine reads asyncromously either from the
188     `items()` iterator or the `batches()` iterator and outputs the items with `put()`. Here is an
189     example of the simplest stage that only passes data::
190 
191         class MyStage(Stage):
192             async def run(self):
193                 async for d_content in self.items():  # Fetch items from the previous stage
194                     await self.put(d_content)  # Hand them over to the next stage
195 
196     Args:
197         stages (list of coroutines): A list of Stages API compatible coroutines.
198         maxsize (int): The maximum amount of items a queue between two stages should hold. Optional
199             and defaults to 100.
200 
201     Returns:
202         A single coroutine that can be used to run, wait, or cancel the entire pipeline with.
203     Raises:
204         ValueError: When a stage instance is specified more than once.
205     """
206     futures = []
207     history = set()
208     in_q = None
209     for i, stage in enumerate(stages):
210         if stage in history:
211             raise ValueError(_("Each stage instance must be unique."))
212         history.add(stage)
213         if i < len(stages) - 1:
214             if settings.PROFILE_STAGES_API:
215                 out_q = ProfilingQueue.make_and_record_queue(stages[i + 1], i + 1, maxsize)
216             else:
217                 out_q = asyncio.Queue(maxsize=maxsize)
218         else:
219             out_q = None
220         stage._connect(in_q, out_q)
221         futures.append(asyncio.ensure_future(stage()))
222         in_q = out_q
223 
224     try:
225         await asyncio.gather(*futures)
226     except Exception:
227         # One of the stages raised an exception, cancel all stages...
228         pending = []
229         for task in futures:
230             if not task.done():
231                 task.cancel()
232                 pending.append(task)
233         # ...and run until all Exceptions show up
234         if pending:
235             await asyncio.wait(pending, timeout=60)
236         raise
237 
238 
239 class EndStage(Stage):
240     """
241     A Stages API stage that drains incoming items and does nothing with the items. This is
242     required at the end of all pipelines.
243 
244     Without this stage, the `maxsize` of the last stage's `_out_q` could fill up and block the
245     entire pipeline.
246     """
247 
248     async def __call__(self):
249         """
250         This method drains items from the last queue and drops them.
251 
252         Importantly it does not try to put items into the nonexistent next queue.
253         """
254         # We overwrite __call__ here to avoid trying to put None in `self._out_q`.
255         async for _ in self.items():  # noqa
256             pass
257
```
--- 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/pulpcore/plugin/stages/api.py b/pulpcore/plugin/stages/api.py
--- a/pulpcore/plugin/stages/api.py
+++ b/pulpcore/plugin/stages/api.py
@@ -179,12 +179,12 @@
         return "[{id}] {name}".format(id=id(self), name=self.__class__.__name__)
 
 
-async def create_pipeline(stages, maxsize=1000):
+async def create_pipeline(stages, maxsize=1):
     """
     A coroutine that builds a Stages API linear pipeline from the list `stages` and runs it.
 
     Each stage is an instance of a class derived from :class:`pulpcore.plugin.stages.Stage` that
-    implements the :meth:`run` coroutine. This coroutine reads asyncromously either from the
+    implements the :meth:`run` coroutine. This coroutine reads asynchronously either from the
     `items()` iterator or the `batches()` iterator and outputs the items with `put()`. Here is an
     example of the simplest stage that only passes data::
 
@@ -196,7 +196,7 @@
     Args:
         stages (list of coroutines): A list of Stages API compatible coroutines.
         maxsize (int): The maximum amount of items a queue between two stages should hold. Optional
-            and defaults to 100.
+            and defaults to 1.
 
     Returns:
         A single coroutine that can be used to run, wait, or cancel the entire pipeline with.

{"golden_diff": "diff --git a/pulpcore/plugin/stages/api.py b/pulpcore/plugin/stages/api.py\n--- a/pulpcore/plugin/stages/api.py\n+++ b/pulpcore/plugin/stages/api.py\n@@ -179,12 +179,12 @@\n         return \"[{id}] {name}\".format(id=id(self), name=self.__class__.__name__)\n \n \n-async def create_pipeline(stages, maxsize=1000):\n+async def create_pipeline(stages, maxsize=1):\n     \"\"\"\n     A coroutine that builds a Stages API linear pipeline from the list `stages` and runs it.\n \n     Each stage is an instance of a class derived from :class:`pulpcore.plugin.stages.Stage` that\n-    implements the :meth:`run` coroutine. This coroutine reads asyncromously either from the\n+    implements the :meth:`run` coroutine. This coroutine reads asynchronously either from the\n     `items()` iterator or the `batches()` iterator and outputs the items with `put()`. Here is an\n     example of the simplest stage that only passes data::\n \n@@ -196,7 +196,7 @@\n     Args:\n         stages (list of coroutines): A list of Stages API compatible coroutines.\n         maxsize (int): The maximum amount of items a queue between two stages should hold. Optional\n-            and defaults to 100.\n+            and defaults to 1.\n \n     Returns:\n         A single coroutine that can be used to run, wait, or cancel the entire pipeline with.\n", "issue": "Reduce memory usage of the pipeline\nAuthor: @bmbouter (bmbouter)\n\n\nRedmine Issue: 9635, https://pulp.plan.io/issues/9635\n\n---\n\n## Motivation\r\n\r\nIt would be nice if users could specify a desired maximum amount of RAM to be used during sync. For example, a user can say I only want 1500 MB of RAM to be used max.\r\n\r\n## What is already in place\r\n\r\nThe stages pipeline restricts memory usage by only allowing 1000 declarative content objects between each stage (so for 8-9 stages that's 8000-9000 declarative content objects. This happens [here](https://github.com/pulp/pulpcore/blob/main/pulpcore/plugin/stages/api.py#L217).\r\n\r\nInterestingly the docstring says this defaults to 100, but it seems to actually be 1000!\r\n\r\nAlso the stages perform batching, so they will only taking in a limited number of items (the batch size). That happens [with minsize](https://github.com/pulp/pulpcore/blob/main/pulpcore/plugin/stages/api.py#L84).\r\n\r\n## Why this isn't enough\r\n\r\nThese are count-based mechnisms and don't correspond to actual MB or GB of memory used. Some content units vary a lot in how much memory each DeclarativeContent objects take up.\r\n\r\nAnother lesser problem is that it doesn't help plugin writers restrict their usage of memory in FirstStage.\r\n\r\n## Idea\r\n\r\nAdd a new param called `max_mb` to base Remote, which defaults to None. If specified, the user will be specifying the desired maximum MB used by process syncing.\r\n\r\nHave the queues between the stages, and the bather implementation, both check the total memory the current process is using and asyncio.sleep() polling until it goes down. This should keep the maximum amount used by all objects roughly to that number.\r\n\r\n## Details\r\n\r\nIntroduce a new `MBSizeQueue` which is a wrapper around `asyncio.Queue` used today. It will have the same `put()` call, only wait if the amount of memory in use is greater than the remote is configured for.\r\n\r\nThen introduce the same memory checking feature in the batcher. I'm not completely sure this second part is needed though.\r\n\r\nWe have to be very careful not to deadlock with this feature. For example, we have to account for the base case where even a single item is larger than the memory desired. Repos in pulp_rpm have had a single unit use more than 1.2G if I remember right, so if someone was syncing with 800 MB and we weren't careful to allow that unit to still flow through the pipeline we'd deadlock.....\n\n\n\n", "before_files": [{"content": "import asyncio\nimport logging\n\nfrom gettext import gettext as _\n\nfrom django.conf import settings\n\nfrom .profiler import ProfilingQueue\n\n\nlog = logging.getLogger(__name__)\n\n\nclass Stage:\n    \"\"\"\n    The base class for all Stages API stages.\n\n    To make a stage, inherit from this class and implement :meth:`run` on the subclass.\n    \"\"\"\n\n    def __init__(self):\n        self._in_q = None\n        self._out_q = None\n\n    def _connect(self, in_q, out_q):\n        \"\"\"\n        Connect to queues within a pipeline.\n\n        Args:\n            in_q (asyncio.Queue): The stage input queue.\n            out_q (asyncio.Queue): The stage output queue.\n        \"\"\"\n        self._in_q = in_q\n        self._out_q = out_q\n\n    async def __call__(self):\n        \"\"\"\n        This coroutine makes the stage callable.\n\n        It calls :meth:`run` and signals the next stage that its work is finished.\n        \"\"\"\n        log.debug(_(\"%(name)s - begin.\"), {\"name\": self})\n        await self.run()\n        await self._out_q.put(None)\n        log.debug(_(\"%(name)s - put end-marker.\"), {\"name\": self})\n\n    async def run(self):\n        \"\"\"\n        The coroutine that is run as part of this stage.\n\n        Returns:\n            The coroutine that runs this stage.\n\n        \"\"\"\n        raise NotImplementedError(_(\"A plugin writer must implement this method\"))\n\n    async def items(self):\n        \"\"\"\n        Asynchronous iterator yielding items of :class:`DeclarativeContent` from `self._in_q`.\n\n        The iterator will get instances of :class:`DeclarativeContent` one by one as they get\n        available.\n\n        Yields:\n            An instance of :class:`DeclarativeContent`\n\n        Examples:\n            Used in stages to get d_content instances one by one from `self._in_q`::\n\n                class MyStage(Stage):\n                    async def run(self):\n                        async for d_content in self.items():\n                            # process declarative content\n                            await self.put(d_content)\n\n        \"\"\"\n        while True:\n            content = await self._in_q.get()\n            if content is None:\n                break\n            log.debug(\"%(name)s - next: %(content)s.\", {\"name\": self, \"content\": content})\n            yield content\n\n    async def batches(self, minsize=500):\n        \"\"\"\n        Asynchronous iterator yielding batches of :class:`DeclarativeContent` from `self._in_q`.\n\n        The iterator will try to get as many instances of\n        :class:`DeclarativeContent` as possible without blocking, but\n        at least `minsize` instances.\n\n        Args:\n            minsize (int): The minimum batch size to yield (unless it is the final batch)\n\n        Yields:\n            A list of :class:`DeclarativeContent` instances\n\n        Examples:\n            Used in stages to get large chunks of d_content instances from `self._in_q`::\n\n                class MyStage(Stage):\n                    async def run(self):\n                        async for batch in self.batches():\n                            for d_content in batch:\n                                # process declarative content\n                                await self.put(d_content)\n\n        \"\"\"\n        batch = []\n        shutdown = False\n        no_block = False\n        thaw_queue_event = asyncio.Event()\n\n        def add_to_batch(content):\n            nonlocal batch\n            nonlocal shutdown\n            nonlocal no_block\n            nonlocal thaw_queue_event\n\n            if content is None:\n                shutdown = True\n                log.debug(_(\"%(name)s - shutdown.\"), {\"name\": self})\n            else:\n                if not content.does_batch:\n                    no_block = True\n                content._thaw_queue_event = thaw_queue_event\n                batch.append(content)\n\n        get_listener = asyncio.ensure_future(self._in_q.get())\n        thaw_event_listener = asyncio.ensure_future(thaw_queue_event.wait())\n        while not shutdown:\n            done, pending = await asyncio.wait(\n                [thaw_event_listener, get_listener], return_when=asyncio.FIRST_COMPLETED\n            )\n            if thaw_event_listener in done:\n                thaw_event_listener = asyncio.ensure_future(thaw_queue_event.wait())\n                no_block = True\n            if get_listener in done:\n                content = await get_listener\n                add_to_batch(content)\n                get_listener = asyncio.ensure_future(self._in_q.get())\n            while not shutdown:\n                try:\n                    content = self._in_q.get_nowait()\n                except asyncio.QueueEmpty:\n                    break\n                else:\n                    add_to_batch(content)\n\n            if batch and (len(batch) >= minsize or shutdown or no_block):\n                log.debug(\n                    _(\"%(name)s - next batch[%(length)d].\"), {\"name\": self, \"length\": len(batch)}\n                )\n                for content in batch:\n                    content._thaw_queue_event = None\n                thaw_queue_event.clear()\n                yield batch\n                batch = []\n                no_block = False\n        thaw_event_listener.cancel()\n        get_listener.cancel()\n\n    async def put(self, item):\n        \"\"\"\n        Coroutine to pass items to the next stage.\n\n        Args:\n            item: A handled instance of :class:`pulpcore.plugin.stages.DeclarativeContent`\n\n        Raises:\n            ValueError: When `item` is None.\n        \"\"\"\n        if item is None:\n            raise ValueError(_(\"(None) not permitted.\"))\n        await self._out_q.put(item)\n        log.debug(\"{name} - put: {content}\".format(name=self, content=item))\n\n    def __str__(self):\n        return \"[{id}] {name}\".format(id=id(self), name=self.__class__.__name__)\n\n\nasync def create_pipeline(stages, maxsize=1000):\n    \"\"\"\n    A coroutine that builds a Stages API linear pipeline from the list `stages` and runs it.\n\n    Each stage is an instance of a class derived from :class:`pulpcore.plugin.stages.Stage` that\n    implements the :meth:`run` coroutine. This coroutine reads asyncromously either from the\n    `items()` iterator or the `batches()` iterator and outputs the items with `put()`. Here is an\n    example of the simplest stage that only passes data::\n\n        class MyStage(Stage):\n            async def run(self):\n                async for d_content in self.items():  # Fetch items from the previous stage\n                    await self.put(d_content)  # Hand them over to the next stage\n\n    Args:\n        stages (list of coroutines): A list of Stages API compatible coroutines.\n        maxsize (int): The maximum amount of items a queue between two stages should hold. Optional\n            and defaults to 100.\n\n    Returns:\n        A single coroutine that can be used to run, wait, or cancel the entire pipeline with.\n    Raises:\n        ValueError: When a stage instance is specified more than once.\n    \"\"\"\n    futures = []\n    history = set()\n    in_q = None\n    for i, stage in enumerate(stages):\n        if stage in history:\n            raise ValueError(_(\"Each stage instance must be unique.\"))\n        history.add(stage)\n        if i < len(stages) - 1:\n            if settings.PROFILE_STAGES_API:\n                out_q = ProfilingQueue.make_and_record_queue(stages[i + 1], i + 1, maxsize)\n            else:\n                out_q = asyncio.Queue(maxsize=maxsize)\n        else:\n            out_q = None\n        stage._connect(in_q, out_q)\n        futures.append(asyncio.ensure_future(stage()))\n        in_q = out_q\n\n    try:\n        await asyncio.gather(*futures)\n    except Exception:\n        # One of the stages raised an exception, cancel all stages...\n        pending = []\n        for task in futures:\n            if not task.done():\n                task.cancel()\n                pending.append(task)\n        # ...and run until all Exceptions show up\n        if pending:\n            await asyncio.wait(pending, timeout=60)\n        raise\n\n\nclass EndStage(Stage):\n    \"\"\"\n    A Stages API stage that drains incoming items and does nothing with the items. This is\n    required at the end of all pipelines.\n\n    Without this stage, the `maxsize` of the last stage's `_out_q` could fill up and block the\n    entire pipeline.\n    \"\"\"\n\n    async def __call__(self):\n        \"\"\"\n        This method drains items from the last queue and drops them.\n\n        Importantly it does not try to put items into the nonexistent next queue.\n        \"\"\"\n        # We overwrite __call__ here to avoid trying to put None in `self._out_q`.\n        async for _ in self.items():  # noqa\n            pass\n", "path": "pulpcore/plugin/stages/api.py"}], "after_files": [{"content": "import asyncio\nimport logging\n\nfrom gettext import gettext as _\n\nfrom django.conf import settings\n\nfrom .profiler import ProfilingQueue\n\n\nlog = logging.getLogger(__name__)\n\n\nclass Stage:\n    \"\"\"\n    The base class for all Stages API stages.\n\n    To make a stage, inherit from this class and implement :meth:`run` on the subclass.\n    \"\"\"\n\n    def __init__(self):\n        self._in_q = None\n        self._out_q = None\n\n    def _connect(self, in_q, out_q):\n        \"\"\"\n        Connect to queues within a pipeline.\n\n        Args:\n            in_q (asyncio.Queue): The stage input queue.\n            out_q (asyncio.Queue): The stage output queue.\n        \"\"\"\n        self._in_q = in_q\n        self._out_q = out_q\n\n    async def __call__(self):\n        \"\"\"\n        This coroutine makes the stage callable.\n\n        It calls :meth:`run` and signals the next stage that its work is finished.\n        \"\"\"\n        log.debug(_(\"%(name)s - begin.\"), {\"name\": self})\n        await self.run()\n        await self._out_q.put(None)\n        log.debug(_(\"%(name)s - put end-marker.\"), {\"name\": self})\n\n    async def run(self):\n        \"\"\"\n        The coroutine that is run as part of this stage.\n\n        Returns:\n            The coroutine that runs this stage.\n\n        \"\"\"\n        raise NotImplementedError(_(\"A plugin writer must implement this method\"))\n\n    async def items(self):\n        \"\"\"\n        Asynchronous iterator yielding items of :class:`DeclarativeContent` from `self._in_q`.\n\n        The iterator will get instances of :class:`DeclarativeContent` one by one as they get\n        available.\n\n        Yields:\n            An instance of :class:`DeclarativeContent`\n\n        Examples:\n            Used in stages to get d_content instances one by one from `self._in_q`::\n\n                class MyStage(Stage):\n                    async def run(self):\n                        async for d_content in self.items():\n                            # process declarative content\n                            await self.put(d_content)\n\n        \"\"\"\n        while True:\n            content = await self._in_q.get()\n            if content is None:\n                break\n            log.debug(\"%(name)s - next: %(content)s.\", {\"name\": self, \"content\": content})\n            yield content\n\n    async def batches(self, minsize=500):\n        \"\"\"\n        Asynchronous iterator yielding batches of :class:`DeclarativeContent` from `self._in_q`.\n\n        The iterator will try to get as many instances of\n        :class:`DeclarativeContent` as possible without blocking, but\n        at least `minsize` instances.\n\n        Args:\n            minsize (int): The minimum batch size to yield (unless it is the final batch)\n\n        Yields:\n            A list of :class:`DeclarativeContent` instances\n\n        Examples:\n            Used in stages to get large chunks of d_content instances from `self._in_q`::\n\n                class MyStage(Stage):\n                    async def run(self):\n                        async for batch in self.batches():\n                            for d_content in batch:\n                                # process declarative content\n                                await self.put(d_content)\n\n        \"\"\"\n        batch = []\n        shutdown = False\n        no_block = False\n        thaw_queue_event = asyncio.Event()\n\n        def add_to_batch(content):\n            nonlocal batch\n            nonlocal shutdown\n            nonlocal no_block\n            nonlocal thaw_queue_event\n\n            if content is None:\n                shutdown = True\n                log.debug(_(\"%(name)s - shutdown.\"), {\"name\": self})\n            else:\n                if not content.does_batch:\n                    no_block = True\n                content._thaw_queue_event = thaw_queue_event\n                batch.append(content)\n\n        get_listener = asyncio.ensure_future(self._in_q.get())\n        thaw_event_listener = asyncio.ensure_future(thaw_queue_event.wait())\n        while not shutdown:\n            done, pending = await asyncio.wait(\n                [thaw_event_listener, get_listener], return_when=asyncio.FIRST_COMPLETED\n            )\n            if thaw_event_listener in done:\n                thaw_event_listener = asyncio.ensure_future(thaw_queue_event.wait())\n                no_block = True\n            if get_listener in done:\n                content = await get_listener\n                add_to_batch(content)\n                get_listener = asyncio.ensure_future(self._in_q.get())\n            while not shutdown:\n                try:\n                    content = self._in_q.get_nowait()\n                except asyncio.QueueEmpty:\n                    break\n                else:\n                    add_to_batch(content)\n\n            if batch and (len(batch) >= minsize or shutdown or no_block):\n                log.debug(\n                    _(\"%(name)s - next batch[%(length)d].\"), {\"name\": self, \"length\": len(batch)}\n                )\n                for content in batch:\n                    content._thaw_queue_event = None\n                thaw_queue_event.clear()\n                yield batch\n                batch = []\n                no_block = False\n        thaw_event_listener.cancel()\n        get_listener.cancel()\n\n    async def put(self, item):\n        \"\"\"\n        Coroutine to pass items to the next stage.\n\n        Args:\n            item: A handled instance of :class:`pulpcore.plugin.stages.DeclarativeContent`\n\n        Raises:\n            ValueError: When `item` is None.\n        \"\"\"\n        if item is None:\n            raise ValueError(_(\"(None) not permitted.\"))\n        await self._out_q.put(item)\n        log.debug(\"{name} - put: {content}\".format(name=self, content=item))\n\n    def __str__(self):\n        return \"[{id}] {name}\".format(id=id(self), name=self.__class__.__name__)\n\n\nasync def create_pipeline(stages, maxsize=1):\n    \"\"\"\n    A coroutine that builds a Stages API linear pipeline from the list `stages` and runs it.\n\n    Each stage is an instance of a class derived from :class:`pulpcore.plugin.stages.Stage` that\n    implements the :meth:`run` coroutine. This coroutine reads asynchronously either from the\n    `items()` iterator or the `batches()` iterator and outputs the items with `put()`. Here is an\n    example of the simplest stage that only passes data::\n\n        class MyStage(Stage):\n            async def run(self):\n                async for d_content in self.items():  # Fetch items from the previous stage\n                    await self.put(d_content)  # Hand them over to the next stage\n\n    Args:\n        stages (list of coroutines): A list of Stages API compatible coroutines.\n        maxsize (int): The maximum amount of items a queue between two stages should hold. Optional\n            and defaults to 1.\n\n    Returns:\n        A single coroutine that can be used to run, wait, or cancel the entire pipeline with.\n    Raises:\n        ValueError: When a stage instance is specified more than once.\n    \"\"\"\n    futures = []\n    history = set()\n    in_q = None\n    for i, stage in enumerate(stages):\n        if stage in history:\n            raise ValueError(_(\"Each stage instance must be unique.\"))\n        history.add(stage)\n        if i < len(stages) - 1:\n            if settings.PROFILE_STAGES_API:\n                out_q = ProfilingQueue.make_and_record_queue(stages[i + 1], i + 1, maxsize)\n            else:\n                out_q = asyncio.Queue(maxsize=maxsize)\n        else:\n            out_q = None\n        stage._connect(in_q, out_q)\n        futures.append(asyncio.ensure_future(stage()))\n        in_q = out_q\n\n    try:\n        await asyncio.gather(*futures)\n    except Exception:\n        # One of the stages raised an exception, cancel all stages...\n        pending = []\n        for task in futures:\n            if not task.done():\n                task.cancel()\n                pending.append(task)\n        # ...and run until all Exceptions show up\n        if pending:\n            await asyncio.wait(pending, timeout=60)\n        raise\n\n\nclass EndStage(Stage):\n    \"\"\"\n    A Stages API stage that drains incoming items and does nothing with the items. This is\n    required at the end of all pipelines.\n\n    Without this stage, the `maxsize` of the last stage's `_out_q` could fill up and block the\n    entire pipeline.\n    \"\"\"\n\n    async def __call__(self):\n        \"\"\"\n        This method drains items from the last queue and drops them.\n\n        Importantly it does not try to put items into the nonexistent next queue.\n        \"\"\"\n        # We overwrite __call__ here to avoid trying to put None in `self._out_q`.\n        async for _ in self.items():  # noqa\n            pass\n", "path": "pulpcore/plugin/stages/api.py"}]}