[Spice-devel] [PATCH spice-gtk] webdav: don't buffer input from phodav

Frediano Ziglio fziglio at redhat.com
Mon Jul 1 11:38:50 UTC 2019


> 
> > 
> > The current approach with OutputQueue in webdav has several problems:
> > 
> > * if the connection is slow, webdav keeps reading from phodav
> > and pushing messages to the internal channel xmit_queue.
> > This way, the queue can grow very quickly and the whole file
> > that is being transferred using webdav essentially gets loaded into memory.
> > 
> > * spice channel first flushes all messages in the xmit_queue and
> > then proceeds to reading. If webdav floods the xmit_queue with
> > a ton of messages, spice channel does not leave iterate_write until
> > the queue gets empty. This way, reading from the channel is blocked
> > till the whole file is transferred.
> > 
> > * OutputQueue uses g_output_stream_flush_async() on SpiceVmcOutputStream
> > that does not implement flush
> > 
> > To solve these issues, don't read from phodav until the last message
> > for a given client is written out to the socket.
> > (main channel currently uses the same approach when transferring files)
> > 
> > OutputQueue used an idle function to schedule the write and then
> > called mux_pushed_cb which started reading from phodav with priority
> > G_PRIORITY_DEFAULT.
> > Since this new approach does not utilize the idle scheduling,
> > lower the priority in client_start_read() to G_PRIORITY_DEFAULT_IDLE
> > to make sure webdav does not block other channels.
> > 
> 
> This sounds a bit worrying. Why with all coroutine/async code we
> have we should have a blockage? This sounds wrong.
> 
> > Also implement spice_webdav_channel_reset(). This is necessary because
> > spice_vmc_write_async() references the channel. If the channel is to be
> > disconnected, the write operations need to be cancelled so that
> > the references to the channel are released asap. Otherwise,
> > spice session would be stuck waiting for the channel to finalize.
> > 
> > Signed-off-by: Jakub Janků <jjanku at redhat.com>
> > ---
> > 
> > Note: I left the OutputQueue code still in place since I'm considering
> > using it for the drag&drop functionality.
> > 
> >  src/channel-webdav.c | 49 ++++++++++++++++++++++++++++++++------------
> >  1 file changed, 36 insertions(+), 13 deletions(-)
> > 
> > diff --git a/src/channel-webdav.c b/src/channel-webdav.c
> > index f5a38ad..66f8a5b 100644
> > --- a/src/channel-webdav.c
> > +++ b/src/channel-webdav.c
> > @@ -45,13 +45,12 @@
> >   * Since: 0.24
> >   */
> >  
> > -typedef struct _OutputQueue OutputQueue;
> > +/* typedef struct _OutputQueue OutputQueue; */
> >  
> >  struct _SpiceWebdavChannelPrivate {
> >      SpiceVmcStream *stream;
> >      GCancellable *cancellable;
> >      GHashTable *clients;
> > -    OutputQueue *queue;
> >  
> >      gboolean demuxing;
> >      struct _demux {
> > @@ -65,6 +64,7 @@ G_DEFINE_TYPE_WITH_PRIVATE(SpiceWebdavChannel,
> > spice_webdav_channel, SPICE_TYPE_
> >  
> >  static void spice_webdav_handle_msg(SpiceChannel *channel, SpiceMsgIn
> >  *msg);
> >  
> > +#if 0
> >  struct _OutputQueue {
> >      GOutputStream *output;
> >      gboolean flushing;
> > @@ -179,6 +179,8 @@ static void output_queue_push(OutputQueue *q, const
> > guint8 *buf, gsize size,
> >          q->idle_id = g_idle_add(output_queue_idle, q);
> >  }
> >  
> > +#endif
> > +
> >  typedef struct Client
> >  {
> >      guint refs;
> > @@ -227,10 +229,17 @@ static void remove_client(Client *client)
> >      g_hash_table_remove(client->self->priv->clients, &client->id);
> >  }
> >  
> > -static void mux_pushed_cb(OutputQueue *q, gpointer user_data)
> > +#define MAX_MUX_SIZE G_MAXUINT16
> > +
> > +static void
> > +mux_msg_flushed_cb(GObject *source_object,
> > +                   GAsyncResult *result,
> > +                   gpointer user_data)
> >  {
> >      Client *client = user_data;
> >  
> > +    g_output_stream_write_finish(G_OUTPUT_STREAM(source_object), result,
> > NULL);
> > +
> >      if (client->mux.size == 0 ||
> >          !client_start_read(client)) {
> >          remove_client(client);
> > @@ -239,14 +248,13 @@ static void mux_pushed_cb(OutputQueue *q, gpointer
> > user_data)
> >      client_unref(client);
> >  }
> >  
> > -#define MAX_MUX_SIZE G_MAXUINT16
> > -
> >  static void server_reply_cb(GObject *source_object,
> >                              GAsyncResult *res,
> >                              gpointer user_data)
> >  {
> >      Client *client = user_data;
> >      SpiceWebdavChannelPrivate *c = client->self->priv;
> > +    GOutputStream *mux_out;
> >      GError *err = NULL;
> >      gssize size;
> >  
> > @@ -258,10 +266,15 @@ static void server_reply_cb(GObject *source_object,
> >      g_return_if_fail(size >= 0);
> >      client->mux.size = size;
> >  
> > -    output_queue_push(c->queue, (guint8 *)&client->mux.id, sizeof(gint64),
> > NULL, NULL);
> > +    mux_out = g_io_stream_get_output_stream(G_IO_STREAM(c->stream));
> > +
> > +    g_output_stream_write(mux_out, (guint8 *)&client->mux.id,
> > sizeof(gint64), NULL, NULL);
> >      client->mux.size = GUINT16_TO_LE(client->mux.size);
> > -    output_queue_push(c->queue, (guint8 *)&client->mux.size,
> > sizeof(guint16), NULL, NULL);
> > -    output_queue_push(c->queue, (guint8 *)client->mux.buf, size,
> > (GFunc)mux_pushed_cb, client);
> > +    g_output_stream_write(mux_out, (guint8 *)&client->mux.size,
> > sizeof(guint16), NULL, NULL);
> > +    /* this internally uses spice_vmc_write_async(), priority is ignored;
> > +     * the callback is invoked once the msg is written out to the socket
> > */
> > +    g_output_stream_write_async(mux_out, (guint8 *)client->mux.buf, size,
> > +        G_PRIORITY_DEFAULT, client->cancellable, mux_msg_flushed_cb,
> > client);
> 
> Why don't you use g_output_stream_writev_all_async here writing all 3 buffers
> together and avoid blocking calls?
> Well, it's available starting from GLib 2.60 so possibly you'll have
> to provide a replacement for GLib 2.60 using g_output_stream_write_all_async.
> 

Maybe something like this (not tested) could help
https://cgit.freedesktop.org/~fziglio/spice-gtk/commit/?h=commit&id=5fcdfe6bcf8e8cacceef01b2b36c5506da653896

> >  
> >      return;
> >  
> > @@ -284,8 +297,9 @@ static bool client_start_read(Client *client)
> >      if (g_input_stream_is_closed(input)) {
> >          return false;
> >      }
> > +    /* use low priority to leave enough time for other channels */
> >      g_input_stream_read_async(input, client->mux.buf, MAX_MUX_SIZE,
> > -                              G_PRIORITY_DEFAULT, client->cancellable,
> > server_reply_cb,
> > +                              G_PRIORITY_DEFAULT_IDLE,
> > client->cancellable,
> > server_reply_cb,
> >                                client_ref(client));
> >      return true;
> >  }
> > @@ -540,9 +554,6 @@ static void
> > spice_webdav_channel_init(SpiceWebdavChannel
> > *channel)
> >      c->clients = g_hash_table_new_full(g_int64_hash, g_int64_equal,
> >                                         NULL, client_remove_unref);
> >      c->demux.buf = g_malloc0(MAX_MUX_SIZE);
> > -
> > -    GOutputStream *ostream =
> > g_io_stream_get_output_stream(G_IO_STREAM(c->stream));
> > -    c->queue = output_queue_new(ostream);
> >  }
> >  
> >  static void spice_webdav_channel_finalize(GObject *object)
> > @@ -560,7 +571,6 @@ static void spice_webdav_channel_dispose(GObject
> > *object)
> >  
> >      g_cancellable_cancel(c->cancellable);
> >      g_clear_object(&c->cancellable);
> > -    g_clear_pointer(&c->queue, output_queue_free);
> >      g_clear_object(&c->stream);
> >      g_hash_table_unref(c->clients);
> >  
> > @@ -572,6 +582,18 @@ static void spice_webdav_channel_up(SpiceChannel
> > *channel)
> >      CHANNEL_DEBUG(channel, "up");
> >  }
> >  
> > +static void spice_webdav_channel_reset(SpiceChannel *channel, gboolean
> > migrating)
> > +{
> > +    SpiceWebdavChannelPrivate *c;
> > +    c =
> > spice_webdav_channel_get_instance_private(SPICE_WEBDAV_CHANNEL(channel));
> > +
> > +    g_cancellable_cancel(c->cancellable);
> > +    c->demuxing = FALSE;
> > +    g_hash_table_remove_all(c->clients);
> > +
> > +
> > SPICE_CHANNEL_CLASS(spice_webdav_channel_parent_class)->channel_reset(channel,
> > migrating);
> > +}
> > +
> >  static void spice_webdav_channel_class_init(SpiceWebdavChannelClass
> >  *klass)
> >  {
> >      GObjectClass *gobject_class = G_OBJECT_CLASS(klass);
> > @@ -581,6 +603,7 @@ static void
> > spice_webdav_channel_class_init(SpiceWebdavChannelClass *klass)
> >      gobject_class->finalize     = spice_webdav_channel_finalize;
> >      channel_class->handle_msg   = spice_webdav_handle_msg;
> >      channel_class->channel_up   = spice_webdav_channel_up;
> > +    channel_class->channel_reset = spice_webdav_channel_reset;
> >  
> >      g_signal_override_class_handler("port-event",
> >                                      SPICE_TYPE_WEBDAV_CHANNEL,
> 
> Frediano
> _______________________________________________
> Spice-devel mailing list
> Spice-devel at lists.freedesktop.org
> https://lists.freedesktop.org/mailman/listinfo/spice-devel


More information about the Spice-devel mailing list