KittyPatch/kitty
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Handle blocking with unused input in librsync

Browse Source
This commit is contained in:
Kovid Goyal 2021-09-18 13:08:53 +05:30
parent cf517effb3
commit 77508bfe0d
No known key found for this signature in database
GPG Key ID: 06BC317B515ACE7C
3 changed files with 15 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
kittens/transfer/librsync.py
View File

@ -5,7 +5,7 @@
import os import os
from typing import Iterator from typing import Iterator
from .rsync import IO_BUFFER_SIZE, begin_create_signature, iter_job from .rsync import IO_BUFFER_SIZE, RsyncError, begin_create_signature, iter_job
def signature_of_file(path: str) -> Iterator[bytes]: def signature_of_file(path: str) -> Iterator[bytes]:
@ -15,7 +15,16 @@ def signature_of_file(path: str) -> Iterator[bytes]:
job = begin_create_signature(fsz) job = begin_create_signature(fsz)
f.seek(0) f.seek(0)
finished = False finished = False
prev_unused_input = b''
while not finished: while not finished:
input_data = f.read(IO_BUFFER_SIZE) input_data = f.read(IO_BUFFER_SIZE)
output, finished = iter_job(job, input_data) no_more_data = not input_data
if prev_unused_input:
input_data = prev_unused_input + input_data
prev_unused_input = b''
output, finished, sz_of_unused_input = iter_job(job, input_data)
if sz_of_unused_input > 0 and not finished:
if no_more_data:
raise RsyncError(f"{sz_of_unused_input} bytes of input data were not used")
prev_unused_input = input_data[-sz_of_unused_input:]
yield output yield output

6
kittens/transfer/rsync.c
View File

@ -69,14 +69,13 @@ iter_job(PyObject *self UNUSED, PyObject *args) {
size_t before = buffer.avail_out; size_t before = buffer.avail_out;
result = rs_job_iter(job, &buffer); result = rs_job_iter(job, &buffer);
output_size += before - buffer.avail_out; output_size += before - buffer.avail_out;
if (result == RS_DONE) break; if (result == RS_DONE || result == RS_BLOCKED) break;
if (buffer.avail_in) { if (buffer.avail_in) {
if (_PyBytes_Resize(&ans, PyBytes_GET_SIZE(ans) * 2) != 0) return NULL; if (_PyBytes_Resize(&ans, PyBytes_GET_SIZE(ans) * 2) != 0) return NULL;
buffer.avail_out = PyBytes_GET_SIZE(ans) - output_size; buffer.avail_out = PyBytes_GET_SIZE(ans) - output_size;
buffer.next_out = PyBytes_AS_STRING(ans) + output_size; buffer.next_out = PyBytes_AS_STRING(ans) + output_size;
continue; continue;
} }
if (result == RS_BLOCKED) break;
Py_DECREF(ans); Py_DECREF(ans);
PyErr_SetString(RsyncError, rs_strerror(result)); PyErr_SetString(RsyncError, rs_strerror(result));
return NULL; return NULL;
@ -84,7 +83,8 @@ iter_job(PyObject *self UNUSED, PyObject *args) {
if ((ssize_t)output_size != PyBytes_GET_SIZE(ans)) { if ((ssize_t)output_size != PyBytes_GET_SIZE(ans)) {
if (_PyBytes_Resize(&ans, output_size) != 0) return NULL; if (_PyBytes_Resize(&ans, output_size) != 0) return NULL;
} }
return Py_BuildValue("NO", ans, result == RS_DONE ? Py_True : Py_False); Py_ssize_t unused_input = buffer.avail_in;
return Py_BuildValue("NOn", ans, result == RS_DONE ? Py_True : Py_False, unused_input);
} }
static PyObject* static PyObject*

2
kittens/transfer/rsync.pyi
View File

@ -27,5 +27,5 @@ def make_hash_table(sig: SignatureCapsule) -> None:
pass pass
def iter_job(job_capsule: JobCapsule, input_data: bytes, eof: Optional[bool] = None) -> Tuple[bytes, bool]: def iter_job(job_capsule: JobCapsule, input_data: bytes, eof: Optional[bool] = None) -> Tuple[bytes, bool, int]:
pass pass