On Mon, May 22, 2017 at 12:57:12PM +0200, Michal Privoznik wrote: s/OKI/OLI/, I believe

Yet again, we need a custom wrapper over virStreamRecvFlags because our generator is not capable of generating it. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- generator.py | 1 + libvirt-override-virStream.py | 18 ++++++++++++++++++ libvirt-override.c | 39 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 58 insertions(+) diff --git a/generator.py b/generator.py index ca1df35..0e07fc8 100755 --- a/generator.py +++ b/generator.py @@ -545,6 +545,7 @@ skip_function = ( 'virStreamSend', # overridden in libvirt-override-virStream.py 'virStreamRecvHole', # overridden in libvirt-override-virStream.py 'virStreamSendHole', # overridden in libvirt-override-virStream.py + 'virStreamRecvFlags', # overridden in libvirt-override-virStream.py 'virConnectUnregisterCloseCallback', # overridden in virConnect.py 'virConnectRegisterCloseCallback', # overridden in virConnect.py diff --git a/libvirt-override-virStream.py b/libvirt-override-virStream.py index 62c1328..66d2bf6 100644 --- a/libvirt-override-virStream.py +++ b/libvirt-override-virStream.py @@ -146,3 +146,21 @@ ret = libvirtmod.virStreamSendHole(self._o, length, flags) if ret == -1: raise libvirtError('virStreamSendHole() failed') return ret + + def recvFlags(self, nbytes, flags = 0): + """Reads a series of bytes from the stream. This method may + block the calling application for an arbitrary amount + of time. This is just like recv except it has flags + argument. + + Errors are not guaranteed to be reported synchronously + with the call, but may instead be delayed until a + subsequent call. + + On success, the received data is returned. On failure, an + exception is raised. If the stream is a NONBLOCK stream and + the request would block, integer -2 is returned. + """ + ret = libvirtmod.virStreamRecvFlags(self._o, nbytes, flags) + if ret is None: raise libvirtError ('virStreamRecvFlags() failed') + return ret diff --git a/libvirt-override.c b/libvirt-override.c index b5e11ed..0310c35 100644 --- a/libvirt-override.c +++ b/libvirt-override.c @@ -9519,6 +9519,44 @@ libvirt_virStreamSendHole(PyObject *self ATTRIBUTE_UNUSED, return libvirt_intWrap(ret); } + +static PyObject * +libvirt_virStreamRecvFlags(PyObject *self ATTRIBUTE_UNUSED, + PyObject *args) +{ + PyObject *pyobj_stream; + PyObject *rv; + virStreamPtr stream; + char *buf = NULL; + size_t nbytes; + unsigned int flags; + int ret; + + if (!PyArg_ParseTuple(args, (char *) "OkI:virStreamRecvFlags", + &pyobj_stream, &nbytes, &flags)) + return NULL; + + stream = PyvirStream_Get(pyobj_stream); + + if (VIR_ALLOC_N(buf, nbytes + 1) < 0) + return PyErr_NoMemory(); + + LIBVIRT_BEGIN_ALLOW_THREADS; + ret = virStreamRecvFlags(stream, buf, nbytes, flags); + LIBVIRT_END_ALLOW_THREADS; + + buf[ret > -1 ? ret : 0] = '\0'; + DEBUG("StreamRecvFlags ret=%d strlen=%d\n", ret, (int) strlen(buf)); + + if (ret == -2 || ret == -3) + return libvirt_intWrap(ret); + if (ret < 0) + return VIR_PY_NONE; + rv = libvirt_charPtrSizeWrap((char *) buf, (Py_ssize_t) ret); + VIR_FREE(buf); + return rv; +} + #endif /* LIBVIR_CHECK_VERSION(3, 4, 0) */ @@ -9749,6 +9787,7 @@ static PyMethodDef libvirtMethods[] = { #if LIBVIR_CHECK_VERSION(3, 4, 0) {(char *) "virStreamRecvHole", libvirt_virStreamRecvHole, METH_VARARGS, NULL}, {(char *) "virStreamSendHole", libvirt_virStreamSendHole, METH_VARARGS, NULL}, + {(char *) "virStreamRecvFlags", libvirt_virStreamRecvFlags, METH_VARARGS, NULL}, #endif /* LIBVIR_CHECK_VERSION(3, 4, 0) */ {NULL, NULL, 0, NULL} }; -- 2.13.0

Yet again, our parser is not capable of generating proper wrapper. To be fair, this one wold be really tough anyway. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- generator.py | 2 + libvirt-override-virStream.py | 117 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 119 insertions(+) diff --git a/generator.py b/generator.py index 0e07fc8..93d1dc3 100755 --- a/generator.py +++ b/generator.py @@ -546,6 +546,8 @@ skip_function = ( 'virStreamRecvHole', # overridden in libvirt-override-virStream.py 'virStreamSendHole', # overridden in libvirt-override-virStream.py 'virStreamRecvFlags', # overridden in libvirt-override-virStream.py + 'virStreamSparseRecvAll', # overridden in libvirt-override-virStream.py + 'virStreamSparseSendAll', # overridden in libvirt-override-virStream.py 'virConnectUnregisterCloseCallback', # overridden in virConnect.py 'virConnectRegisterCloseCallback', # overridden in virConnect.py diff --git a/libvirt-override-virStream.py b/libvirt-override-virStream.py index 66d2bf6..568bd9f 100644 --- a/libvirt-override-virStream.py +++ b/libvirt-override-virStream.py @@ -164,3 +164,120 @@ ret = libvirtmod.virStreamRecvFlags(self._o, nbytes, flags) if ret is None: raise libvirtError ('virStreamRecvFlags() failed') return ret + + def sparseRecvAll(self, handler, holeHandler, opaque): + """Receive the entire data stream, sending the data to + the requested data sink handler and calling the skip + holeHandler to generate holes for sparse stream targets. + This is simply a convenient alternative to recvFlags, for + apps that do blocking-I/O. + + Hypothetical callbacks can look like this: + + def handler(stream, # virStream instance + buf, # string containing received data + opaque): # extra data passed to sparseRecvAll as opaque + fd = opaque + return os.write(fd, buf) + + def holeHandler(stream, # virStream instance + length, # number of bytes to skip + opaque): # extra data passed to sparseRecvAll as opaque + fd = opaque + cur = os.lseek(fd, length, os.SEEK_CUR) + return os.ftruncate(fd, cur) # take this extra step to + # actually allocate the hole + """ + while True: + want = 64 * 1024 + got = self.recvFlags(want, VIR_STREAM_RECV_STOP_AT_HOLE) + if got == -2: + raise libvirtError("cannot use sparseRecvAll with " + "nonblocking stream") + if got == -3: + length = self.recvHole() + if (length is None): + self.abort() + raise RuntimeError("recvHole handler failed") + try: + ret = holeHandler(self, length, opaque) + except: + self.abort() + continue + + if len(got) == 0: + break + + try: + ret = handler(self, got, opaque) + if type(ret) is int and ret < 0: + raise RuntimeError("recvAll handler returned %d" % ret) + except Exception: + e = sys.exc_info()[1] + try: + self.abort() + except: + pass + raise e + + def sparseSendAll(self, handler, holeHandler, skipHandler, opaque): + """Send the entire data stream, reading the data from the + requested data source. This is simply a convenient + alternative to virStreamSend, for apps that do + blocking-I/O. + + Hypothetical callbacks can look like this: + + def handler(stream, # virStream instance + nbytes, # int amt of data to read + opaque): # extra data passed to sparseSendAll as opaque + fd = opaque + return os.read(fd, nbytes) + + def holeHandler(stream, # virStream instance + opaque): # extra data passed to sparseSendAll as opaque + fd = opaque + cur = os.lseek(fd, 0, os.SEEK_CUR) + # ... find out current section and its boundaries + # and set inData = True/False and sectionLen correspondingly + os.lseek(fd, cur, os.SEEK_SET) + return [inData, sectionLen] + + def skipHandler(stream, # virStream instance + length, # number of bytes to skip + opaque): # extra data passed to sparseSendAll as opaque + fd = opaque + return os.lseek(fd, length, os.SEEK_CUR) + + """ + while True: + try: + [inData, sectionLen] = holeHandler(self, opaque) + if (inData == False and sectionLen > 0): + self.sendHole(sectionLen) + skipHandler(self, sectionLen, opaque) + continue + except: + self.abort() + + want = 64 * 1024 + if (want > sectionLen): + want = sectionLen + + try: + got = handler(self, want, opaque) + except: + e = sys.exc_info()[1] + try: + self.abort() + except: + pass + raise e + + if not got: + break + + ret = self.send(got) + if ret == -2: + raise libvirtError("cannot use sendAll with " + "nonblocking stream") -- 2.13.0

Sparse streams are not that straight forward to use for the very first time. Especially the sparseRecvAll() and sparseSendAll() methods which expects callbacks. What we can do to make it easier for developers is to have an example where they can take an inspiration from. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- examples/sparsestream.py | 119 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 119 insertions(+) create mode 100755 examples/sparsestream.py diff --git a/examples/sparsestream.py b/examples/sparsestream.py new file mode 100755 index 0000000..b572a31 --- /dev/null +++ b/examples/sparsestream.py @@ -0,0 +1,119 @@ +#!/usr/bin/env python +# Example of sparse streams usage +# +# Authors: +# Michal Privoznik <mprivozn(a)redhat.com&gt; + +import libvirt, sys, os + +def bytesWriteHandler(stream, buf, opaque): + fd = opaque + return os.write(fd, buf) + +def bytesReadHandler(stream, nbytes, opaque): + fd = opaque + return os.read(fd, nbytes) + +def recvSkipHandler(stream, length, opaque): + fd = opaque + cur = os.lseek(fd, length, os.SEEK_CUR) + return os.ftruncate(fd, cur) + +def sendSkipHandler(stream, length, opaque): + fd = opaque + return os.lseek(fd, length, os.SEEK_CUR) + +def holeHandler(stream, opaque): + fd = opaque + cur = os.lseek(fd, 0, os.SEEK_CUR) + + try: + try: + data = os.lseek(fd, cur, os.SEEK_DATA) + # There are three options: + # 1) data == cur; @cur is in data + # 2) data > cur; @cur is in a hole, next data at @data + # 3) data < 0; either @cur is in trailing hole, or @cur is beyond EOF. + except: + # case 3 + inData = False + eof = os.lseek(fd, 0, os.SEEK_END) + if (eof < cur): + raise RuntimeError("Current position in file after EOF: %d" % cur) + sectionLen = eof - cur + else: + if (data > cur): + # case 2 + inData = False + sectionLen = data - cur + else: + # case 1 + inData = True + + # We don't know where does the next hole start. Let's find out. + # Here we get the same options as above + try: + hole = os.lseek(fd, data, os.SEEK_HOLE) + except: + # case 3. But wait a second. There is always a trailing hole. + # Do the best what we can here + raise RuntimeError("No trailing hole") + + if (hole == data): + # case 1. Again, this is suspicious. The reason we are here is + # because we are in data. But at the same time we are in a + # hole. WAT? + raise RuntimeError("Impossible happened") + else: + # case 2 + sectionLen = hole - data + finally: + os.lseek(fd, cur, os.SEEK_SET) + return [inData, sectionLen] + +def download(vol, st, filename): + offset = 0 + length = 0 + + try: + fd = os.open(filename, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, mode=0o0660) + vol.download(st, offset, length, libvirt.VIR_STORAGE_VOL_DOWNLOAD_SPARSE_STREAM) + st.sparseRecvAll(bytesWriteHandler, recvSkipHandler, fd) + except OSError: + print(sys.exc_info()[0]) + os.unlink(filename) + + os.close(fd) + +def upload(vol, st, filename): + offset = 0 + length = 0 + + fd = os.open(filename, os.O_RDONLY) + vol.upload(st, offset, length, libvirt.VIR_STORAGE_VOL_UPLOAD_SPARSE_STREAM) + st.sparseSendAll(bytesReadHandler, holeHandler, sendSkipHandler, fd) + + os.close(fd) + +# main +if len(sys.argv) != 5: + print("Usage: ", sys.argv[0], " URI --upload/--download VOLUME FILE") + print("Either uploads local FILE to libvirt VOLUME, or downloads libvirt ") + print("VOLUME into local FILE while preserving FILE/VOLUME sparseness") + sys.exit(1) + +conn = libvirt.open(sys.argv[1]) +vol = conn.storageVolLookupByKey(sys.argv[3]) + +st = conn.newStream() + +if sys.argv[2] == "--download": + download(vol, st, sys.argv[4]) +elif sys.argv[2] == "--upload": + upload(vol, st, sys.argv[4]) +else: + print("Unknown operation: %s " % sys.argv[1]) + sys.exit(1) + +st.finish() +conn.close() -- 2.13.0

On 05/22/2017 12:57 PM, Michal Privoznik wrote: D'oh! I haven't had nose installed. Consider this squashed into respective commits: diff --git i/sanitytest.py w/sanitytest.py index 7183baa..deec200 100644 --- i/sanitytest.py +++ w/sanitytest.py @@ -167,7 +167,8 @@ for cname in wantfunctions: # These aren't functions, they're callback signatures if name in ["virConnectAuthCallbackPtr", "virConnectCloseFunc", "virStreamSinkFunc", "virStreamSourceFunc", "virStreamEventCallback", - "virEventHandleCallback", "virEventTimeoutCallback", "virFreeCallback"]: + "virEventHandleCallback", "virEventTimeoutCallback", "virFreeCallback", + "virStreamSinkHoleFunc", "virStreamSourceHoleFunc", "virStreamSourceSkipFunc"]: continue if name[0:21] == "virConnectDomainEvent" and name[-8:] == "Callback": continue @@ -373,7 +374,8 @@ for name in sorted(finalklassmap): # These exist in C and exist in python, but we've got # a pure-python impl so don't check them - if name in ["virStreamRecvAll", "virStreamSendAll"]: + if name in ["virStreamRecvAll", "virStreamSendAll", + "virStreamSparseRecvAll", "virStreamSparseSendAll"]: continue try: Michal