server/red_channel: introduce urgent marshaller

When red_channel::red_channel_client_begin_send_message is called,
the message that is pending in the urgent marshaller will be sent before
the one in the main channel.
The urgent marshaller should be used if in the middle of marshalling one message,
you find out you need to send another message before. This functionality
is equivalent to the sub_list messages. It will replace them in the following
patches, when sub_list is removed from Spice data header.

2 files changed, 85 insertions, 6 deletions

diff --git a/server/red_channel.c b/server/red_channel.c
index 2ce0094c..671bcf58 100644
--- a/server/red_channel.c
+++ b/server/red_channel.c
@@ -38,6 +38,7 @@
 static void red_channel_client_event(int fd, int event, void *data);
 static void red_client_add_channel(RedClient *client, RedChannelClient *rcc);
 static void red_client_remove_channel(RedChannelClient *rcc);
+static void red_channel_client_restore_main_sender(RedChannelClient *rcc);
 
 /* return the number of bytes read. -1 in case of error */
 static int red_peer_receive(RedsStream *stream, uint8_t *buf, uint32_t size)
@@ -204,10 +205,13 @@ static void red_peer_handle_outgoing(RedsStream *stream, OutgoingHandler *handle
             handler->pos += n;
             handler->cb->on_output(handler->opaque, n);
             if (handler->pos == handler->size) { // finished writing data
-                handler->cb->on_msg_done(handler->opaque);
+                /* reset handler before calling on_msg_done, since it
+                 * can trigger another call to red_peer_handle_outgoing (when
+                 * switching from the urgent marshaller to the main one */
                 handler->vec = handler->vec_buf;
                 handler->pos = 0;
                 handler->size = 0;
+                handler->cb->on_msg_done(handler->opaque);
                 return;
             }
         }
@@ -252,6 +256,11 @@ static void red_channel_client_peer_on_out_block(void *opaque)
                                      SPICE_WATCH_EVENT_WRITE);
 }
 
+static inline int red_channel_client_urgent_marshaller_is_active(RedChannelClient *rcc)
+{
+    return (rcc->send_data.marshaller == rcc->send_data.urgent.marshaller);
+}
+
 static void red_channel_client_reset_send_data(RedChannelClient *rcc)
 {
     spice_marshaller_reset(rcc->send_data.marshaller);
@@ -261,7 +270,16 @@ static void red_channel_client_reset_send_data(RedChannelClient *rcc)
     rcc->send_data.header->type = 0;
     rcc->send_data.header->size = 0;
     rcc->send_data.header->sub_list = 0;
-    rcc->send_data.header->serial = ++rcc->send_data.serial;
+
+    if (!red_channel_client_urgent_marshaller_is_active(rcc)) {
+        rcc->send_data.header->serial = ++rcc->send_data.serial;
+    } else {
+        /*  The serial was incremented by the call to reset_send_data
+         *  that was done for the main marshaller. The urgent msg should
+         *  receive this serial, and the main msg serial should be
+         *  the following one. */
+        rcc->send_data.header->serial = rcc->send_data.serial++;
+    }
 }
 
 void red_channel_client_push_set_ack(RedChannelClient *rcc)
@@ -343,6 +361,12 @@ static void red_channel_peer_on_out_msg_done(void *opaque)
         rcc->channel->core->watch_update_mask(rcc->stream->watch,
                                          SPICE_WATCH_EVENT_READ);
     }
+
+    if (red_channel_client_urgent_marshaller_is_active(rcc)) {
+        red_channel_client_restore_main_sender(rcc);
+        ASSERT(rcc->send_data.header != NULL);
+        red_channel_client_begin_send_message(rcc);
+    }
 }
 
 static void red_channel_client_pipe_remove(RedChannelClient *rcc, PipeItem *item)
@@ -407,7 +431,10 @@ RedChannelClient *red_channel_client_create(int size, RedChannel *channel, RedCl
                                              // block flags)
     rcc->ack_data.client_generation = ~0;
     rcc->ack_data.client_window = CLIENT_ACK_WINDOW;
-    rcc->send_data.marshaller = spice_marshaller_new();
+    rcc->send_data.main.marshaller = spice_marshaller_new();
+    rcc->send_data.urgent.marshaller = spice_marshaller_new();
+
+    rcc->send_data.marshaller = rcc->send_data.main.marshaller;
 
     rcc->incoming.opaque = rcc;
     rcc->incoming.cb = &channel->incoming_cb;
@@ -643,9 +670,14 @@ void red_channel_client_destroy(RedChannelClient *rcc)
         red_channel_client_disconnect(rcc);
     }
     red_client_remove_channel(rcc);
-    if (rcc->send_data.marshaller) {
-        spice_marshaller_destroy(rcc->send_data.marshaller);
+    if (rcc->send_data.main.marshaller) {
+        spice_marshaller_destroy(rcc->send_data.main.marshaller);
+    }
+
+    if (rcc->send_data.urgent.marshaller) {
+        spice_marshaller_destroy(rcc->send_data.urgent.marshaller);
     }
+
     red_channel_client_destroy_remote_caps(rcc);
     free(rcc);
 }
@@ -874,6 +906,28 @@ void red_channel_client_begin_send_message(RedChannelClient *rcc)
     red_channel_client_send(rcc);
 }
 
+SpiceMarshaller *red_channel_client_switch_to_urgent_sender(RedChannelClient *rcc)
+{
+    ASSERT(red_channel_client_no_item_being_sent(rcc));
+    ASSERT(rcc->send_data.header != NULL);
+    rcc->send_data.main.header = rcc->send_data.header;
+    rcc->send_data.main.item = rcc->send_data.item;
+
+    rcc->send_data.marshaller = rcc->send_data.urgent.marshaller;
+    rcc->send_data.item = NULL;
+    red_channel_client_reset_send_data(rcc);
+    return rcc->send_data.marshaller;
+}
+
+static void red_channel_client_restore_main_sender(RedChannelClient *rcc)
+{
+    spice_marshaller_reset(rcc->send_data.urgent.marshaller);
+    rcc->send_data.marshaller = rcc->send_data.main.marshaller;
+    rcc->send_data.header = rcc->send_data.main.header;
+    rcc->send_data.header->serial = rcc->send_data.serial;
+    rcc->send_data.item = rcc->send_data.main.item;
+}
+
 uint64_t red_channel_client_get_message_serial(RedChannelClient *rcc)
 {
     return rcc->send_data.serial;
diff --git a/server/red_channel.h b/server/red_channel.h
index cce69658..cb80100e 100644
--- a/server/red_channel.h
+++ b/server/red_channel.h
@@ -207,6 +207,17 @@ struct RedChannelClient {
         PipeItem *item;
         int blocked;
         uint64_t serial;
+
+        struct {
+            SpiceMarshaller *marshaller;
+            SpiceDataHeader *header;
+            PipeItem *item;
+        } main;
+
+        struct {
+            SpiceMarshaller *marshaller;
+            SpiceDataHeader *header;
+        } urgent;
     } send_data;
 
     OutgoingHandler outgoing;
@@ -331,9 +342,23 @@ void red_channel_client_init_send_data(RedChannelClient *rcc, uint16_t msg_type,
 uint64_t red_channel_client_get_message_serial(RedChannelClient *channel);
 void red_channel_client_set_message_serial(RedChannelClient *channel, uint64_t);
 
-/* when sending a msg. should first call red_channel_client_begin_send_message */
+/* When sending a msg. Should first call red_channel_client_begin_send_message.
+ * It will first send the pending urgent data, if there is any, and then
+ * the rest of the data.
+ */
 void red_channel_client_begin_send_message(RedChannelClient *rcc);
 
+/*
+ * Stores the current send data, and switches to urgent send data.
+ * When it begins the actual send, it will send first the urgent data
+ * and afterward the rest of the data.
+ * Should be called only if during the marshalling of on message,
+ * the need to send another message, before, rises.
+ * Important: the serial of the non-urgent sent data, will be succeeded.
+ * return: the urgent send data marshaller
+ */
+SpiceMarshaller *red_channel_client_switch_to_urgent_sender(RedChannelClient *rcc);
+
 void red_channel_pipe_item_init(RedChannel *channel, PipeItem *item, int type);
 
 // TODO: add back the channel_pipe_add functionality - by adding reference counting