[libvirt] [PATCH 01/11] Add public API definition for data stream handling

[Daniel Veillard]

On Mon, Aug 24, 2009 at 09:51:04PM +0100, Daniel P. Berrange wrote:
> * include/libvirt/libvirt.h, include/libvirt/libvirt.h.in: Public
>   API contract for virStreamPtr object
> * src/libvirt_public.syms: Export data stream APIs
> * src/libvirt_private.syms: Export internal helper APIs
> * src/libvirt.c: Data stream API driver dispatch
> * src/datatypes.h, src/datatypes.c: Internal helpers for virStreamPtr
>   object
> * src/driver.h: Define internal driver API for streams
> * .x-sc_avoid_write: Ignore src/libvirt.c because it trips
>   up on comments including write()
[...]
> --- a/include/libvirt/libvirt.h
> +++ b/include/libvirt/libvirt.h
> @@ -110,6 +110,24 @@ typedef enum {
>       VIR_DOMAIN_NONE = 0
>  } virDomainCreateFlags;
>  
> +
> +
> +/**
> + * virStream:
> + *
> + * a virStream is a private structure representing a data stream.
> + */
> +typedef struct _virStream virStream;
> +
> +/**
> + * virStreamPtr:
> + *
> + * a virStreamPtr is pointer to a virStream private structure, this is the
> + * type used to reference a data stream in the API.
> + */
> +typedef virStream *virStreamPtr;
> +
> +
>  /**
>   * VIR_SECURITY_LABEL_BUFLEN:
>   *
> @@ -1448,6 +1466,81 @@ void virEventRegisterImpl(virEventAddHandleFunc addHandle,
>                            virEventAddTimeoutFunc addTimeout,
>                            virEventUpdateTimeoutFunc updateTimeout,
>                            virEventRemoveTimeoutFunc removeTimeout);
> +
> +enum {
> +    VIR_STREAM_NONBLOCK = (1 << 0),
> +};
> +
> +virStreamPtr virStreamNew(virConnectPtr conn,
> +                          unsigned int flags);

 Would flags be sufficient if we were to encode some priorities to
streams? If we end up limiting the number of active streams, giving
higher priority or some reserved slots for quicker operations may
be important, flags should be sufficient for this if the need arise
so I don't see a problem but I raise the point.

> +int virStreamRef(virStreamPtr st);
> +
> +int virStreamSend(virStreamPtr st,
> +                  const char *data,
> +                  size_t nbytes);
> +
> +int virStreamRecv(virStreamPtr st,
> +                  char *data,
> +                  size_t nbytes);
> +
> +
> +typedef int (*virStreamSourceFunc)(virStreamPtr st,
> +                                   char *data,
> +                                   size_t nbytes,
> +                                   void *opaque);

  I think the signature is fine but we need to document all the
arguments and the return values.

> +int virStreamSendAll(virStreamPtr st,
> +                     virStreamSourceFunc handler,
> +                     void *opaque);

  Hum, I had to look at the comment to really understand. I'm not
100% sure, maybe we should allow for handler() to be called multiple
time, not just once.
  For example I would allow virStreamSendAll() code to call handler
multiple time, until the handler like a read() returns 0 (or -1 on
error), in any case the signature should be documented fully.

> +typedef int (*virStreamSinkFunc)(virStreamPtr st,
> +                                 const char *data,
> +                                 size_t nbytes,
> +                                 void *opaque);

  Same thing do we allow a sink function to be called repeatedly ?
If we want to allow this in some ways we will need an extra argument
to indicate the end of the stream. Even if we don't plan this yet, I
would suggest to add a flags to allow for this possibility in the
future. With a chunk size of 256K at the protocol level it may not
be a good idea to keep the full data in memory, so I would allow
for this interface to call the sink multiple times. And IMHO it's best
to pass the indication of end of transfer directly at the sink level
rather than wait for the virStreamFree() coming from the user.

> +int virStreamRecvAll(virStreamPtr st,
> +                     virStreamSinkFunc handler,
> +                     void *opaque);

  Okay

> +typedef enum {
> +    VIR_STREAM_EVENT_READABLE  = (1 << 0),
> +    VIR_STREAM_EVENT_WRITABLE  = (1 << 1),
> +    VIR_STREAM_EVENT_ERROR     = (1 << 2),
> +    VIR_STREAM_EVENT_HANGUP    = (1 << 3),
> +} virStreamEventType;
> +
> +
> +/**
> + * virStreamEventCallback:
> + *
> + * @stream: stream on which the event occurred
> + * @events: bitset of events from virEventHandleType constants
> + * @opaque: user data registered with handle
> + *
> + * Callback for receiving stream events. The callback will
> + * be invoked once for each event which is pending.
> + */
> +typedef void (*virStreamEventCallback)(virStreamPtr stream, int events, void *opaque);
> +
> +int virStreamEventAddCallback(virStreamPtr stream,
> +                              int events,
> +                              virStreamEventCallback cb,
> +                              void *opaque,
> +                              virFreeCallback ff);
> +
> +int virStreamEventUpdateCallback(virStreamPtr stream,
> +                                 int events);
> +
> +int virStreamEventRemoveCallback(virStreamPtr stream);
> +
> +
> +int virStreamFinish(virStreamPtr st);
> +int virStreamAbort(virStreamPtr st);

For those 2 maybe add a flag, to allow for example background disconnection.
Even if the stream wasn't created ASYNC, we may want sometime to
abruptly end without waiting.

> +int virStreamFree(virStreamPtr st);

With the exception of the extra flags suggestion for the sink callback
and finish/abort, the aPI looks fine to me.

[...]
> +/**
> + * virStreamSendAll:
> + * @stream: pointer to the stream object
> + * @handler: source callback for reading data from application
> + * @opaque: application defined data
> + *
> + * Send the entire data stream, reading the data from the
> + * requested data source. This is simply a convenient alternative
> + * to virStreamSend, for apps that do blocking-I/o.
> + *
> + * A example using this with a hypothetical file upload
> + * API looks like
> + *
> + *   int mysource(virStreamPtr st, char *buf, int nbytes, void *opaque) {
> + *       int *fd = opaque;
> + *
> + *       return read(*fd, buf, nbytes);
> + *   }
> + *
> + *   virStreamPtr st = virStreamNew(conn, 0);
> + *   int fd = open("demo.iso", O_RDONLY)
> + *
> + *   virConnectUploadFile(conn, st);
> + *   virStreamSendAll(st, mysource, &fd);
> + *   virStreamFree(st);
> + *   close(fd);


 Hum, the clasic example of blocking I/Os is if people use multithreaded
apps.
  What is a second thread calls calls virStreamAbort() while
virStreamSendAll() is in progress, e.g. the user clicked on an abort
button from the UI ?
  Same question for virStreamRecvAll() ?

> + * Returns 0 if all the data was succesfully sent. The stream
> + * will be marked as finished on success, so the caller need
> + * only call virStreamFree().
> + *
> + * Returns -1 upon any error, with the stream being marked as
> + * aborted, so the caller need only call virStreamFree()
> + */
> +int virStreamSendAll(virStreamPtr stream,
> +                     virStreamSourceFunc handler,
> +                     void *opaque)


> + *   virConnectUploadFile(conn, st);
> + *   virStreamRecvAll(st, mysink, &fd);
> + *   virStreamFree(st);
> + *   close(fd);

 Would the current API allow for the sink callback to close the fd()
at the end of the transfer ? Right now, I don't think so because we
don't know what is the last callback (assuming multiple ones).

  Except for those couple of uncertainties, the API looks fine to me !

Daniel

-- 
Daniel Veillard      | libxml Gnome XML XSLT toolkit  http://xmlsoft.org/
daniel at veillard.com  | Rpmfind RPM search engine http://rpmfind.net/
http://veillard.com/ | virtualization library  http://libvirt.org/