[libvirt] [PATCH v2 15/38] Introduce virStreamSparseSendAll

[Michal Privoznik]

This is just a wrapper over new function that have been just
introduced: virStreamSkip() . It's very similar to
virStreamSendAll() except it handles sparse streams well.

Signed-off-by: Michal Privoznik <mprivozn at redhat.com>
---
 include/libvirt/libvirt-stream.h |  59 +++++++++++++++++++--
 src/libvirt-stream.c             | 107 +++++++++++++++++++++++++++++++++++++++
 src/libvirt_public.syms          |   1 +
 3 files changed, 164 insertions(+), 3 deletions(-)

diff --git a/include/libvirt/libvirt-stream.h b/include/libvirt/libvirt-stream.h
index e5f5126..3e9ff11 100644
--- a/include/libvirt/libvirt-stream.h
+++ b/include/libvirt/libvirt-stream.h
@@ -69,9 +69,9 @@ int virStreamHoleSize(virStreamPtr,
  * @nbytes: size of the data array
  * @opaque: optional application provided data
  *
- * The virStreamSourceFunc callback is used together
- * with the virStreamSendAll function for libvirt to
- * obtain the data that is to be sent.
+ * The virStreamSourceFunc callback is used together with
+ * the virStreamSendAll and virStreamSparseSendAll functions
+ * for libvirt to obtain the data that is to be sent.
  *
  * The callback will be invoked multiple times,
  * fetching data in small chunks. The application
@@ -95,6 +95,59 @@ int virStreamSendAll(virStreamPtr st,
                      void *opaque);
 
 /**
+ * virStreamSourceHoleFunc:
+ * @st: the stream object
+ * @inData: are we in data section
+ * @length: how long is the section we are currently in
+ * @opaque: optional application provided data
+ *
+ * The virStreamSourceHoleFunc callback is used together
+ * with the virStreamSparseSendAll function for libvirt to
+ * obtain the length of section stream is currently in.
+ *
+ * Moreover, upon successful return, @length should be
+ * updated with how much bytes are there left until current
+ * section ends (be it data section or hole section) and if
+ * the stream is currently in data section, @inData should
+ * be set to a non-zero value and vice versa.
+ * As a corner case, there's an implicit hole at the end of
+ * each file. If that's the case, @inData should be set to 0
+ * as well as @length.
+ * Moreover, this function should upon its return leave the
+ * file in the position it was called with.
+ *
+ * Returns 0 on success,
+ *        -1 upon error
+ */
+typedef int (*virStreamSourceHoleFunc)(virStreamPtr st,
+                                       int *inData,
+                                       unsigned long long *length,
+                                       void *opaque);
+
+/**
+ * virStreamSourceSkipFunc:
+ * @st: the stream object
+ * @length: stream hole size
+ * @opaque: optional application provided data
+ *
+ * This callback is used together with the virStreamSparseSendAll
+ * to skip holes in the underlying file as reported by
+ * virStreamSourceHoleFunc.
+ *
+ * Returns 0 on success,
+ *        -1 upon error.
+ */
+typedef int (*virStreamSourceSkipFunc)(virStreamPtr st,
+                                       unsigned long long length,
+                                       void *opaque);
+
+int virStreamSparseSendAll(virStreamPtr st,
+                           virStreamSourceFunc handler,
+                           virStreamSourceHoleFunc holeHandler,
+                           virStreamSourceSkipFunc skipHandler,
+                           void *opaque);
+
+/**
  * virStreamSinkFunc:
  *
  * @st: the stream object
diff --git a/src/libvirt-stream.c b/src/libvirt-stream.c
index 81190cc..7ad5a38 100644
--- a/src/libvirt-stream.c
+++ b/src/libvirt-stream.c
@@ -565,7 +565,114 @@ virStreamSendAll(virStreamPtr stream,
 }
 
 
+
 /**
+ * virStreamSparseSendAll:
+ * @stream: pointer to the stream object
+ * @handler: source callback for reading data from application
+ * @holeHandler: source callback for determining holes
+ * @skipHandler: skip holes as reported by @holeHandler
+ * @opaque: application defined data
+ *
+ * Some dummy description here.
+ *
+ * Opaque data in @opaque are shared between @handler, @holeHandler and @skipHandler.
+ *
+ * Returns 0 if all the data was successfully sent. The caller
+ * should invoke virStreamFinish(st) to flush the stream upon
+ * success and then virStreamFree.
+ *
+ * Returns -1 upon any error, with virStreamAbort() already
+ * having been called,  so the caller need only call
+ * virStreamFree().
+ */
+int virStreamSparseSendAll(virStreamPtr stream,
+                           virStreamSourceFunc handler,
+                           virStreamSourceHoleFunc holeHandler,
+                           virStreamSourceSkipFunc skipHandler,
+                           void *opaque)
+{
+    char *bytes = NULL;
+    size_t want = VIR_NET_MESSAGE_LEGACY_PAYLOAD_MAX;
+    int ret = -1;
+    unsigned long long dataLen = 0;
+
+    VIR_DEBUG("stream=%p handler=%p holeHandler=%p opaque=%p",
+              stream, handler, holeHandler, opaque);
+
+    virResetLastError();
+
+    virCheckStreamReturn(stream, -1);
+    virCheckNonNullArgGoto(handler, cleanup);
+    virCheckNonNullArgGoto(holeHandler, cleanup);
+    virCheckNonNullArgGoto(skipHandler, cleanup);
+
+    if (stream->flags & VIR_STREAM_NONBLOCK) {
+        virReportError(VIR_ERR_OPERATION_INVALID, "%s",
+                       _("data sources cannot be used for non-blocking streams"));
+        goto cleanup;
+    }
+
+    if (VIR_ALLOC_N(bytes, want) < 0)
+        goto cleanup;
+
+    for (;;) {
+        int inData, got, offset = 0;
+        unsigned long long sectionLen;
+
+        if (!dataLen) {
+            if (holeHandler(stream, &inData, &sectionLen, opaque) < 0) {
+                virStreamAbort(stream);
+                goto cleanup;
+            }
+
+            if (!inData && sectionLen) {
+                if (virStreamSkip(stream, sectionLen) < 0) {
+                    virStreamAbort(stream);
+                    goto cleanup;
+                }
+
+                if (skipHandler(stream, sectionLen, opaque) < 0) {
+                    virReportSystemError(errno, "%s",
+                                         _("unable to skip hole"));
+                    virStreamAbort(stream);
+                    goto cleanup;
+                }
+                continue;
+            } else {
+                dataLen = sectionLen;
+            }
+        }
+
+        if (want > dataLen)
+            want = dataLen;
+
+        got = (handler)(stream, bytes, want, opaque);
+        if (got < 0) {
+            virStreamAbort(stream);
+            goto cleanup;
+        }
+        if (got == 0)
+            break;
+        while (offset < got) {
+            int done;
+            done = virStreamSend(stream, bytes + offset, got - offset);
+            if (done < 0)
+                goto cleanup;
+            offset += done;
+            dataLen -= done;
+        }
+    }
+    ret = 0;
+
+ cleanup:
+    VIR_FREE(bytes);
+
+    if (ret != 0)
+        virDispatchError(stream->conn);
+
+    return ret;
+}/**
  * virStreamRecvAll:
  * @stream: pointer to the stream object
  * @handler: sink callback for writing data to application
diff --git a/src/libvirt_public.syms b/src/libvirt_public.syms
index 008dc59..ea4ddd5 100644
--- a/src/libvirt_public.syms
+++ b/src/libvirt_public.syms
@@ -765,6 +765,7 @@ LIBVIRT_3.3.0 {
         virStreamRecvFlags;
         virStreamSkip;
         virStreamSparseRecvAll;
+        virStreamSparseSendAll;
 } LIBVIRT_3.1.0;
 
 # .... define new API here using predicted next version number ....
-- 
2.10.2