External/libvirt
1
0
Fork 0
mirror of https://gitlab.com/libvirt/libvirt.git synced 2025-02-02 09:55:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

virNetClientStream: Wire up VIR_NET_STREAM_HOLE

Browse Source 
Whenever server sends a client stream packet (either regular with
actual data or stream skip one) it is queued on @st->rx. So the
list is a mixture of both types of stream packets. So now that we
have all the helpers needed we can wire their processing up. But
since virNetClientStreamRecvPacket doesn't support
VIR_STREAM_RECV_STOP_AT_HOLE flag yet, let's turn all received
skips into zeroes repeating requested times.

Signed-off-by: Michal Privoznik <mprivozn@redhat.com>
This commit is contained in:
Michal Privoznik 2016-05-20 16:35:13 +02:00
parent d6e5347ce3
commit 022705b81f
1 changed files with 43 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
src/rpc/virnetclientstream.c
View File

@ -296,6 +296,8 @@ int virNetClientStreamQueuePacket(virNetClientStreamPtr st,
virObjectLock(st); virObjectLock(st);
/* Don't distinguish VIR_NET_STREAM and VIR_NET_STREAM_SKIP
* here just yet. We want in order processing! */
virNetMessageQueuePush(&st->rx, tmp_msg); virNetMessageQueuePush(&st->rx, tmp_msg);
virNetClientStreamEventTimerUpdate(st); virNetClientStreamEventTimerUpdate(st);
@ -391,7 +393,7 @@ virNetClientStreamSetHole(virNetClientStreamPtr st,
* Returns: 0 on success, * Returns: 0 on success,
* -1 otherwise. * -1 otherwise.
*/ */
static int ATTRIBUTE_UNUSED static int
virNetClientStreamHandleHole(virNetClientPtr client, virNetClientStreamHandleHole(virNetClientPtr client,
virNetClientStreamPtr st) virNetClientStreamPtr st)
{ {
@ -469,6 +471,8 @@ int virNetClientStreamRecvPacket(virNetClientStreamPtr st,
virCheckFlags(0, -1); virCheckFlags(0, -1);
virObjectLock(st); virObjectLock(st);
reread:
if (!st->rx && !st->incomingEOF) { if (!st->rx && !st->incomingEOF) {
virNetMessagePtr msg; virNetMessagePtr msg;
int ret; int ret;
@ -500,8 +504,45 @@ int virNetClientStreamRecvPacket(virNetClientStreamPtr st,
} }
VIR_DEBUG("After IO rx=%p", st->rx); VIR_DEBUG("After IO rx=%p", st->rx);
if (st->rx &&
st->rx->header.type == VIR_NET_STREAM_HOLE &&
st->holeLength == 0) {
/* Handle skip sent to us by server. */
if (virNetClientStreamHandleHole(client, st) < 0)
goto cleanup;
}
if (!st->rx && !st->incomingEOF && st->holeLength == 0) {
if (nonblock) {
VIR_DEBUG("Non-blocking mode and no data available");
rv = -2;
goto cleanup;
}
/* We have consumed all packets from incoming queue but those
* were only skip packets, no data. Read the stream again. */
goto reread;
}
want = nbytes; want = nbytes;
while (want && st->rx) {
if (st->holeLength) {
/* Pretend holeLength zeroes was read from stream. */
size_t len = want;
if (len > st->holeLength)
len = st->holeLength;
memset(data, 0, len);
st->holeLength -= len;
want -= len;
}
while (want &&
st->rx &&
st->rx->header.type == VIR_NET_STREAM) {
virNetMessagePtr msg = st->rx; virNetMessagePtr msg = st->rx;
size_t len = want; size_t len = want;