Re: [PATCH spice-server v3 2/2] stream-channel: Send the full frame in a single message

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

 



On Wed, 2018-05-09 at 05:18 -0400, Frediano Ziglio wrote:
> > 
> > On Tue, 2018-05-08 at 10:26 +0100, Frediano Ziglio wrote:
> > > The current implementation of server and client assumes that a single
> > > data message contains an encoded frame.
> > > This is not a problem for most encoding but for MJPEG this causes
> > > the client to fail decoding.
> > > Collapse frame data into a single message before sending to the client.
> > > This is done in the channel as the channel code is responsible to take care
> > > of client protocol details. This allows for instance to support chunked
> > > transfer to client if implemented.
> > 
> > So the proper fix if we didn't care about compatibility with old/other
> > clients would be to implement the "chunked transfer" in the client,
> > right? But we don't want to break the compatibility. So it is basically
> > a property of the protocol that a frame must be sent in a single
> > message.
> > 
> 
> Yes, there are different assumptions in the code about this, not clear if
> this was due by MJPEG encoding (first and only encoding for long time), or
> something by design, the documentation on the protocol is basically
> nothing (just the "stream_data" message with no description at all).

Ok, so perhaps defining this (using the limitations of the current
implementation) should be part of the series?

> > Why not also send it in a single message from the streaming agent to
> > the server, so that the whole message can be forwarded and the messages
> > are 1:1? Instead of adding this accumulation code and having to copy
> > the frame in the process?
> > 
> 
> The agent send a single message, but reads/writes to the device are
> not atomic. Note that the current protocol introduce additional
> delays as the frames cannot be partially decoded but must wait for the
> full message (maybe the client can change its read code to handle this,
> at the moment it does nothing about, on the server is less of a problem
> as the message is build quickly from device to memory so not much delay
> is added).

This is the part I don't understand... AFAICS, you read the whole
message in red-stream-device.c:handle_msg_data(). That should be the
whole frame? Then you send the whole frame with
stream_channel_send_data(). So it should never be partial?

Also see below.

> > If I don't understand this properly, please excuse my ignorace...
> > 
> > Cheers,
> > Lukas
> > 
> > > Signed-off-by: Frediano Ziglio <fziglio@xxxxxxxxxx>
> > > ---
> > >  server/red-stream-device.c | 10 +++--
> > >  server/stream-channel.c    | 83 +++++++++++++++++++++++++++++++++-----
> > >  server/stream-channel.h    | 12 ++++++
> > >  3 files changed, 90 insertions(+), 15 deletions(-)
> > > 
> > > diff --git a/server/red-stream-device.c b/server/red-stream-device.c
> > > index df6a366f..864be99e 100644
> > > --- a/server/red-stream-device.c
> > > +++ b/server/red-stream-device.c
> > > @@ -314,11 +314,13 @@ handle_msg_data(StreamDevice *dev,
> > > SpiceCharDeviceInstance *sin)
> > >          if (n <= 0) {
> > >              break;
> > >          }
> > > -        // TODO collect all message ??
> > > -        // up: we send a single frame together
> > > -        // down: guest can cause a crash
> > > -        stream_channel_send_data(dev->stream_channel, buf, n,
> > > reds_get_mm_time());
> > > +        uint32_t mm_time = reds_get_mm_time();
> > > +        if (dev->msg_pos == 0) {
> > > +            stream_channel_start_data(dev->stream_channel, dev->hdr.size,
> > > mm_time);
> > > +        }
> > > +        stream_channel_send_data(dev->stream_channel, buf, n, mm_time);
> > >          dev->hdr.size -= n;
> > > +        dev->msg_pos += n;
> > >      }
> > >  
> > >      return dev->hdr.size == 0;
> > > diff --git a/server/stream-channel.c b/server/stream-channel.c
> > > index fc409ee6..0ac7ba1d 100644
> > > --- a/server/stream-channel.c
> > > +++ b/server/stream-channel.c
> > > @@ -44,6 +44,7 @@
> > >  
> > >  typedef struct StreamChannelClient StreamChannelClient;
> > >  typedef struct StreamChannelClientClass StreamChannelClientClass;
> > > +typedef struct StreamDataItem StreamDataItem;
> > >  
> > >  /* we need to inherit from CommonGraphicsChannelClient
> > >   * to get buffer handling */
> > > @@ -74,6 +75,10 @@ struct StreamChannel {
> > >  
> > >      StreamQueueStat queue_stat;
> > >  
> > > +    /* pending partial data item */
> > > +    StreamDataItem *data_item;
> > > +    uint32_t data_item_pos;
> > > +
> > >      /* callback to notify when a stream should be started or stopped */
> > >      stream_channel_start_proc start_cb;
> > >      void *start_opaque;
> > > @@ -105,12 +110,12 @@ typedef struct StreamCreateItem {
> > >      SpiceMsgDisplayStreamCreate stream_create;
> > >  } StreamCreateItem;
> > >  
> > > -typedef struct StreamDataItem {
> > > +struct StreamDataItem {
> > >      RedPipeItem base;
> > >      StreamChannel *channel;
> > >      // NOTE: this must be the last field in the structure
> > >      SpiceMsgDisplayStreamData data;
> > > -} StreamDataItem;
> > > +};
> > >  
> > >  #define PRIMARY_SURFACE_ID 0
> > >  
> > > @@ -130,6 +135,16 @@ stream_channel_client_init(StreamChannelClient
> > > *client)
> > >      client->stream_id = -1;
> > >  }
> > >  
> > > +static void
> > > +stream_channel_unref_data_item(StreamChannel *channel)
> > > +{
> > > +    if (channel->data_item) {
> > > +        red_pipe_item_unref(&channel->data_item->base);
> > > +        channel->data_item = NULL;
> > > +        channel->data_item_pos = 0;
> > > +    }
> > > +}
> > > +
> > >  static void
> > >  request_new_stream(StreamChannel *channel, StreamMsgStartStop *start)
> > >  {
> > > @@ -153,6 +168,7 @@ stream_channel_client_on_disconnect(RedChannelClient
> > > *rcc)
> > >      channel->stream_id = -1;
> > >      channel->width = 0;
> > >      channel->height = 0;
> > > +    stream_channel_unref_data_item(channel);
> > >  
> > >      // send stream stop to device
> > >      StreamMsgStartStop stop = { 0, };
> > > @@ -452,6 +468,16 @@ stream_channel_constructed(GObject *object)
> > >      reds_register_channel(reds, red_channel);
> > >  }
> > >  
> > > +static void
> > > +stream_channel_finalize(GObject *object)
> > > +{
> > > +    StreamChannel *channel = STREAM_CHANNEL(object);
> > > +
> > > +    stream_channel_unref_data_item(channel);
> > > +
> > > +    G_OBJECT_CLASS(stream_channel_parent_class)->finalize(object);
> > > +}
> > > +
> > >  static void
> > >  stream_channel_class_init(StreamChannelClass *klass)
> > >  {
> > > @@ -459,6 +485,7 @@ stream_channel_class_init(StreamChannelClass *klass)
> > >      RedChannelClass *channel_class = RED_CHANNEL_CLASS(klass);
> > >  
> > >      object_class->constructed = stream_channel_constructed;
> > > +    object_class->finalize = stream_channel_finalize;
> > >  
> > >      channel_class->parser =
> > >      spice_get_client_channel_parser(SPICE_CHANNEL_DISPLAY, NULL);
> > >      channel_class->handle_message = handle_message;
> > > @@ -532,14 +559,18 @@ data_item_free(RedPipeItem *base)
> > >  {
> > >      StreamDataItem *pipe_item = SPICE_UPCAST(StreamDataItem, base);
> > >  
> > > -    stream_channel_update_queue_stat(pipe_item->channel, -1,
> > > -pipe_item->data.data_size);
> > > +    if (pipe_item->channel->data_item != pipe_item) {
> > > +        stream_channel_update_queue_stat(pipe_item->channel, -1,
> > > -pipe_item->data.data_size);
> > > +    }
> > >  
> > >      g_free(pipe_item);
> > >  }
> > >  
> > > -static StreamDataItem*
> > > -stream_data_item_new(StreamChannel *channel, size_t size, uint32_t
> > > mm_time)
> > > +static void
> > > +stream_channel_init_data_item(StreamChannel *channel, size_t size,
> > > uint32_t mm_time)
> > >  {
> > > +    stream_channel_unref_data_item(channel);
> > > +
> > >      StreamDataItem *item = g_malloc(sizeof(*item) + size);
> > >      red_pipe_item_init_full(&item->base, RED_PIPE_ITEM_TYPE_STREAM_DATA,
> > >                              data_item_free);
> > > @@ -548,7 +579,22 @@ stream_data_item_new(StreamChannel *channel, size_t
> > > size, uint32_t mm_time)
> > >      item->data.data_size = size;
> > >      item->channel = channel;
> > >  
> > > -    return item;
> > > +    channel->data_item = item;
> > > +    channel->data_item_pos = 0;
> > > +}
> > > +
> > > +void
> > > +stream_channel_start_data(StreamChannel *channel, size_t size, uint32_t
> > > mm_time)
> > > +{
> > > +    // see stream_channel_send_data comment
> > > +    if (channel->stream_id < 0) {
> > > +        return;
> > > +    }
> > > +
> > > +    // TODO this collects all chunks in a single message
> > > +    // up: we send a single frame together (more compatible)
> > > +    // down: guest can cause a crash due to DoS. As a safe measure we
> > > limit the maximum message
> > > +    stream_channel_init_data_item(channel, MIN(size, 32*1024*1024),
> > > mm_time);
> > >  }
> > >  
> > >  void
> > > @@ -563,11 +609,25 @@ stream_channel_send_data(StreamChannel *channel,
> > > const void *data, size_t size,
> > >  
> > >      RedChannel *red_channel = RED_CHANNEL(channel);
> > >  
> > > -    StreamDataItem *item = stream_data_item_new(channel, size, mm_time);
> > > -    stream_channel_update_queue_stat(channel, 1, size);
> > > -    // TODO try to optimize avoiding the copy
> > > -    memcpy(item->data.data, data, size);
> > > -    red_channel_pipes_add(red_channel, &item->base);
> > > +    while (size) {
> > > +        if (channel->data_item == NULL) {
> > > +            stream_channel_init_data_item(channel, size, mm_time);
> > > +        }
> > > +
> > > +        StreamDataItem *item = channel->data_item;
> > > +
> > > +        size_t copy_size = item->data.data_size - channel->data_item_pos;
> > > +        copy_size = MIN(copy_size, size);
> > > +        // TODO try to optimize avoiding the copy
> > > +        memcpy(item->data.data + channel->data_item_pos, data, copy_size);
> > > +        size -= copy_size;
> > > +        channel->data_item_pos += copy_size;
> > > +        if (channel->data_item_pos == item->data.data_size) {
> > > +            channel->data_item = NULL;
> > > +            stream_channel_update_queue_stat(channel, 1,
> > > item->data.data_size);
> > > +            red_channel_pipes_add(red_channel, &item->base);
> > > +        }
> > > +    }

What does the while (size) loop do here? It will do more than one
iteration only if copy_size < size, which means there is not enough
space in the item buffer and in that case it seems to me it will loop
forever? Am I missing something?

> > >  }
> > >  
> > >  void
> > > @@ -607,6 +667,7 @@ stream_channel_reset(StreamChannel *channel)
> > >      channel->stream_id = -1;
> > >      channel->width = 0;
> > >      channel->height = 0;
> > > +    stream_channel_unref_data_item(channel);
> > >  
> > >      if (!red_channel_is_connected(red_channel)) {
> > >          return;
> > > diff --git a/server/stream-channel.h b/server/stream-channel.h
> > > index e8bec80b..18a1bdea 100644
> > > --- a/server/stream-channel.h
> > > +++ b/server/stream-channel.h
> > > @@ -60,6 +60,18 @@ struct StreamMsgStartStop;
> > >  
> > >  void stream_channel_change_format(StreamChannel *channel,
> > >                                    const struct StreamMsgFormat *fmt);
> > > +
> > > +/**
> > > + * Tell the channel that a new data packet is starting.
> > > + * This can be used to group all chunks together.
> > > + */
> > > +void stream_channel_start_data(StreamChannel *channel,
> > > +                               size_t size,
> > > +                               uint32_t mm_time);
> > > +
> > > +/**
> > > + * Send to channel a chunk of data.
> > > + */
> > >  void stream_channel_send_data(StreamChannel *channel,
> > >                                const void *data, size_t size,
> > >                                uint32_t mm_time);
_______________________________________________
Spice-devel mailing list
Spice-devel@xxxxxxxxxxxxxxxxxxxxx
https://lists.freedesktop.org/mailman/listinfo/spice-devel




[Index of Archives]     [Linux Virtualization]     [Linux Virtualization]     [Linux ARM Kernel]     [Linux ARM]     [Linux Omap]     [Fedora ARM]     [IETF Annouce]     [Security]     [Bugtraq]     [Linux OMAP]     [Linux MIPS]     [ECOS]     [Asterisk Internet PBX]     [Linux API]     [Monitors]