uniworx/k8s-gitlab-borg
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 4 Wiki Activity

move pipe chunking downstream

Browse Source
This commit is contained in:
Gregor Kleen 2023-05-29 16:30:56 +02:00
parent f455d9f4d6
commit f41a21208e
1 changed files with 11 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
k8s_gitlab_borg/__main__.py
View File

@ -425,12 +425,10 @@ def main():
download_stream = download.stream(amt=1024 * 1024) download_stream = download.stream(amt=1024 * 1024)
try: try:
while chunk := next(download_stream, b""): while chunk := next(download_stream, b""):
logger.debug("Read chunk of %d bytes", len(chunk)) logger.debug("Read chunk of length %d", len(chunk))
offset += len(chunk) offset += len(chunk)
retries = 10 retries = 10
while pagesize_chunk := chunk[:PIPE_BUF]: yield chunk
yield pagesize_chunk
chunk = chunk[PIPE_BUF:]
else: else:
break break
except IncompleteRead as e: except IncompleteRead as e:
@ -482,17 +480,23 @@ def main():
pollc = 3 pollc = 3
logger.debug("First poll...") logger.debug("First poll...")
events = poll.poll() events = poll.poll()
logger.debug("Done, %d events", len(events)) logger.debug("Done, %d event(s)", len(events))
pipe_buffer = b""
while pollc > 0 and len(events) > 0: while pollc > 0 and len(events) > 0:
for rfd, event in events: for rfd, event in events:
if event & select.POLLOUT: if event & select.POLLOUT:
if rfd == proc.stdin.fileno(): if rfd == proc.stdin.fileno():
if chunk := next(download_stream, b""): if chunk := pipe_buffer[:PIPE_BUF]:
logger.debug( logger.debug(
"Writing chunk of length %d...", len(chunk) "Writing chunk of length %d...", len(chunk)
) )
proc.stdin.buffer.write(chunk) proc.stdin.buffer.write(chunk)
logger.debug("Done") logger.debug("Done")
pipe_buffer = pipe_buffer[PIPE_BUF:]
elif pipe_buffer := next(download_stream, b""):
logger.debug(
"Accepted chunk of length %d", len(chunk)
)
else: else:
proc.stdin.close() proc.stdin.close()
if event & select.POLLIN or event & select.POLLPRI: if event & select.POLLIN or event & select.POLLPRI:
@ -522,10 +526,9 @@ def main():
if pollc > 0: if pollc > 0:
logger.debug("Poll...") logger.debug("Poll...")
events = poll.poll() events = poll.poll()
logger.debug("Done, %d events", len(events)) logger.debug("Done, %d event(s)", len(events))
else: else:
logger.debug("Nothing left to poll") logger.debug("Nothing left to poll")
events = []
for handler in proc_logger.handlers: for handler in proc_logger.handlers:
handler.flush() handler.flush()