[libvirt] [PATCH v2 30/38] virNetClientStream: Wire up VIR_NET_STREAM_SKIP

[John Ferlan]

On 05/09/2017 04:17 AM, Michal Privoznik wrote:
> On 05/05/2017 05:43 PM, John Ferlan wrote:
>>
>>
>> On 04/20/2017 06:01 AM, Michal Privoznik wrote:
>>> 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.
>>>
>>
>> Up to this point - I thought I had a good idea of what was going on, but
>> this patch loses me.
>>
>> I thought there was an "ordered" message queue to be processed...
>> ordered in so much as we're "reading" a file and sending either a 'data'
>> segment w/ data or a 'skip' segment with some number of bytes to skip.
>> Where it could be:
>>
>>    start...skip...data...[skip...data...]end
>>    start...data...skip...[data...skip...]end
>>    start...data...end
>>    start...skip...end
>>
>> So why does the code process and sum up the skips?
> 
> Because our streams are slightly more generic than just a file transfer.
> I mean, we use it for just any data transfer. Therefore it could be used
> even for this
> 
> start -> data -> skip -> skip -> data -> end.
> 
> You won't have this with regular files stored on a disk, but then again,
> virStream can be used (and is used) for generic data transfer.
> vol-download an vol-upload is just a subset of its usage.

True, but it's hard (yet) to think beyond that usage since that's the
reference being discussed during the series. Combining N consecutive
skips (or data for that matter) is an optimization, but does that
optimization make things more confusing?

If the sender sends data, data, skip, skip, skip - who cares - process
in the order your receive.  Combining for optimization is possible, but
at what cost?  You have to "peek" ahead and futz with stuff. The
optimization should come from the consumer. We can just process in the
order sent (or well, received).

> 
> Now, with that example, it makes sense to merge those skips into one
> big, doesn't it? But I agree, it looks a bit weird. I can drop the
> merging - it's not a performance issue. I was just trying to save caller
> couple of calls to our APIs.
> 
>>
>> Is this because of some implementation detail (that I already forgot) of
>> the message queue where signals are done after each "data..." or
>> "skip...", so it won't matter?
>>
>> Why not have everything you need in place before you wire this up -
>> patch 25 and 30 seem to be able to go after "most of" patch 31.
>>
>> John
>>
>> I'm going to stop here.
> 
> Fair enough. I'm gonna send v2 with most of your review points worked
> in. Except, for now I'm still gonna use Skip() and HoleSize() - changing
> that now would be a non-trivial piece of work and thus I'd like to do it
> just once, when we have clear agreement on naming. Hope you understand that.
> 

Understood about the naming thing - it's awfully frustrating to have to
go back and deal with merges once a personal naming scheme is panned
during review. Been there many times before. The primary reason for
mentioning it was the "newer" group norm of naming functions.

I also know by this point in the series my brain was mush. Perhaps a few
more comments in the code would have helped me along the way. I'm not
looking at commit messages when reading the code, so a description of
what the non-trivial algorithm is about to do could go a long way. Not
only to help the reviewer, but the poor sole that may come about said
algorithm N weeks, months, years from now wondering WTF is going on ;-)


John

>>
>>> Signed-off-by: Michal Privoznik <mprivozn at redhat.com>
>>> ---
>>>  src/rpc/virnetclientstream.c | 44 ++++++++++++++++++++++++++++++++++++++++++--
>>>  1 file changed, 42 insertions(+), 2 deletions(-)
>>>
>>> diff --git a/src/rpc/virnetclientstream.c b/src/rpc/virnetclientstream.c
>>> index c773524..ff35137 100644
>>> --- a/src/rpc/virnetclientstream.c
>>> +++ b/src/rpc/virnetclientstream.c
>>> @@ -296,6 +296,8 @@ int virNetClientStreamQueuePacket(virNetClientStreamPtr 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);
>>>  
>>>      virNetClientStreamEventTimerUpdate(st);
>>> @@ -359,7 +361,7 @@ int virNetClientStreamSendPacket(virNetClientStreamPtr st,
>>>  }
>>>  
>>>  
>>> -static int ATTRIBUTE_UNUSED
>>> +static int
>>>  virNetClientStreamHandleSkip(virNetClientPtr client,
>>>                               virNetClientStreamPtr st)
>>>  {
>>> @@ -435,6 +437,8 @@ int virNetClientStreamRecvPacket(virNetClientStreamPtr st,
>>>      virCheckFlags(0, -1);
>>>  
>>>      virObjectLock(st);
>>> +
>>> + reread:
>>>      if (!st->rx && !st->incomingEOF) {
>>>          virNetMessagePtr msg;
>>>          int ret;
>>> @@ -466,8 +470,44 @@ int virNetClientStreamRecvPacket(virNetClientStreamPtr st,
>>>      }
>>>  
>>>      VIR_DEBUG("After IO rx=%p", st->rx);
>>> +
>>> +    while (st->rx &&
>>> +           st->rx->header.type == VIR_NET_STREAM_SKIP) {
>>> +        /* Handle skip sent to us by server. */
>>> +
>>> +        if (virNetClientStreamHandleSkip(client, st) < 0)
>>> +            goto cleanup;
> 
> You can see my reasoning from above in effect here.
> 
>>> +    }
>>> +
>>> +    if (!st->rx && !st->incomingEOF && !st->skipLength) {
>>> +        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;
>>> -    while (want && st->rx) {
>>> +
>>> +    if (st->skipLength) {
>>> +        /* Pretend skipLength zeroes was read from stream. */
>>> +        size_t len = want;
>>> +
>>> +        if (len > st->skipLength)
>>> +            len = st->skipLength;
>>> +
>>> +        memset(data, 0, len);
>>> +        st->skipLength -= len;
>>> +        want -= len;
>>> +    }
>>> +
>>> +    while (want &&
>>> +           st->rx &&
>>> +           st->rx->header.type == VIR_NET_STREAM) {
>>>          virNetMessagePtr msg = st->rx;
>>>          size_t len = want;
>>>  
>>>
> 
> Michal
>