Issue #8295 ยป sync_disk_usage_change.patch
| pulpcore/plugin/stages/api.py | ||
|---|---|---|
|
log.debug("%(name)s - next: %(content)s.", {"name": self, "content": content})
|
||
|
yield content
|
||
|
async def batches(self, minsize=500):
|
||
|
# the default value for the parameter minsize should be changed
|
||
|
async def batches(self, minsize=50):
|
||
|
"""
|
||
|
Asynchronous iterator yielding batches of :class:`DeclarativeContent` from `self._in_q`.
|
||
| pulpcore/plugin/stages/artifact_stages.py | ||
|---|---|---|
|
import os
|
||
|
import asyncio
|
||
|
from collections import defaultdict
|
||
|
from gettext import gettext as _
|
||
| ... | ... | |
|
The coroutine for this stage.
|
||
|
"""
|
||
|
async for batch in self.batches():
|
||
|
# TODO: artifacts are saved only here when the minimal size of the batch is reached;
|
||
|
# otherwise, all the artifacts are saved only when the queue is filled with 500 file
|
||
|
# units (temporary files)
|
||
|
da_to_save = []
|
||
|
tmp_files = []
|
||
|
for d_content in batch:
|
||
|
for d_artifact in d_content.d_artifacts:
|
||
|
if d_artifact.artifact._state.adding and not d_artifact.deferred_download:
|
||
|
d_artifact.artifact.file = str(d_artifact.artifact.file)
|
||
|
tmp_files.append(str(d_artifact.artifact.file))
|
||
|
da_to_save.append(d_artifact)
|
||
|
if da_to_save:
|
||
| ... | ... | |
|
for d_content in batch:
|
||
|
await self.put(d_content)
|
||
|
# the caller should do the clean-up since temporary files are initialized with the
|
||
|
# parameter delete=False
|
||
|
for path in tmp_files:
|
||
|
os.unlink(path)
|
||
|
class RemoteArtifactSaver(Stage):
|
||
|
"""
|
||
.