On Wed, Jun 22, 2016 at 04:43:18PM +0200, Michal Privoznik wrote: Is it really safe to ignore the value here ? IIUC, callers of this function would be justified in thinking it would *not* have a side effect on file position. IOW, I think we'd probably want to treat this error as fatal too. I think it'd be desirable to have a unit test written explicitly for this function, since there's a few fun edge cases to worry about here. Regards, Daniel -- |: http://berrange.com -o- http://www.flickr.com/photos/dberrange/ :| |: http://libvirt.org -o- http://virt-manager.org :| |: http://autobuild.org -o- http://search.cpan.org/~danberr/ :| |: http://entangle-photo.org -o- http://live.gnome.org/gtk-vnc :|

On Wed, Jun 22, 2016 at 04:43:24PM +0200, Michal Privoznik wrote: We can put this in a 2.1.0 block now ACK Regards, Daniel -- |: http://berrange.com -o- http://www.flickr.com/photos/dberrange/ :| |: http://libvirt.org -o- http://virt-manager.org :| |: http://autobuild.org -o- http://search.cpan.org/~danberr/ :| |: http://entangle-photo.org -o- http://live.gnome.org/gtk-vnc :|

On Wed, Jun 22, 2016 at 04:43:25PM +0200, Michal Privoznik wrote: Use 2.1.0 now ACK Regards, Daniel -- |: http://berrange.com -o- http://www.flickr.com/photos/dberrange/ :| |: http://libvirt.org -o- http://virt-manager.org :| |: http://autobuild.org -o- http://search.cpan.org/~danberr/ :| |: http://entangle-photo.org -o- http://live.gnome.org/gtk-vnc :|

On Wed, Jun 22, 2016 at 04:43:27PM +0200, Michal Privoznik wrote: ACK 2.1.0 Regards, Daniel -- |: http://berrange.com -o- http://www.flickr.com/photos/dberrange/ :| |: http://libvirt.org -o- http://virt-manager.org :| |: http://autobuild.org -o- http://search.cpan.org/~danberr/ :| |: http://entangle-photo.org -o- http://live.gnome.org/gtk-vnc :|

This has no real added value right now, but is going to be very helpful later. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/remote/remote_driver.c | 6 ++++-- src/rpc/gendispatch.pl | 2 +- src/rpc/virnetclientstream.c | 6 +++++- src/rpc/virnetclientstream.h | 3 ++- 4 files changed, 12 insertions(+), 5 deletions(-) diff --git a/src/remote/remote_driver.c b/src/remote/remote_driver.c index bfd8e8e..e80ec58 100644 --- a/src/remote/remote_driver.c +++ b/src/remote/remote_driver.c @@ -5760,7 +5760,8 @@ remoteDomainMigratePrepareTunnel3(virConnectPtr dconn, memset(&args, 0, sizeof(args)); memset(&ret, 0, sizeof(ret)); - if (!(netst = virNetClientStreamNew(priv->remoteProgram, + if (!(netst = virNetClientStreamNew(st, + priv->remoteProgram, REMOTE_PROC_DOMAIN_MIGRATE_PREPARE_TUNNEL3, priv->counter))) goto done; @@ -6684,7 +6685,8 @@ remoteDomainMigratePrepareTunnel3Params(virConnectPtr dconn, goto cleanup; } - if (!(netst = virNetClientStreamNew(priv->remoteProgram, + if (!(netst = virNetClientStreamNew(st, + priv->remoteProgram, REMOTE_PROC_DOMAIN_MIGRATE_PREPARE_TUNNEL3_PARAMS, priv->counter))) goto cleanup; diff --git a/src/rpc/gendispatch.pl b/src/rpc/gendispatch.pl index 173189c..e608812 100755 --- a/src/rpc/gendispatch.pl +++ b/src/rpc/gendispatch.pl @@ -1738,7 +1738,7 @@ elsif ($mode eq "client") { if ($call->{streamflag} ne "none") { print "\n"; - print " if (!(netst = virNetClientStreamNew(priv->remoteProgram, $call->{constname}, priv->counter)))\n"; + print " if (!(netst = virNetClientStreamNew(st, priv->remoteProgram, $call->{constname}, priv->counter)))\n"; print " goto done;\n"; print "\n"; print " if (virNetClientAddStream(priv->client, netst) < 0) {\n"; diff --git a/src/rpc/virnetclientstream.c b/src/rpc/virnetclientstream.c index 34989a9..8920395 100644 --- a/src/rpc/virnetclientstream.c +++ b/src/rpc/virnetclientstream.c @@ -36,6 +36,8 @@ VIR_LOG_INIT("rpc.netclientstream"); struct _virNetClientStream { virObjectLockable parent; + virStreamPtr stream; /* Reverse pointer to parent stream */ + virNetClientProgramPtr prog; int proc; unsigned serial; @@ -133,7 +135,8 @@ virNetClientStreamEventTimer(int timer ATTRIBUTE_UNUSED, void *opaque) } -virNetClientStreamPtr virNetClientStreamNew(virNetClientProgramPtr prog, +virNetClientStreamPtr virNetClientStreamNew(virStreamPtr stream, + virNetClientProgramPtr prog, int proc, unsigned serial) { @@ -145,6 +148,7 @@ virNetClientStreamPtr virNetClientStreamNew(virNetClientProgramPtr prog, if (!(st = virObjectLockableNew(virNetClientStreamClass))) return NULL; + st->stream = stream; st->prog = prog; st->proc = proc; st->serial = serial; diff --git a/src/rpc/virnetclientstream.h b/src/rpc/virnetclientstream.h index a0d2be9..e278dab 100644 --- a/src/rpc/virnetclientstream.h +++ b/src/rpc/virnetclientstream.h @@ -32,7 +32,8 @@ typedef virNetClientStream *virNetClientStreamPtr; typedef void (*virNetClientStreamEventCallback)(virNetClientStreamPtr stream, int events, void *opaque); -virNetClientStreamPtr virNetClientStreamNew(virNetClientProgramPtr prog, +virNetClientStreamPtr virNetClientStreamNew(virStreamPtr stream, + virNetClientProgramPtr prog, int proc, unsigned serial); -- 2.8.4

This is going to be RPC representation for virStreamSkip. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/rpc/virnetprotocol.x | 4 ++++ src/virnetprotocol-structs | 3 +++ 2 files changed, 7 insertions(+) diff --git a/src/rpc/virnetprotocol.x b/src/rpc/virnetprotocol.x index 9ce33b0..3623588 100644 --- a/src/rpc/virnetprotocol.x +++ b/src/rpc/virnetprotocol.x @@ -236,3 +236,7 @@ struct virNetMessageError { int int2; virNetMessageNetwork net; /* unused */ }; + +struct virNetStreamSkip { + unsigned hyper length; +}; diff --git a/src/virnetprotocol-structs b/src/virnetprotocol-structs index af4526c..a8cc603 100644 --- a/src/virnetprotocol-structs +++ b/src/virnetprotocol-structs @@ -42,3 +42,6 @@ struct virNetMessageError { int int2; virNetMessageNetwork net; }; +struct virNetStreamSkip { + uint64_t length; +}; -- 2.8.4

Ideally, this would be generated, but to achieve that corresponding XDR definitions needed to go into a different .x file. But they belong just to the one that they are right now. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- tools/wireshark/src/packet-libvirt.c | 40 ++++++++++++++++++++++++++++++++++++ tools/wireshark/src/packet-libvirt.h | 2 ++ 2 files changed, 42 insertions(+) diff --git a/tools/wireshark/src/packet-libvirt.c b/tools/wireshark/src/packet-libvirt.c index aa1c323..aa49559 100644 --- a/tools/wireshark/src/packet-libvirt.c +++ b/tools/wireshark/src/packet-libvirt.c @@ -52,8 +52,11 @@ static int hf_libvirt_serial = -1; static int hf_libvirt_status = -1; static int hf_libvirt_stream = -1; static int hf_libvirt_num_of_fds = -1; +static int hf_libvirt_stream_skip_length = -1; +static int hf_libvirt_stream_skip = -1; int hf_libvirt_unknown = -1; static gint ett_libvirt = -1; +static gint ett_libvirt_stream_skip = -1; #define XDR_PRIMITIVE_DISSECTOR(xtype, ctype, ftype) \ static gboolean \ @@ -328,6 +331,28 @@ dissect_libvirt_payload_xdr_data(tvbuff_t *tvb, proto_tree *tree, gint payload_l dissect_libvirt_fds(tvb, start + payload_length, nfds); } +static gboolean +dissect_xdr_stream_skip(tvbuff_t *tvb, proto_tree *tree, XDR *xdrs, int hf) +{ + goffset start = VIR_HEADER_LEN; + proto_item *ti; + + ti = proto_tree_add_item(tree, hf_libvirt_stream_skip, tvb, start, -1, ENC_NA); + tree = proto_item_add_subtree(ti, ett_libvirt_stream_skip); + + hf = hf_libvirt_stream_skip_length; + if (!dissect_xdr_u_hyper(tvb, tree, xdrs, hf)) return FALSE; + proto_item_set_len(ti, xdr_getpos(xdrs) - start); + return TRUE; +} + + +static void +dissect_libvirt_stream_skip(tvbuff_t *tvb, proto_tree *tree, gint payload_length, guint32 status) +{ + proto_tree_add_item(tree, hf_libvirt_stream_skip_length, tvb, VIR_HEADER_LEN, -1, ENC_NA); +} + static void dissect_libvirt_payload(tvbuff_t *tvb, proto_tree *tree, guint32 prog, guint32 proc, guint32 type, guint32 status) @@ -348,6 +373,8 @@ dissect_libvirt_payload(tvbuff_t *tvb, proto_tree *tree, dissect_libvirt_payload_xdr_data(tvb, tree, payload_length, status, VIR_ERROR_MESSAGE_DISSECTOR); } else if (type == VIR_NET_STREAM) { /* implicitly, status == VIR_NET_CONTINUE */ dissect_libvirt_stream(tvb, tree, payload_length); + } else if (type == VIR_NET_STREAM_SKIP) { + dissect_libvirt_payload_xdr_data(tvb, tree, payload_length, status, dissect_xdr_stream_skip); } else { goto unknown; } @@ -517,6 +544,18 @@ proto_register_libvirt(void) NULL, 0x0, NULL, HFILL} }, + { &hf_libvirt_stream_skip, + { "stream_skip", "libvirt.stream_skip", + FT_BYTES, BASE_NONE, + NULL, 0x0, + NULL, HFILL} + }, + { &hf_libvirt_stream_skip_length, + { "length", "libvirt.stream_skip.length", + FT_UINT64, BASE_DEC, + NULL, 0x0, + NULL, HFILL} + }, { &hf_libvirt_unknown, { "unknown", "libvirt.unknown", FT_BYTES, BASE_NONE, @@ -527,6 +566,7 @@ proto_register_libvirt(void) static gint *ett[] = { VIR_DYNAMIC_ETTSET + &ett_libvirt_stream_skip, &ett_libvirt }; diff --git a/tools/wireshark/src/packet-libvirt.h b/tools/wireshark/src/packet-libvirt.h index 5f99fdf..006aa6d 100644 --- a/tools/wireshark/src/packet-libvirt.h +++ b/tools/wireshark/src/packet-libvirt.h @@ -53,6 +53,7 @@ enum vir_net_message_type { VIR_NET_STREAM = 3, VIR_NET_CALL_WITH_FDS = 4, VIR_NET_REPLY_WITH_FDS = 5, + VIR_NET_STREAM_SKIP = 6, }; enum vir_net_message_status { @@ -76,6 +77,7 @@ static const value_string type_strings[] = { { VIR_NET_STREAM, "STREAM" }, { VIR_NET_CALL_WITH_FDS, "CALL_WITH_FDS" }, { VIR_NET_REPLY_WITH_FDS, "REPLY_WITH_FDS" }, + { VIR_NET_STREAM_SKIP, "STREAM_SKIP" }, { -1, NULL } }; -- 2.8.4

While the previous commit implemented a helper for sending a STREAM_SKIP packet for daemon, this is a client's counterpart. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/libvirt_remote.syms | 1 + src/rpc/virnetclientstream.c | 52 ++++++++++++++++++++++++++++++++++++++++++++ src/rpc/virnetclientstream.h | 4 ++++ 3 files changed, 57 insertions(+) diff --git a/src/libvirt_remote.syms b/src/libvirt_remote.syms index b97b9b1..16fde79 100644 --- a/src/libvirt_remote.syms +++ b/src/libvirt_remote.syms @@ -54,6 +54,7 @@ virNetClientStreamQueuePacket; virNetClientStreamRaiseError; virNetClientStreamRecvPacket; virNetClientStreamSendPacket; +virNetClientStreamSendSkip; virNetClientStreamSetError; diff --git a/src/rpc/virnetclientstream.c b/src/rpc/virnetclientstream.c index fb83693..d7d2b0e 100644 --- a/src/rpc/virnetclientstream.c +++ b/src/rpc/virnetclientstream.c @@ -431,6 +431,58 @@ int virNetClientStreamRecvPacket(virNetClientStreamPtr st, } +int +virNetClientStreamSendSkip(virNetClientStreamPtr st, + virNetClientPtr client, + unsigned long long length) +{ + virNetMessagePtr msg = NULL; + virNetStreamSkip data; + int ret = -1; + + VIR_DEBUG("st=%p length=%llu", st, length); + + if (!st->skippable) { + virReportError(VIR_ERR_OPERATION_INVALID, "%s", + _("Skipping is not supported with this stream")); + return -1; + } + + memset(&data, 0, sizeof(data)); + data.length = length; + + if (!(msg = virNetMessageNew(false))) + return -1; + + virObjectLock(st); + + msg->header.prog = virNetClientProgramGetProgram(st->prog); + msg->header.vers = virNetClientProgramGetVersion(st->prog); + msg->header.status = VIR_NET_CONTINUE; + msg->header.type = VIR_NET_STREAM_SKIP; + msg->header.serial = st->serial; + msg->header.proc = st->proc; + + virObjectUnlock(st); + + if (virNetMessageEncodeHeader(msg) < 0) + goto cleanup; + + if (virNetMessageEncodePayload(msg, + (xdrproc_t) xdr_virNetStreamSkip, + &data) < 0) + goto cleanup; + + if (virNetClientSendNoReply(client, msg) < 0) + goto cleanup; + + ret = 0; + cleanup: + virNetMessageFree(msg); + return ret; +} + + int virNetClientStreamEventAddCallback(virNetClientStreamPtr st, int events, virNetClientStreamEventCallback cb, diff --git a/src/rpc/virnetclientstream.h b/src/rpc/virnetclientstream.h index 0a5aafd..a648b7c 100644 --- a/src/rpc/virnetclientstream.h +++ b/src/rpc/virnetclientstream.h @@ -61,6 +61,10 @@ int virNetClientStreamRecvPacket(virNetClientStreamPtr st, size_t nbytes, bool nonblock); +int virNetClientStreamSendSkip(virNetClientStreamPtr st, + virNetClientPtr client, + unsigned long long length); + int virNetClientStreamEventAddCallback(virNetClientStreamPtr st, int events, virNetClientStreamEventCallback cb, -- 2.8.4

This is a function that handles an incoming STREAM_SKIP packet. Even though it is not wired up yet, it will be soon. At the beginning do couple of checks whether server plays nicely and sent us a STREAM_SKIP packed only after we've enabled sparse streams. Then decodes the message payload to see how big the hole is and stores it in passed @length argument. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/rpc/virnetclientstream.c | 63 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 63 insertions(+) diff --git a/src/rpc/virnetclientstream.c b/src/rpc/virnetclientstream.c index d7d2b0e..0e982ba 100644 --- a/src/rpc/virnetclientstream.c +++ b/src/rpc/virnetclientstream.c @@ -28,6 +28,7 @@ #include "virerror.h" #include "virlog.h" #include "virthread.h" +#include "libvirt_internal.h" #define VIR_FROM_THIS VIR_FROM_RPC @@ -55,6 +56,7 @@ struct _virNetClientStream { bool incomingEOF; bool skippable; /* User requested skippable stream */ + unsigned long long skipLength; /* Size of incoming hole in stream. */ virNetClientStreamEventCallback cb; void *cbOpaque; @@ -358,6 +360,67 @@ int virNetClientStreamSendPacket(virNetClientStreamPtr st, return -1; } + +static int ATTRIBUTE_UNUSED +virNetClientStreamHandleSkip(virNetClientPtr client, + virNetClientStreamPtr st) +{ + virNetMessagePtr msg; + virNetStreamSkip data; + int ret = -1; + + VIR_DEBUG("client=%p st=%p", client, st); + + msg = st->rx; + memset(&data, 0, sizeof(data)); + + /* We should not be called unless there's VIR_NET_STREAM_SKIP + * message at the head of the list. But doesn't hurt to check */ + if (!msg) { + virReportError(VIR_ERR_INTERNAL_ERROR, "%s", + _("No message in the queue")); + goto cleanup; + } + + if (msg->header.type != VIR_NET_STREAM_SKIP) { + virReportError(VIR_ERR_INTERNAL_ERROR, + _("Invalid message prog=%d type=%d serial=%u proc=%d"), + msg->header.prog, + msg->header.type, + msg->header.serial, + msg->header.proc); + goto cleanup; + } + + /* Server should not send us VIR_NET_STREAM_SKIP unless we + * have requested so. But does not hurt to check ... */ + if (!st->skippable) { + virReportError(VIR_ERR_RPC, "%s", + _("Unexpected stream skip")); + goto cleanup; + } + + if (virNetMessageDecodePayload(msg, + (xdrproc_t) xdr_virNetStreamSkip, + &data) < 0) { + virReportError(VIR_ERR_INTERNAL_ERROR, "%s", + _("Malformed stream skip packet")); + goto cleanup; + } + + virNetMessageQueueServe(&st->rx); + virNetMessageFree(msg); + st->skipLength += data.length; + + ret = 0; + cleanup: + if (ret < 0) { + /* Abort stream? */ + } + return ret; +} + + int virNetClientStreamRecvPacket(virNetClientStreamPtr st, virNetClientPtr client, char *data, -- 2.8.4

Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/remote/remote_driver.c | 3 ++- src/rpc/virnetclientstream.c | 10 +++++++--- src/rpc/virnetclientstream.h | 3 ++- 3 files changed, 11 insertions(+), 5 deletions(-) diff --git a/src/remote/remote_driver.c b/src/remote/remote_driver.c index 2dc21d2..7192f56 100644 --- a/src/remote/remote_driver.c +++ b/src/remote/remote_driver.c @@ -5255,7 +5255,8 @@ remoteStreamRecvFlags(virStreamPtr st, priv->client, data, nbytes, - (st->flags & VIR_STREAM_NONBLOCK)); + (st->flags & VIR_STREAM_NONBLOCK), + flags); VIR_DEBUG("Done %d", rv); diff --git a/src/rpc/virnetclientstream.c b/src/rpc/virnetclientstream.c index 0e982ba..c8c89ec 100644 --- a/src/rpc/virnetclientstream.c +++ b/src/rpc/virnetclientstream.c @@ -425,13 +425,17 @@ int virNetClientStreamRecvPacket(virNetClientStreamPtr st, virNetClientPtr client, char *data, size_t nbytes, - bool nonblock) + bool nonblock, + unsigned int flags) { int rv = -1; size_t want; - VIR_DEBUG("st=%p client=%p data=%p nbytes=%zu nonblock=%d", - st, client, data, nbytes, nonblock); + VIR_DEBUG("st=%p client=%p data=%p nbytes=%zu nonblock=%d flags=%x", + st, client, data, nbytes, nonblock, flags); + + virCheckFlags(0, -1); + virObjectLock(st); if (!st->rx && !st->incomingEOF) { virNetMessagePtr msg; diff --git a/src/rpc/virnetclientstream.h b/src/rpc/virnetclientstream.h index a648b7c..2835066 100644 --- a/src/rpc/virnetclientstream.h +++ b/src/rpc/virnetclientstream.h @@ -59,7 +59,8 @@ int virNetClientStreamRecvPacket(virNetClientStreamPtr st, virNetClientPtr client, char *data, size_t nbytes, - bool nonblock); + bool nonblock, + unsigned int flags); int virNetClientStreamSendSkip(virNetClientStreamPtr st, virNetClientPtr client, -- 2.8.4

Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/remote/remote_driver.c | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/src/remote/remote_driver.c b/src/remote/remote_driver.c index 7192f56..445b7ce 100644 --- a/src/remote/remote_driver.c +++ b/src/remote/remote_driver.c @@ -5302,6 +5302,29 @@ remoteStreamSkip(virStreamPtr st, } +static int +remoteStreamHoleSize(virStreamPtr st, + unsigned long long *length) +{ + VIR_DEBUG("st=%p length=%p", st, length); + struct private_data *priv = st->conn->privateData; + virNetClientStreamPtr privst = st->privateData; + int rv; + + if (virNetClientStreamRaiseError(privst)) + return -1; + + remoteDriverLock(priv); + priv->localUses++; + remoteDriverUnlock(priv); + + rv = virNetClientStreamHoleSize(priv->client, privst, length); + + remoteDriverLock(priv); + priv->localUses--; + remoteDriverUnlock(priv); + return rv; +} struct remoteStreamCallbackData { virStreamPtr st; virStreamEventCallback cb; @@ -5477,6 +5500,7 @@ static virStreamDriver remoteStreamDrv = { .streamRecvFlags = remoteStreamRecvFlags, .streamSend = remoteStreamSend, .streamSkip = remoteStreamSkip, + .streamHoleSize = remoteStreamHoleSize, .streamFinish = remoteStreamFinish, .streamAbort = remoteStreamAbort, .streamEventAddCallback = remoteStreamEventAddCallback, -- 2.8.4

Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/remote/remote_driver.c | 2 +- src/rpc/virnetclientstream.c | 11 ++++++++++- 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/src/remote/remote_driver.c b/src/remote/remote_driver.c index 445b7ce..0206510 100644 --- a/src/remote/remote_driver.c +++ b/src/remote/remote_driver.c @@ -5242,7 +5242,7 @@ remoteStreamRecvFlags(virStreamPtr st, virNetClientStreamPtr privst = st->privateData; int rv; - virCheckFlags(0, -1); + virCheckFlags(VIR_STREAM_RECV_STOP_AT_HOLE, -1); if (virNetClientStreamRaiseError(privst)) return -1; diff --git a/src/rpc/virnetclientstream.c b/src/rpc/virnetclientstream.c index f63072e..e8ef3a3 100644 --- a/src/rpc/virnetclientstream.c +++ b/src/rpc/virnetclientstream.c @@ -436,7 +436,7 @@ int virNetClientStreamRecvPacket(virNetClientStreamPtr st, VIR_DEBUG("st=%p client=%p data=%p nbytes=%zu nonblock=%d flags=%x", st, client, data, nbytes, nonblock, flags); - virCheckFlags(0, -1); + virCheckFlags(VIR_STREAM_RECV_STOP_AT_HOLE, -1); virObjectLock(st); @@ -499,6 +499,15 @@ int virNetClientStreamRecvPacket(virNetClientStreamPtr st, /* Pretend skipLength zeroes was read from stream. */ size_t len = want; + /* Yes, pretend unless we are asked not to. */ + if (flags & VIR_STREAM_RECV_STOP_AT_HOLE) { + virReportError(VIR_ERR_INTERNAL_ERROR, "%s", + _("Stream is in a hole")); + rv = -3; + goto cleanup; + } + + if (len > st->skipLength) len = st->skipLength; -- 2.8.4

Implement virStreamSkip and virStreamInData callbacks. These callbacks do no magic, just skip a hole or detect whether we are in a data section of a file or in a hole and how much bytes can we read until section changes. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/fdstream.c | 48 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 48 insertions(+) diff --git a/src/fdstream.c b/src/fdstream.c index bebeac3..be94325 100644 --- a/src/fdstream.c +++ b/src/fdstream.c @@ -504,11 +504,59 @@ static int virFDStreamRead(virStreamPtr st, char *bytes, size_t nbytes) } +static int +virFDStreamSkip(virStreamPtr st, + unsigned long long length) +{ + virFDStreamDataPtr fdst = st->privateData; + off_t off; + int ret = -1; + + virObjectLock(fdst); + if (fdst->length) { + if (fdst->offset + length > fdst->length) { + virReportSystemError(ENOSPC, "%s", + _("cannot write to stream")); + goto cleanup; + } + fdst->offset += length; + } + + off = lseek(fdst->fd, length, SEEK_CUR); + if (off == (off_t) -1) { + virReportSystemError(errno, "%s", + _("unable to seek")); + goto cleanup; + } + ret = 0; + cleanup: + virObjectUnlock(fdst); + return ret; +} + + +static int +virFDStreamInData(virStreamPtr st, + int *inData, + unsigned long long *length) +{ + virFDStreamDataPtr fdst = st->privateData; + int ret = -1; + + virObjectLock(fdst); + ret = virFileInData(fdst->fd, inData, length); + virObjectUnlock(fdst); + return ret; +} + + static virStreamDriver virFDStreamDrv = { .streamSend = virFDStreamWrite, .streamRecv = virFDStreamRead, .streamFinish = virFDStreamClose, .streamAbort = virFDStreamAbort, + .streamSkip = virFDStreamSkip, + .streamInData = virFDStreamInData, .streamEventAddCallback = virFDStreamAddCallback, .streamEventUpdateCallback = virFDStreamUpdateCallback, .streamEventRemoveCallback = virFDStreamRemoveCallback -- 2.8.4

These flags to APIs will tell if caller wants to use sparse stream for storage transfer. At the same time, it's safe to enable them in storage driver frontend and rely on our backends checking the flags. This way we can enable specific flags only on some specific backends, e.g. enable VIR_STORAGE_VOL_DOWNLOAD_SPARSE_STREAM for filesystem backend but not iSCSI backend. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- include/libvirt/libvirt-storage.h | 9 +++++++++ src/libvirt-storage.c | 4 ++-- src/remote/remote_protocol.x | 2 ++ src/storage/storage_driver.c | 4 ++-- 4 files changed, 15 insertions(+), 4 deletions(-) diff --git a/include/libvirt/libvirt-storage.h b/include/libvirt/libvirt-storage.h index a67a97f..e08af50 100644 --- a/include/libvirt/libvirt-storage.h +++ b/include/libvirt/libvirt-storage.h @@ -337,11 +337,20 @@ virStorageVolPtr virStorageVolCreateXMLFrom (virStoragePoolPtr pool, const char *xmldesc, virStorageVolPtr clonevol, unsigned int flags); + +typedef enum { + VIR_STORAGE_VOL_DOWNLOAD_SPARSE_STREAM = 1 << 0, /* Use sparse stream */ +} virStorageVolDownloadFlags; + int virStorageVolDownload (virStorageVolPtr vol, virStreamPtr stream, unsigned long long offset, unsigned long long length, unsigned int flags); +typedef enum { + VIR_STORAGE_VOL_UPLOAD_SPARSE_STREAM = 1 << 0, /* Use sparse stream */ +} virStorageVolUploadFlags; + int virStorageVolUpload (virStorageVolPtr vol, virStreamPtr stream, unsigned long long offset, diff --git a/src/libvirt-storage.c b/src/libvirt-storage.c index cebe02f..5ab719f 100644 --- a/src/libvirt-storage.c +++ b/src/libvirt-storage.c @@ -1549,7 +1549,7 @@ virStorageVolCreateXMLFrom(virStoragePoolPtr pool, * @stream: stream to use as output * @offset: position in @vol to start reading from * @length: limit on amount of data to download - * @flags: extra flags; not used yet, so callers should always pass 0 + * @flags: bitwise-OR of virStorageVolDownloadFlags * * Download the content of the volume as a stream. If @length * is zero, then the remaining contents of the volume after @@ -1613,7 +1613,7 @@ virStorageVolDownload(virStorageVolPtr vol, * @stream: stream to use as input * @offset: position to start writing to * @length: limit on amount of data to upload - * @flags: extra flags; not used yet, so callers should always pass 0 + * @flags: bitwise-OR of virStorageVolUploadFlags * * Upload new content to the volume from a stream. This call * will fail if @offset + @length exceeds the size of the diff --git a/src/remote/remote_protocol.x b/src/remote/remote_protocol.x index d11bfdf..65dc2a8 100644 --- a/src/remote/remote_protocol.x +++ b/src/remote/remote_protocol.x @@ -4798,6 +4798,7 @@ enum remote_procedure { /** * @generate: both * @writestream: 1 + * @sparseflag: VIR_STORAGE_VOL_UPLOAD_SPARSE_STREAM * @acl: storage_vol:data_write */ REMOTE_PROC_STORAGE_VOL_UPLOAD = 208, @@ -4805,6 +4806,7 @@ enum remote_procedure { /** * @generate: both * @readstream: 1 + * @sparseflag: VIR_STORAGE_VOL_DOWNLOAD_SPARSE_STREAM * @acl: storage_vol:data_read */ REMOTE_PROC_STORAGE_VOL_DOWNLOAD = 209, diff --git a/src/storage/storage_driver.c b/src/storage/storage_driver.c index e2d729f..a153a59 100644 --- a/src/storage/storage_driver.c +++ b/src/storage/storage_driver.c @@ -2224,7 +2224,7 @@ storageVolDownload(virStorageVolPtr obj, virStorageVolDefPtr vol = NULL; int ret = -1; - virCheckFlags(0, -1); + virCheckFlags(VIR_STORAGE_VOL_DOWNLOAD_SPARSE_STREAM, -1); if (!(vol = virStorageVolDefFromVol(obj, &pool, &backend))) return -1; @@ -2394,7 +2394,7 @@ storageVolUpload(virStorageVolPtr obj, virStorageVolStreamInfoPtr cbdata = NULL; int ret = -1; - virCheckFlags(0, -1); + virCheckFlags(VIR_STORAGE_VOL_UPLOAD_SPARSE_STREAM, -1); if (!(vol = virStorageVolDefFromVol(obj, &pool, &backend))) return -1; -- 2.8.4

Similarly to previous commit, implement sparse streams feature for vol-upload. This is, however, slightly different approach, because we must implement a function that will tell us whether we are in a data section or in a hole. But there's no magic hidden in here. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- tools/virsh-volume.c | 37 ++++++++++++++++++++++++------------- tools/virsh.c | 28 ++++++++++++++++++++++++++++ tools/virsh.h | 14 ++++++++++++++ tools/virsh.pod | 3 ++- 4 files changed, 68 insertions(+), 14 deletions(-) diff --git a/tools/virsh-volume.c b/tools/virsh-volume.c index 27744c7..d6434d5 100644 --- a/tools/virsh-volume.c +++ b/tools/virsh-volume.c @@ -659,18 +659,13 @@ static const vshCmdOptDef opts_vol_upload[] = { .type = VSH_OT_INT, .help = N_("amount of data to upload") }, + {.name = "sparse", + .type = VSH_OT_BOOL, + .help = N_("preserve sparseness of volume") + }, {.name = NULL} }; -static int -cmdVolUploadSource(virStreamPtr st ATTRIBUTE_UNUSED, - char *bytes, size_t nbytes, void *opaque) -{ - int *fd = opaque; - - return saferead(*fd, bytes, nbytes); -} - static bool cmdVolUpload(vshControl *ctl, const vshCmd *cmd) { @@ -682,6 +677,8 @@ cmdVolUpload(vshControl *ctl, const vshCmd *cmd) const char *name = NULL; unsigned long long offset = 0, length = 0; virshControlPtr priv = ctl->privData; + unsigned int flags = 0; + virshStreamCallbackData cbData; if (vshCommandOptULongLong(ctl, cmd, "offset", &offset) < 0) return false; @@ -700,19 +697,33 @@ cmdVolUpload(vshControl *ctl, const vshCmd *cmd) goto cleanup; } + cbData.ctl = ctl; + cbData.fd = fd; + + if (vshCommandOptBool(cmd, "sparse")) + flags |= VIR_STORAGE_VOL_UPLOAD_SPARSE_STREAM; + if (!(st = virStreamNew(priv->conn, 0))) { vshError(ctl, _("cannot create a new stream")); goto cleanup; } - if (virStorageVolUpload(vol, st, offset, length, 0) < 0) { + if (virStorageVolUpload(vol, st, offset, length, flags) < 0) { vshError(ctl, _("cannot upload to volume %s"), name); goto cleanup; } - if (virStreamSendAll(st, cmdVolUploadSource, &fd) < 0) { - vshError(ctl, _("cannot send data to volume %s"), name); - goto cleanup; + if (flags & VIR_STORAGE_VOL_UPLOAD_SPARSE_STREAM) { + if (virStreamSparseSendAll(st, virshStreamSource, + virshStreamInData, &cbData) < 0) { + vshError(ctl, _("cannot send data to volume %s"), name); + goto cleanup; + } + } else { + if (virStreamSendAll(st, virshStreamSource, &cbData) < 0) { + vshError(ctl, _("cannot send data to volume %s"), name); + goto cleanup; + } } if (VIR_CLOSE(fd) < 0) { diff --git a/tools/virsh.c b/tools/virsh.c index 080d675..18e899e 100644 --- a/tools/virsh.c +++ b/tools/virsh.c @@ -263,6 +263,17 @@ int virshStreamSink(virStreamPtr st ATTRIBUTE_UNUSED, return safewrite(*fd, bytes, nbytes); } +int +virshStreamSource(virStreamPtr st ATTRIBUTE_UNUSED, + char *bytes, size_t nbytes, void *opaque) +{ + virshStreamCallbackDataPtr cbData = opaque; + int fd = cbData->fd; + + return saferead(fd, bytes, nbytes); +} + + int virshStreamSkip(virStreamPtr st ATTRIBUTE_UNUSED, unsigned long long offset, void *opaque) { @@ -271,6 +282,23 @@ int virshStreamSkip(virStreamPtr st ATTRIBUTE_UNUSED, return lseek(*fd, offset, SEEK_CUR) == (off_t) -1 ? -1 : 0; } +int virshStreamInData(virStreamPtr st ATTRIBUTE_UNUSED, + int *inData, + unsigned long long *offset, + void *opaque) +{ + virshStreamCallbackDataPtr cbData = opaque; + vshControl *ctl = cbData->ctl; + int fd = cbData->fd; + int ret; + + if ((ret = virFileInData(fd, inData, offset)) < 0) + vshError(ctl, "%s", _("Unable to get current position in stream")); + + return ret; +} + + /* --------------- * Command Connect * --------------- diff --git a/tools/virsh.h b/tools/virsh.h index 5c382ef..1362819 100644 --- a/tools/virsh.h +++ b/tools/virsh.h @@ -150,7 +150,21 @@ int virshDomainState(vshControl *ctl, virDomainPtr dom, int *reason); int virshStreamSink(virStreamPtr st, const char *bytes, size_t nbytes, void *opaque); +int virshStreamSource(virStreamPtr st, + char *bytes, size_t nbytes, void *opaque); + int virshStreamSkip(virStreamPtr st, unsigned long long offset, void *opaque); +typedef struct _virshStreamCallbackData virshStreamCallbackData; +typedef virshStreamCallbackData *virshStreamCallbackDataPtr; +struct _virshStreamCallbackData { + vshControl *ctl; + int fd; +}; + +int virshStreamInData(virStreamPtr st, + int *data, + unsigned long long *offset, + void *opaque); #endif /* VIRSH_H */ diff --git a/tools/virsh.pod b/tools/virsh.pod index e04921d..ba5567f 100644 --- a/tools/virsh.pod +++ b/tools/virsh.pod @@ -3609,13 +3609,14 @@ the storage volume should be deleted as well. Not all storage drivers support this option, presently only rbd. =item B<vol-upload> [I<--pool> I<pool-or-uuid>] [I<--offset> I<bytes>] -[I<--length> I<bytes>] I<vol-name-or-key-or-path> I<local-file> +[I<--length> I<bytes>] [I<--sparse>] I<vol-name-or-key-or-path> I<local-file> Upload the contents of I<local-file> to a storage volume. I<--pool> I<pool-or-uuid> is the name or UUID of the storage pool the volume is in. I<vol-name-or-key-or-path> is the name or key or path of the volume where the file will be uploaded. +If I<--sparse> is specified, this command will preserve volume sparseness. I<--offset> is the position in the storage volume at which to start writing the data. The value must be 0 or larger. I<--length> is an upper bound of the amount of data to be uploaded. A negative value is interpreted -- 2.8.4

Now that we have everything prepared, let's enable the feature for these two APIs. Well, except we can't. Not just yet. Because of the previous commit. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/storage/storage_backend.c | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/storage/storage_backend.c b/src/storage/storage_backend.c index d041530..19d5cd5 100644 --- a/src/storage/storage_backend.c +++ b/src/storage/storage_backend.c @@ -2030,7 +2030,8 @@ virStorageBackendVolUploadLocal(virConnectPtr conn ATTRIBUTE_UNUSED, int ret = -1; int has_snap = 0; - virCheckFlags(0, -1); + virCheckFlags(VIR_STORAGE_VOL_UPLOAD_SPARSE_STREAM, -1); + /* if volume has target format VIR_STORAGE_FILE_PLOOP * we need to restore DiskDescriptor.xml, according to * new contents of volume. This operation will be perfomed @@ -2076,7 +2077,8 @@ virStorageBackendVolDownloadLocal(virConnectPtr conn ATTRIBUTE_UNUSED, int ret = -1; int has_snap = 0; - virCheckFlags(0, -1); + virCheckFlags(VIR_STORAGE_VOL_DOWNLOAD_SPARSE_STREAM, -1); + if (vol->target.format == VIR_STORAGE_FILE_PLOOP) { has_snap = virStorageBackendPloopHasSnapshots(vol->target.path); if (has_snap < 0) { -- 2.8.4

As described in the previous commit, we need to handle communication with iohelper differently in some cases. This patch creates function stubs for that. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/fdstream.c | 83 +++++++++++++++++++++++++++++++++++++++++++++++++--------- 1 file changed, 71 insertions(+), 12 deletions(-) diff --git a/src/fdstream.c b/src/fdstream.c index 911fac0..8ab7cd5 100644 --- a/src/fdstream.c +++ b/src/fdstream.c @@ -399,6 +399,21 @@ virFDStreamAbort(virStreamPtr st) return virFDStreamCloseInt(st, true); } +static ssize_t +virFDStreamWriteInternal(virFDStreamDataPtr fdst, + const char *bytes, + size_t nbytes) +{ + if (fdst->formatted) { + virReportError(VIR_ERR_INTERNAL_ERROR, "%s", + _("sparse stream not supported")); + return -1; + } else { + return write(fdst->fd, bytes, nbytes); + } +} + + static int virFDStreamWrite(virStreamPtr st, const char *bytes, size_t nbytes) { virFDStreamDataPtr fdst = st->privateData; @@ -431,7 +446,7 @@ static int virFDStreamWrite(virStreamPtr st, const char *bytes, size_t nbytes) } retry: - ret = write(fdst->fd, bytes, nbytes); + ret = virFDStreamWriteInternal(fdst, bytes, nbytes); if (ret < 0) { VIR_WARNINGS_NO_WLOGICALOP_EQUAL_EXPR if (errno == EAGAIN || errno == EWOULDBLOCK) { @@ -453,6 +468,21 @@ static int virFDStreamWrite(virStreamPtr st, const char *bytes, size_t nbytes) } +static ssize_t +virFDStreamReadInternal(virFDStreamDataPtr fdst, + char *bytes, + size_t nbytes) +{ + if (fdst->formatted) { + virReportError(VIR_ERR_INTERNAL_ERROR, "%s", + _("sparse stream not supported")); + return -1; + } else { + return read(fdst->fd, bytes, nbytes); + } +} + + static int virFDStreamRead(virStreamPtr st, char *bytes, size_t nbytes) { virFDStreamDataPtr fdst = st->privateData; @@ -483,7 +513,7 @@ static int virFDStreamRead(virStreamPtr st, char *bytes, size_t nbytes) } retry: - ret = read(fdst->fd, bytes, nbytes); + ret = virFDStreamReadInternal(fdst, bytes, nbytes); if (ret < 0) { VIR_WARNINGS_NO_WLOGICALOP_EQUAL_EXPR if (errno == EAGAIN || errno == EWOULDBLOCK) { @@ -506,11 +536,26 @@ static int virFDStreamRead(virStreamPtr st, char *bytes, size_t nbytes) static int +virFDStreamSkipInternal(virFDStreamDataPtr fdst, + unsigned long long length) +{ + off_t off; + if (fdst->formatted) { + virReportError(VIR_ERR_INTERNAL_ERROR, "%s", + _("sparse stream not supported")); + return -1; + } else { + off = lseek(fdst->fd, length, SEEK_CUR); + return off == (off_t) -1 ? -1 : 0; + } +} + + +static int virFDStreamSkip(virStreamPtr st, unsigned long long length) { virFDStreamDataPtr fdst = st->privateData; - off_t off; int ret = -1; virObjectLock(fdst); @@ -520,15 +565,13 @@ virFDStreamSkip(virStreamPtr st, _("cannot write to stream")); goto cleanup; } + } + + if (virFDStreamSkipInternal(fdst, length) < 0) + goto cleanup; + + if (fdst->length) fdst->offset += length; - } - - off = lseek(fdst->fd, length, SEEK_CUR); - if (off == (off_t) -1) { - virReportSystemError(errno, "%s", - _("unable to seek")); - goto cleanup; - } ret = 0; cleanup: virObjectUnlock(fdst); @@ -536,6 +579,22 @@ virFDStreamSkip(virStreamPtr st, } + +static int +virFDStreamInDataInternal(virFDStreamDataPtr fdst, + int *inData, + unsigned long long *length) +{ + if (fdst->formatted) { + virReportError(VIR_ERR_INTERNAL_ERROR, "%s", + _("sparse stream not supported")); + return -1; + } else { + return virFileInData(fdst->fd, inData, length); + } +} + + static int virFDStreamInData(virStreamPtr st, int *inData, @@ -545,7 +604,7 @@ virFDStreamInData(virStreamPtr st, int ret = -1; virObjectLock(fdst); - ret = virFileInData(fdst->fd, inData, length); + ret = virFDStreamInDataInternal(fdst, inData, length); virObjectUnlock(fdst); return ret; } -- 2.8.4

And switch fdstream to call them. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- po/POTFILES.in | 1 + src/Makefile.am | 5 ++- src/fdstream.c | 29 +++++++------ src/iohelper/iohelper_message.c | 94 +++++++++++++++++++++++++++++++++++++++++ src/iohelper/iohelper_message.h | 26 ++++++++++++ 5 files changed, 139 insertions(+), 16 deletions(-) diff --git a/po/POTFILES.in b/po/POTFILES.in index 9f4866c..7f40200 100644 --- a/po/POTFILES.in +++ b/po/POTFILES.in @@ -64,6 +64,7 @@ src/interface/interface_backend_netcf.c src/interface/interface_backend_udev.c src/internal.h src/iohelper/iohelper.c +src/iohelper/iohelper_message.c src/libvirt-admin.c src/libvirt-domain-snapshot.c src/libvirt-domain.c diff --git a/src/Makefile.am b/src/Makefile.am index af5ef72..1cce603 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -1170,9 +1170,10 @@ libvirt_driver_la_SOURCES = $(DRIVER_SOURCES) libvirt_driver_la_CFLAGS = \ $(GNUTLS_CFLAGS) $(CURL_CFLAGS) \ - -I$(srcdir)/conf $(AM_CFLAGS) + -I$(srcdir)/conf -I$(srcdir)/iohelper $(AM_CFLAGS) libvirt_driver_la_LIBADD = \ - $(GNUTLS_LIBS) $(CURL_LIBS) $(DLOPEN_LIBS) + $(GNUTLS_LIBS) $(CURL_LIBS) $(DLOPEN_LIBS) \ + libvirt-iohelper.la # All .syms files should be placed in exactly one of these three lists, # depending on whether they are stored in git and/or used in the build. diff --git a/src/fdstream.c b/src/fdstream.c index 8ab7cd5..dc41164 100644 --- a/src/fdstream.c +++ b/src/fdstream.c @@ -44,6 +44,7 @@ #include "virstring.h" #include "virtime.h" #include "virprocess.h" +#include "iohelper_message.h" #define VIR_FROM_THIS VIR_FROM_STREAMS @@ -61,6 +62,7 @@ struct virFDStreamData { unsigned long long offset; unsigned long long length; bool formatted; /* True if formatted messages are read from/written to @fd. */ + iohelperCtlPtr ioCtl; int watch; int events; /* events the stream callback is subscribed for */ @@ -89,7 +91,7 @@ virFDStreamDataDispose(void *obj) { virFDStreamDataPtr fdst = obj; - VIR_DEBUG("obj=%p", fdst); + virObjectUnref(fdst->ioCtl); } static int virFDStreamDataOnceInit(void) @@ -405,9 +407,7 @@ virFDStreamWriteInternal(virFDStreamDataPtr fdst, size_t nbytes) { if (fdst->formatted) { - virReportError(VIR_ERR_INTERNAL_ERROR, "%s", - _("sparse stream not supported")); - return -1; + return iohelperWrite(fdst->ioCtl, bytes, nbytes); } else { return write(fdst->fd, bytes, nbytes); } @@ -474,9 +474,7 @@ virFDStreamReadInternal(virFDStreamDataPtr fdst, size_t nbytes) { if (fdst->formatted) { - virReportError(VIR_ERR_INTERNAL_ERROR, "%s", - _("sparse stream not supported")); - return -1; + return iohelperRead(fdst->ioCtl, bytes, nbytes); } else { return read(fdst->fd, bytes, nbytes); } @@ -541,9 +539,7 @@ virFDStreamSkipInternal(virFDStreamDataPtr fdst, { off_t off; if (fdst->formatted) { - virReportError(VIR_ERR_INTERNAL_ERROR, "%s", - _("sparse stream not supported")); - return -1; + return iohelperSkip(fdst->ioCtl, length); } else { off = lseek(fdst->fd, length, SEEK_CUR); return off == (off_t) -1 ? -1 : 0; @@ -586,9 +582,7 @@ virFDStreamInDataInternal(virFDStreamDataPtr fdst, unsigned long long *length) { if (fdst->formatted) { - virReportError(VIR_ERR_INTERNAL_ERROR, "%s", - _("sparse stream not supported")); - return -1; + return iohelperInData(fdst->ioCtl, inData, length); } else { return virFileInData(fdst->fd, inData, length); } @@ -651,11 +645,18 @@ static int virFDStreamOpenInternal(virStreamPtr st, fdst->errfd = errfd; fdst->length = length; fdst->formatted = formatted; + if (formatted && + !(fdst->ioCtl = iohelperCtlNew(fd))) + goto error; + st->driver = &virFDStreamDrv; st->privateData = fdst; - return 0; + + error: + virObjectUnref(fdst); + return -1; } diff --git a/src/iohelper/iohelper_message.c b/src/iohelper/iohelper_message.c index 54b9355..51b283d 100644 --- a/src/iohelper/iohelper_message.c +++ b/src/iohelper/iohelper_message.c @@ -22,3 +22,97 @@ #include <config.h> #include "iohelper_message.h" +#include "virobject.h" +#include "virlog.h" + +#define VIR_FROM_THIS VIR_FROM_STREAMS + +VIR_LOG_INIT("iohelperCtl"); + +struct iohelperCtl { + virObject parent; + + int fd; +}; + +static virClassPtr iohelperCtlClass; + +static void +iohelperCtlDispose(void *obj) +{ + iohelperCtlPtr ctl = obj; + + VIR_DEBUG("obj = %p", ctl); +} + +static int iohelperCtlOnceInit(void) +{ + if (!(iohelperCtlClass = virClassNew(virClassForObject(), + "iohelperCtl", + sizeof(iohelperCtl), + iohelperCtlDispose))) + return -1; + + return 0; +} + +VIR_ONCE_GLOBAL_INIT(iohelperCtl) + +iohelperCtlPtr +iohelperCtlNew(int fd) +{ + iohelperCtlPtr ret; + + if (iohelperCtlInitialize() < 0) + return NULL; + + if (!(ret = virObjectNew(iohelperCtlClass))) + return NULL; + + ret->fd = fd; + + return ret; +} + + +ssize_t +iohelperRead(iohelperCtlPtr ctl ATTRIBUTE_UNUSED, + char *bytes ATTRIBUTE_UNUSED, + size_t nbytes ATTRIBUTE_UNUSED) +{ + virReportError(VIR_ERR_INTERNAL_ERROR, "%s", + _("sparse stream not supported")); + return -1; +} + + +ssize_t +iohelperWrite(iohelperCtlPtr ctl ATTRIBUTE_UNUSED, + const char *bytes ATTRIBUTE_UNUSED, + size_t nbytes ATTRIBUTE_UNUSED) +{ + virReportError(VIR_ERR_INTERNAL_ERROR, "%s", + _("sparse stream not supported")); + return -1; +} + + +int +iohelperSkip(iohelperCtlPtr ctl ATTRIBUTE_UNUSED, + unsigned long long length ATTRIBUTE_UNUSED) +{ + virReportError(VIR_ERR_INTERNAL_ERROR, "%s", + _("sparse stream not supported")); + return -1; +} + + +int +iohelperInData(iohelperCtlPtr ctl ATTRIBUTE_UNUSED, + int *inData ATTRIBUTE_UNUSED, + unsigned long long *length ATTRIBUTE_UNUSED) +{ + virReportError(VIR_ERR_INTERNAL_ERROR, "%s", + _("sparse stream not supported")); + return -1; +} diff --git a/src/iohelper/iohelper_message.h b/src/iohelper/iohelper_message.h index 791a645..74afd49 100644 --- a/src/iohelper/iohelper_message.h +++ b/src/iohelper/iohelper_message.h @@ -23,4 +23,30 @@ #ifndef __VIR_IOHELPER_MESSAGE_H__ # define __VIR_IOHELPER_MESSAGE_H__ +# include "internal.h" + +typedef struct iohelperCtl iohelperCtl; +typedef iohelperCtl *iohelperCtlPtr; + +iohelperCtlPtr iohelperCtlNew(int fd); + +ssize_t +iohelperRead(iohelperCtlPtr ctl, + char *bytes, + size_t nbytes); + +ssize_t +iohelperWrite(iohelperCtlPtr ctl, + const char *bytes, + size_t nbytes); + +int +iohelperSkip(iohelperCtlPtr ctl, + unsigned long long length); + +int +iohelperInData(iohelperCtlPtr ctl, + int *inData, + unsigned long long *length); + #endif /* __VIR_IOHELPER_H__ */ -- 2.8.4

This is nearly the same story as formatted read. With one exception. Due to the recurring pattern of calling write functions in our code: repeat { nwritten = write(dest, buf + offset, buflen - offset); offset += nwritten; } we have to be cautious about the return value of iohelperWrite(). If we would return number of bytes partially written, we would be called again with an offset shifted. But at the beginning of iohelperWrite() we flush the output queue so the remaining data is written then. Moreover, there's no way for us to determine whether this is the case or we are called with fresh data, completely unrelated to the first call. Therefore, we must keep claiming EAGAIN, even though the message is being partially written, and just on the last iteration claim success and return the size of data we were requested to write in the first place. Even if this means that in the last iteration just one byte was written to make the message write complete. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/iohelper/iohelper_message.c | 105 ++++++++++++++++++++++++++++++++++++++-- 1 file changed, 100 insertions(+), 5 deletions(-) diff --git a/src/iohelper/iohelper_message.c b/src/iohelper/iohelper_message.c index fe2304b..d900c2f 100644 --- a/src/iohelper/iohelper_message.c +++ b/src/iohelper/iohelper_message.c @@ -39,9 +39,11 @@ struct iohelperCtl { bool blocking; virNetMessagePtr msg; bool msgReadyRead; + bool msgReadyWrite; }; typedef ssize_t (*readfunc)(int fd, void *buf, size_t count); +typedef ssize_t (*writefunc)(int fd, const void *buf, size_t count); static virClassPtr iohelperCtlClass; @@ -84,6 +86,7 @@ iohelperCtlNew(int fd, ret->fd = fd; ret->blocking = blocking; ret->msgReadyRead = false; + ret->msgReadyWrite = true; return ret; @@ -98,6 +101,7 @@ messageClear(iohelperCtlPtr ctl) { virNetMessageClear(ctl->msg); ctl->msgReadyRead = false; + ctl->msgReadyWrite = true; } @@ -107,6 +111,11 @@ messageReadyRead(iohelperCtlPtr ctl) return ctl->msgReadyRead; } +static inline bool +messageReadyWrite(iohelperCtlPtr ctl) +{ + return ctl->msgReadyWrite; +} static ssize_t messageRecv(iohelperCtlPtr ctl) @@ -165,6 +174,47 @@ messageRecv(iohelperCtlPtr ctl) } +static ssize_t +messageSend(iohelperCtlPtr ctl) +{ + virNetMessagePtr msg = ctl->msg; + writefunc writeF = ctl->blocking ? safewrite : write; + + ctl->msgReadyWrite = false; + + while (true) { + ssize_t nwritten; + size_t want; + + want = msg->bufferLength - msg->bufferOffset; + + rewrite: + errno = 0; + nwritten = writeF(ctl->fd, + msg->buffer + msg->bufferOffset, + want); + + if (nwritten < 0) { + if (errno == EINTR) + goto rewrite; + if (errno == EAGAIN) + return 0; + return -1; + } else if (nwritten == 0) { + /* EOF while writing */ + return 0; + } else { + msg->bufferOffset += nwritten; + } + + if (msg->bufferOffset == msg->bufferLength) { + ctl->msgReadyWrite = true; + return msg->bufferLength; + } + } +} + + ssize_t iohelperRead(iohelperCtlPtr ctl, char *bytes, @@ -206,12 +256,57 @@ iohelperRead(iohelperCtlPtr ctl, ssize_t -iohelperWrite(iohelperCtlPtr ctl ATTRIBUTE_UNUSED, - const char *bytes ATTRIBUTE_UNUSED, - size_t nbytes ATTRIBUTE_UNUSED) +iohelperWrite(iohelperCtlPtr ctl, + const char *bytes, + size_t nbytes) { - virReportError(VIR_ERR_INTERNAL_ERROR, "%s", - _("sparse stream not supported")); + size_t headerLen; + ssize_t nwritten, totalNwritten = 0; + + virNetMessagePtr msg = ctl->msg; + + if (!messageReadyWrite(ctl)) { + /* Okay, the outgoing message is not fully sent. Try to + * finish the sending and recheck. */ + if ((nwritten = messageSend(ctl)) < 0) + return -1; + + if (!nwritten && errno != EAGAIN) + return 0; + + if (!messageReadyWrite(ctl)) { + errno = EAGAIN; + return -2; + } + + totalNwritten += nwritten; + } + + memset(&msg->header, 0, sizeof(msg->header)); + msg->header.type = VIR_NET_STREAM; + msg->header.status = nbytes ? VIR_NET_CONTINUE : VIR_NET_OK; + + /* Encoding a message is fatal and we should discard any + * partially encoded message. */ + if (virNetMessageEncodeHeader(msg) < 0) + goto error; + + headerLen = msg->bufferOffset; + + if (virNetMessageEncodePayloadRaw(msg, bytes, nbytes) < 0) + goto error; + + /* At this point, the message is successfully encoded. Don't + * discard it if something below fails. */ + if ((nwritten = messageSend(ctl)) < 0) + return -1; + + totalNwritten += nwritten - headerLen; + + return totalNwritten; + + error: + messageClear(ctl); return -1; } -- 2.8.4

Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- tests/iohelpermessagetest.c | 295 +++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 292 insertions(+), 3 deletions(-) diff --git a/tests/iohelpermessagetest.c b/tests/iohelpermessagetest.c index 293c107..8c83261 100644 --- a/tests/iohelpermessagetest.c +++ b/tests/iohelpermessagetest.c @@ -27,6 +27,7 @@ #include "internal.h" #include "iohelper_message.h" #include "virlog.h" +#include "virtime.h" VIR_LOG_INIT("tests.iohelpermessagetest"); @@ -35,6 +36,8 @@ VIR_LOG_INIT("tests.iohelpermessagetest"); typedef struct { const char * const *msg; unsigned int *len; + bool blockR; + bool blockW; } testData; typedef testData *testDataPtr; @@ -77,7 +80,7 @@ testBlocking(const void *opaque) char *genMsg = NULL; char buf[1024]; - if (testInit(ioCtl, pipeFD, true, true) < 0) + if (testInit(ioCtl, pipeFD, data->blockR, data->blockW) < 0) goto cleanup; while (!quit) { @@ -143,15 +146,253 @@ testBlocking(const void *opaque) return ret; } +typedef struct { + virMutexPtr lock; + virCondPtr cond; + bool finished; + int ret; + iohelperCtlPtr ioCtl; + char *msg; + size_t msgLen; +} threadData; + +typedef threadData *threadDataPtr; + +static void +readerThread(void *opaque) +{ + threadDataPtr data = opaque; + char *bigBuf = NULL; + size_t bigBufSize = 0; + + virObjectRef(data->ioCtl); + /* Sleep some random time to simulate out of sync read & + * write */ + usleep((rand() % 100) * 1000); + + while (true) { + char buf[10]; /* Simulate reads of small chunks of data */ + ssize_t nread; + + reread: + nread = iohelperRead(data->ioCtl, buf, sizeof(buf)); + if (nread < 0) { + if (errno == EAGAIN) { + usleep(20 * 1000); + goto reread; + } + + virFilePrintf(stderr, "Unable to read message (errno=%d)\n", errno); + goto cleanup; + } + + if (!nread) + break; + + if (VIR_REALLOC_N(bigBuf, bigBufSize + nread) < 0) + goto cleanup; + + memcpy(bigBuf + bigBufSize, buf, nread); + bigBufSize += nread; + } + + if (bigBufSize != data->msgLen) { + virFilePrintf(stderr, "Message length mismatch: expected %zu got %zu", + data->msgLen, bigBufSize); + goto cleanup; + } + + if (memcmp(bigBuf, data->msg, data->msgLen)) { + virFilePrintf(stderr, "Mismatched data"); + goto cleanup; + } + + data->ret = 0; + + cleanup: + VIR_FREE(bigBuf); + virObjectUnref(data->ioCtl); + virMutexLock(data->lock); + data->finished = true; + virCondSignal(data->cond); + virMutexUnlock(data->lock); +} + +static void +writerThread(void *opaque ATTRIBUTE_UNUSED) +{ + threadDataPtr data = opaque; + size_t writeOff = 0; + + virObjectRef(data->ioCtl); + /* Sleep some random time to simulate out of sync read & + * write */ + usleep((rand() % 100) * 1000); + + while (true) { + ssize_t nwritten; + size_t want = data->msgLen - writeOff; + + if (!want) + break; + + rewrite: + nwritten = iohelperWrite(data->ioCtl, + data->msg + writeOff, + want); + + if (nwritten < 0) { + if (errno == EAGAIN) { + usleep(20 * 1000); + goto rewrite; + } + + virFilePrintf(stderr, "Unable to write message (errno=%d)\n", errno); + goto cleanup; + } + + if (!nwritten) + break; + + writeOff += nwritten; + } + + if (writeOff != data->msgLen) { + virFilePrintf(stderr, "Message length mismatch: expected %zu written %zu", + data->msgLen, writeOff); + goto cleanup; + } + + data->ret = 0; + + cleanup: + virObjectUnref(data->ioCtl); + virMutexLock(data->lock); + data->finished = true; + virCondSignal(data->cond); + virMutexUnlock(data->lock); +} + +/* How long wait (in ms) for both reader & writer + * threads to finish? */ +#define WAIT_TIME 10000 + +static int +testNonblocking(const void *opaque) +{ + int ret = -1; + const testData *data = opaque; + iohelperCtlPtr ioCtl[2] = {NULL, NULL}; + int pipeFD[2] = {-1, -1}; + virThread reader, writer; + threadData readerD, writerD; + virMutex lock; + virCond cond; + unsigned long long now; + unsigned long long then; + char *msg = NULL; + size_t msgLen = 0, idx; + + for (idx = 0; data->msg && data->msg[idx]; idx++) { + const char *tmp = data->msg[idx]; + size_t tmpLen = strlen(tmp); + + if (VIR_REALLOC_N(msg, msgLen + tmpLen + 1) < 0) + goto cleanup; + + memcpy(msg + msgLen, tmp, tmpLen + 1); + msgLen += tmpLen; + } + + for (idx = 0; data->len && data->len[idx]; idx++) { + size_t tmpLen = data->len[idx]; + + if (VIR_REALLOC_N(msg, msgLen + tmpLen) < 0) + goto cleanup; + msgLen += tmpLen; + + /* Here @msg contains some garbage that was on the heap + * when the memory was allocated. That's okay, we want to + * be sure iohelper can deal with binary garbage. */ + } + + if (virMutexInit(&lock) < 0 || + virCondInit(&cond) < 0) + goto cleanup; + + if (testInit(ioCtl, pipeFD, data->blockR, data->blockW) < 0) + goto cleanup; + + readerD = writerD = (threadData) {.lock = &lock, .cond = &cond, + .ret = -1, .finished = false, .msg = msg, msgLen = msgLen}; + readerD.ioCtl = ioCtl[0]; + writerD.ioCtl = ioCtl[1]; + + /* Now, ideally we would set the kernel's pipe buffer to be + * small. Really small. Couple of bytes perhaps so that we + * can be sure writes wrap around it just nicely. But the + * smallest possible size is PAGESIZE. Trying to set anything + * smaller than that is silently rounded up to PAGESIZE. + * Okay, in that case we should write multiple of that. */ + fcntl(pipeFD[0], F_SETPIPE_SZ, 0); + + virMutexLock(&lock); + + if (virThreadCreate(&reader, false, readerThread, &readerD) < 0 || + virThreadCreate(&writer, false, writerThread, &writerD) < 0) + goto cleanup; + + if (virTimeMillisNow(&now) < 0) + goto cleanup; + + then = now + WAIT_TIME; + + while (!readerD.finished || + !writerD.finished) { + if (virCondWaitUntil(&cond, &lock, then) < 0) { + if (errno == ETIMEDOUT) { + if (!readerD.finished) + virThreadCancel(&reader); + if (!writerD.finished) + virThreadCancel(&writer); + } + + goto cleanup; + } + if (readerD.finished) + VIR_FORCE_CLOSE(pipeFD[0]); + if (writerD.finished) + VIR_FORCE_CLOSE(pipeFD[1]); + } + + if (readerD.ret < 0 || + writerD.ret < 0) + goto cleanup; + + ret = 0; + cleanup: + virMutexUnlock(&lock); + virMutexDestroy(&lock); + virCondDestroy(&cond); + virObjectUnref(ioCtl[0]); + virObjectUnref(ioCtl[1]); + VIR_FORCE_CLOSE(pipeFD[0]); + VIR_FORCE_CLOSE(pipeFD[1]); + return ret; +} + static int mymain(void) { int ret = 0; + srand(time(NULL)); + #define DO_TEST_BLOCKING_SIMPLE(...) \ do { \ const char *msg[] = { __VA_ARGS__, NULL}; \ - testData data = {.msg = msg, .len = NULL }; \ + testData data = {.blockR = true, .blockW = true, \ + .msg = msg, .len = NULL }; \ if (virTestRun("Blocking simple", testBlocking, &data) < 0) \ ret = -1; \ } while (0) @@ -159,11 +400,47 @@ mymain(void) #define DO_TEST_BLOCKING_LEN(...) \ do { \ unsigned int len[] = { __VA_ARGS__, 0}; \ - testData data = {.msg = NULL, .len = len }; \ + testData data = {.blockR = true, .blockW = true, \ + .msg = NULL, .len = len }; \ if (virTestRun("Blocking len", testBlocking, &data) < 0) \ ret = -1; \ } while (0) +#define DO_TEST_BLOCKR_SIMPLE(...) \ + do { \ + const char *msg[] = { __VA_ARGS__, NULL}; \ + testData data = {.blockR = true, .blockW = false, \ + .msg = msg, .len = NULL }; \ + if (virTestRun("Blocking read simple", testNonblocking, &data) < 0) \ + ret = -1; \ + } while (0) + +#define DO_TEST_BLOCKR_LEN(...) \ + do { \ + unsigned int len[] = { __VA_ARGS__, 0}; \ + testData data = {.blockR = true, .blockW = false, \ + .msg = NULL, .len = len }; \ + if (virTestRun("Blocking read len", testNonblocking, &data) < 0) \ + ret = -1; \ + } while (0) + +#define DO_TEST_BLOCKW_SIMPLE(...) \ + do { \ + const char *msg[] = { __VA_ARGS__, NULL}; \ + testData data = {.blockR = false, .blockW = true, \ + .msg = msg, .len = NULL }; \ + if (virTestRun("Blocking write simple", testNonblocking, &data) < 0) \ + ret = -1; \ + } while (0) +#define DO_TEST_BLOCKW_LEN(...) \ + do { \ + unsigned int len[] = { __VA_ARGS__, 0}; \ + testData data = {.blockR = false, .blockW = true, \ + .msg = NULL, .len = len }; \ + if (virTestRun("Blocking write len", testNonblocking, &data) < 0) \ + ret = -1; \ + } while (0) + DO_TEST_BLOCKING_SIMPLE("Hello world"); DO_TEST_BLOCKING_SIMPLE("Hello world", "Hello", "world"); @@ -171,6 +448,18 @@ mymain(void) DO_TEST_BLOCKING_LEN(1024); DO_TEST_BLOCKING_LEN(32, 64, 128, 512, 1024); + DO_TEST_BLOCKR_SIMPLE("Hello world"); + DO_TEST_BLOCKR_SIMPLE("Hello world", "Hello", "world"); + + DO_TEST_BLOCKR_LEN(1024); + DO_TEST_BLOCKR_LEN(409600); + + DO_TEST_BLOCKW_SIMPLE("Hello world"); + DO_TEST_BLOCKW_SIMPLE("Hello world", "Hello", "world"); + + DO_TEST_BLOCKW_LEN(1024); + DO_TEST_BLOCKW_LEN(409600); + return ret == 0 ? EXIT_SUCCESS : EXIT_FAILURE; } -- 2.8.4

The runIO function is after all initialization (e.g. argument parsing, file opening, ...) is done. It is the heart of the iohelper. Unfortunately, approach implemented in next patches is not easy to be combined with O_DIRECT. Therefore, rename runIO to runIOBasic and create new runIO, which for now just wraps over runIOBasic. But in not so distant future it will decide which IO loop is going to be called. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/iohelper/iohelper.c | 22 ++++++++++++++++++++-- 1 file changed, 20 insertions(+), 2 deletions(-) diff --git a/src/iohelper/iohelper.c b/src/iohelper/iohelper.c index 00f31e7..02fc2ef 100644 --- a/src/iohelper/iohelper.c +++ b/src/iohelper/iohelper.c @@ -73,7 +73,7 @@ prepare(const char *path, int oflags, int mode, } static int -runIO(const char *path, int fd, int oflags, unsigned long long length) +runIOBasic(const char *path, int fd, int oflags, unsigned long long length) { void *base = NULL; /* Location to be freed */ char *buf = NULL; /* Aligned location within base */ @@ -199,6 +199,23 @@ runIO(const char *path, int fd, int oflags, unsigned long long length) return ret; } + +static int +runIO(const char *path, int fd, int oflags, + unsigned long long length, bool sparse) +{ + bool direct = O_DIRECT && ((oflags & O_DIRECT) != 0); + + /* Right now, our implementation do not support O_DIRECT and sparse. */ + if (direct && sparse) { + virReportError(VIR_ERR_INTERNAL_ERROR, "%s", + _("O_DIRECT and sparse streams is not supported at once")); + return -1; + } + + return runIOBasic(path, fd, oflags, length); +} + static const char *program_name; ATTRIBUTE_NORETURN static void @@ -225,6 +242,7 @@ main(int argc, char **argv) unsigned int delete = 0; int fd = -1; int lengthIndex = 0; + bool sparse = false; program_name = argv[0]; @@ -293,7 +311,7 @@ main(int argc, char **argv) exit(EXIT_FAILURE); } - if (fd < 0 || runIO(path, fd, oflags, length) < 0) + if (fd < 0 || runIO(path, fd, oflags, length, sparse) < 0) goto error; if (delete) -- 2.8.4

If formatted messages are required, iohelper is spawned with 'sparse' argument. Teach it to actually deal with it. Signed-off-by: Michal Privoznik <mprivozn(a)redhat.com&gt; --- src/iohelper/iohelper.c | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/iohelper/iohelper.c b/src/iohelper/iohelper.c index 4224d97..65fbc63 100644 --- a/src/iohelper/iohelper.c +++ b/src/iohelper/iohelper.c @@ -252,7 +252,7 @@ usage(int status) fprintf(stderr, _("%s: try --help for more details"), program_name); } else { printf(_("Usage: %s FILENAME OFLAGS MODE OFFSET LENGTH DELETE\n" - " or: %s FILENAME LENGTH FD\n"), + " or: %s FILENAME LENGTH FD [sparse]\n"), program_name, program_name); } exit(status); @@ -307,13 +307,15 @@ main(int argc, char **argv) exit(EXIT_FAILURE); } fd = prepare(path, oflags, mode, offset); - } else if (argc == 4) { /* FILENAME LENGTH FD */ + } else if (argc == 4 || argc == 5) { /* FILENAME LENGTH FD [sparse] */ lengthIndex = 2; if (virStrToLong_i(argv[3], NULL, 10, &fd) < 0) { fprintf(stderr, _("%s: malformed fd %s"), program_name, argv[3]); exit(EXIT_FAILURE); } + if (argc == 5) + sparse = STREQ(argv[4], "sparse"); #ifdef F_GETFL oflags = fcntl(fd, F_GETFL); #else -- 2.8.4