Re: [PATCH v2 06/38] virfdstream: Use messages instead of pipe

[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]

 



On 04/27/2017 07:48 PM, Daniel P. Berrange wrote:
> On Thu, Apr 20, 2017 at 12:01:35PM +0200, Michal Privoznik wrote:
>> One big downside of using the pipe to transfer the data is that
>> we can really transfer just bare data. No metadata can be carried
>> through unless some formatted messages are introduced. That would
>> be quite painful to achieve so let's use a message queue. It's
>> fairly easy to exchange info between threads now that iohelper is
>> no longer used.
>>
>> The reason why we cannot use the FD for plain files directly is
>> that despite us setting noblock flag on the FD, any
>> read()/write() blocks regardless (which is a show stopper since
>> those parts of the code are run from the event loop) and poll()
>> reports such FD as always readable/writable - even though the
>> subsequent operation might block.
>>
>> The pipe is still not gone though. It is used to signal to even
>> loop that an event occurred (e.g. data are available for reading
>> in the queue, or vice versa).
>>
>> Signed-off-by: Michal Privoznik <mprivozn@xxxxxxxxxx>
>> ---
>>  src/util/virfdstream.c | 402 ++++++++++++++++++++++++++++++++++++++++++-------
>>  1 file changed, 350 insertions(+), 52 deletions(-)
>>
> 
>> +static int
>> +virFDStreamMsgQueuePush(virFDStreamDataPtr fdst,
>> +                        virFDStreamMsgPtr msg,
>> +                        int fd,
>> +                        const char *fdname)
>> +{
>> +    virFDStreamMsgPtr *tmp = &fdst->msg;
>> +    char c = '1';
>> +
>> +    while (*tmp)
>> +        tmp = &(*tmp)->next;
>> +
>> +    *tmp = msg;
>> +    virCondSignal(&fdst->threadCond);
>> +
>> +    if (safewrite(fd, &c, sizeof(c)) != sizeof(c)) {
>> +        virReportSystemError(errno,
>> +                             _("Unable to write to %s"),
>> +                             fdname);
>> +        return -1;
>> +    }
>> +
>> +    return 0;
>> +}
>> +
>> +
>> +static virFDStreamMsgPtr
>> +virFDStreamMsgQueuePop(virFDStreamDataPtr fdst,
>> +                       int fd,
>> +                       const char *fdname)
>> +{
>> +    virFDStreamMsgPtr tmp = fdst->msg;
>> +    char c;
>> +
>> +    if (tmp) {
>> +        fdst->msg = tmp->next;
>> +        tmp->next = NULL;
>> +    }
>> +
>> +    virCondSignal(&fdst->threadCond);
>> +
>> +    if (saferead(fd, &c, sizeof(c)) != sizeof(c)) {
>> +        virReportSystemError(errno,
>> +                             _("Unable to read from %s"),
>> +                             fdname);
>> +        return NULL;
>> +    }
>> +
>> +    return tmp;
>> +}
> 
> Both these methods signal the condition
> 
> 
> 
>> +static ssize_t
>> +virFDStreamThreadDoRead(virFDStreamDataPtr fdst,
>> +                        const int fdin,
>> +                        const int fdout,
>> +                        const char *fdinname,
>> +                        const char *fdoutname,
>> +                        size_t buflen)
>> +{
>> +    virFDStreamMsgPtr msg = NULL;
>> +    char *buf = NULL;
>> +    ssize_t got;
>> +
>> +    if (VIR_ALLOC(msg) < 0)
>> +        goto error;
>> +
>> +    if (VIR_ALLOC_N(buf, buflen) < 0)
>> +        goto error;
>> +
>> +    if ((got = saferead(fdin, buf, buflen)) < 0) {
>> +        virReportSystemError(errno,
>> +                             _("Unable to read %s"),
>> +                             fdinname);
>> +        goto error;
>> +    }
>> +
>> +    msg->type = VIR_FDSTREAM_MSG_TYPE_DATA;
>> +    msg->stream.data.buf = buf;
>> +    msg->stream.data.len = got;
>> +    buf = NULL;
>> +
>> +    virFDStreamMsgQueuePush(fdst, msg, fdout, fdoutname);
>> +    msg = NULL;
>> +
>> +    return got;
>> +
>> + error:
>> +    VIR_FREE(buf);
>> +    virFDStreamMsgFree(msg);
>> +    return -1;
>> +}
>> +
>> +
>> +static ssize_t
>> +virFDStreamThreadDoWrite(virFDStreamDataPtr fdst,
>> +                         const int fdin,
>> +                         const int fdout,
>> +                         const char *fdinname,
>> +                         const char *fdoutname)
>> +{
>> +    ssize_t got;
>> +    virFDStreamMsgPtr msg = fdst->msg;
>> +    bool pop = false;
>> +
>> +    switch (msg->type) {
>> +    case VIR_FDSTREAM_MSG_TYPE_DATA:
>> +        got = safewrite(fdout,
>> +                        msg->stream.data.buf + msg->stream.data.offset,
>> +                        msg->stream.data.len - msg->stream.data.offset);
>> +        if (got < 0) {
>> +            virReportSystemError(errno,
>> +                                 _("Unable to write %s"),
>> +                                 fdoutname);
>> +            return -1;
>> +        }
>> +
>> +        msg->stream.data.offset += got;
>> +
>> +        pop = msg->stream.data.offset == msg->stream.data.len;
>> +        break;
>> +    }
>> +
>> +    if (pop) {
>> +        virFDStreamMsgQueuePop(fdst, fdin, fdinname);
>> +        virFDStreamMsgFree(msg);
>> +    }
>> +
>> +    return got;
>> +}
> 
> Both these methods call into the Pop/Push functions which
> signal the condition.
> 
>> @@ -304,14 +496,12 @@ virFDStreamThread(void *opaque)
>>      int fdout = data->fdout;
>>      char *fdoutname = data->fdoutname;
>>      virFDStreamDataPtr fdst = st->privateData;
>> -    char *buf = NULL;
>> +    bool doRead = fdst->threadDoRead;
>>      size_t buflen = 256 * 1024;
>>      size_t total = 0;
>>  
>>      virObjectRef(fdst);
>> -
>> -    if (VIR_ALLOC_N(buf, buflen) < 0)
>> -        goto error;
>> +    virObjectLock(fdst);
>>  
>>      while (1) {
>>          ssize_t got;
>> @@ -323,39 +513,56 @@ virFDStreamThread(void *opaque)
>>          if (buflen == 0)
>>              break; /* End of requested data from client */
>>  
>> -        if ((got = saferead(fdin, buf, buflen)) < 0) {
>> -            virReportSystemError(errno,
>> -                                 _("Unable to read %s"),
>> -                                 fdinname);
>> +        while (doRead == (fdst->msg != NULL) &&
>> +               !fdst->threadQuit) {
>> +            if (virCondWait(&fdst->threadCond, &fdst->parent.lock)) {
>> +                virReportSystemError(errno, "%s",
>> +                                     _("failed to wait on condition"));
>> +                goto error;
>> +            }
>> +        }
>> +
>> +        if (fdst->threadQuit) {
>> +            /* If stream abort was requested, quit early. */
>> +            if (fdst->threadAbort)
>> +                goto cleanup;
>> +
>> +            /* Otherwise flush buffers and quit gracefully. */
>> +            if (doRead == (fdst->msg != NULL))
>> +                break;
>> +        }
>> +
>> +        if (doRead)
>> +            got = virFDStreamThreadDoRead(fdst,
>> +                                          fdin, fdout,
>> +                                          fdinname, fdoutname,
>> +                                          buflen);
>> +        else
>> +            got = virFDStreamThreadDoWrite(fdst,
>> +                                           fdin, fdout,
>> +                                           fdinname, fdoutname);
>> +
>> +        if (got < 0)
> 
> 
> This method waits on the condition.
> 
> So unless I'm mistaken, this thread is signaling & waiting on
> the same condition, which feels wrong. Generally different
> threads would signal vs wait.
> 

I hear what you're saying but I don't think this is a problem. Exactly
because of the way we wait on the condition. Assume this thread is doing
reads. That is it reads from fdin (an actual file on a disk) and feeds
the message queue with the data. Now, it reads some data, push it at the
end of the message queue (fdst->msg != NULL at that point) and goes to
the while loop above. Effectively it eats all the spurious wake ups for
as long as fdst->msg != NULL.
Now assume the thread is doing writes (read data from the queue and
writes into a file a disk). Again, as long as there are some messages to
be written (that is fdst->msg != NULL) the control won't even reach
virCondWait. And if it does, it's because fdst->msg == NULL in which
case there is no data to be written.

But something has just came up to my mind whilst writing these lines - I
wonder if we can ditch the condition entirely and rely on the pipe +
poll(). I mean, this worker would use pipe to signalize to the event
loop that there is a message waiting for it in the queue. Question is
how thread safe this approach would be.

Michal

--
libvir-list mailing list
libvir-list@xxxxxxxxxx
https://www.redhat.com/mailman/listinfo/libvir-list



[Index of Archives]     [Virt Tools]     [Libvirt Users]     [Lib OS Info]     [Fedora Users]     [Fedora Desktop]     [Fedora SELinux]     [Big List of Linux Books]     [Yosemite News]     [KDE Users]     [Fedora Tools]
  Powered by Linux