[Qemu-devel] Re: [RFC][RESEND][PATCH v1 06/15] virtproxy: add read handler for communication channel

[Adam Litke <agl@us.ibm.com>]

On Wed, 2010-11-03 at 10:28 -0500, Michael Roth wrote:
> Handle data coming in over the channel as VPPackets: Process control
> messages and forward data from remote client/server connections to the
> appropriate server/client FD on our end.
> 
> Signed-off-by: Michael Roth <mdroth@linux.vnet.ibm.com>
> ---
>  virtproxy.c |   83 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
>  1 files changed, 83 insertions(+), 0 deletions(-)
> 
> diff --git a/virtproxy.c b/virtproxy.c
> index 20532c2..c9c3022 100644
> --- a/virtproxy.c
> +++ b/virtproxy.c
> @@ -33,6 +33,7 @@
>  #define VP_SERVICE_ID_LEN 32    /* max length of service id string */
>  #define VP_PKT_DATA_LEN 1024    /* max proxied bytes per VPPacket */
>  #define VP_CONN_DATA_LEN 1024   /* max bytes conns can send at a time */
> +#define VP_CHAN_DATA_LEN 4096   /* max bytes channel can send at a time */
>  #define VP_MAGIC 0x1F374059
> 
>  /* listening fd, one for each service we're forwarding to remote end */
> @@ -150,6 +151,8 @@ static QemuOptsList vp_socket_opts = {
>      },
>  };
> 
> +static void vp_channel_read(void *opaque);

Try to get rid of these forward declarations.  If you really need them,
consider adding a virtproxy-internal.h.


> @@ -230,3 +233,83 @@ static void vp_channel_accept(void *opaque)
>      /* dont accept anymore connections until channel_fd is closed */
>      vp_set_fd_handler(drv->listen_fd, NULL, NULL, NULL);
>  }
> +
> +/* read handler for communication channel
> + *
> + * de-multiplexes data coming in over the channel. for control messages
> + * we process them here, for data destined for a service or client we
> + * send it to the appropriate FD.
> + */
> +static void vp_channel_read(void *opaque)
> +{
> +    VPDriver *drv = opaque;
> +    VPPacket pkt;
> +    int count, ret, buf_offset;
> +    char buf[VP_CHAN_DATA_LEN];
> +    char *pkt_ptr, *buf_ptr;
> +
> +    TRACE("called with opaque: %p", drv);
> +
> +    count = read(drv->channel_fd, buf, sizeof(buf));
> +
> +    if (count == -1) {
> +        LOG("read() failed: %s", strerror(errno));
> +        return;
> +    } else if (count == 0) {
> +        /* TODO: channel closed, this probably shouldn't happen for guest-side
> +         * serial/virtio-serial connections, but need to confirm and consider
> +         * what should happen in this case. as it stands this virtproxy instance
> +         * is basically defunct at this point, same goes for "client" instances
> +         * of virtproxy where the remote end has hung-up.
> +         */
> +        LOG("channel connection closed");
> +        vp_set_fd_handler(drv->channel_fd, NULL, NULL, drv);
> +        drv->channel_fd = -1;
> +        if (drv->listen_fd) {
> +            vp_set_fd_handler(drv->listen_fd, vp_channel_accept, NULL, drv);
> +        }
> +        /* TODO: should close/remove/delete all existing VPConns here */

Looks like you have a little work TODO here still.  Perhaps you could
add some reset/init functions that would make handling this easier.  The
ability to reset state at both the channel and VPConn levels should give
you the ability to handle errors more gracefully in other places where
you currently just log them.

> +    }
> +
> +    if (drv->buflen + count >= sizeof(VPPacket)) {
> +        TRACE("initial packet, drv->buflen: %d", drv->buflen);
> +        pkt_ptr = (char *)&pkt;
> +        memcpy(pkt_ptr, drv->buf, drv->buflen);

Can drv->buflen ever be > sizeof(VPPacket) ?  You might consider adding
an assert(drv->buflen < sizeof(VPPacket)) unless it's mathematically
impossible.

> +        pkt_ptr += drv->buflen;
> +        memcpy(pkt_ptr, buf, sizeof(VPPacket) - drv->buflen);
> +        /* handle first packet */
> +        ret = vp_handle_packet(drv, &pkt);
> +        if (ret != 0) {
> +            LOG("error handling packet");
> +        }
> +        /* handle the rest of the buffer */
> +        buf_offset = sizeof(VPPacket) - drv->buflen;
> +        drv->buflen = 0;
> +        buf_ptr = buf + buf_offset;
> +        count -= buf_offset;
> +        while (count > 0) {
> +            if (count >= sizeof(VPPacket)) {
> +                /* handle full packet */
> +                TRACE("additional packet, drv->buflen: %d", drv->buflen);
> +                memcpy((void *)&pkt, buf_ptr, sizeof(VPPacket));
> +                ret = vp_handle_packet(drv, &pkt);
> +                if (ret != 0) {
> +                    LOG("error handling packet");
> +                }
> +                count -= sizeof(VPPacket);
> +                buf_ptr += sizeof(VPPacket);
> +            } else {
> +                /* buffer the remainder */
> +                TRACE("buffering packet");
> +                memcpy(drv->buf, buf_ptr, count);
> +                drv->buflen = count;
> +                break;
> +            }
> +        }
> +    } else {
> +        /* haven't got a full VPPacket yet, buffer for later */
> +        buf_ptr = drv->buf + drv->buflen;
> +        memcpy(buf_ptr, buf, count);
> +        drv->buflen += count;

With this algorithm you can hold some data hostage indefinitely.  You
either need to send out smaller packets of whatever data you receive or
maybe have a timer that periodically fires to flush drv->buf.  Any
thoughts on what you plan to do here?

-- 
Thanks,
Adam