red_channel: cleanup of red_channel_client blocking methods

(1) receive timeout as a parameter.
(2) add a return value and pass the handling
    of failures to the calling routine.

3 files changed, 93 insertions, 57 deletions

diff --git a/server/red_channel.c b/server/red_channel.c
index 961c36c2..2cef2be2 100644
--- a/server/red_channel.c
+++ b/server/red_channel.c
@@ -51,11 +51,7 @@ typedef struct EmptyMsgPipeItem {
 #define PING_TEST_TIMEOUT_MS 15000
 #define PING_TEST_IDLE_NET_TIMEOUT_MS 100
 
-#define DETACH_TIMEOUT 15000000000ULL //nano
-#define DETACH_SLEEP_DURATION 10000 //micro
-
-#define CHANNEL_PUSH_TIMEOUT 30000000000ULL //nano
-#define CHANNEL_PUSH_SLEEP_DURATION 10000 //micro
+#define CHANNEL_BLOCKED_SLEEP_DURATION 10000 //micro
 
 enum QosPingState {
     PING_STATE_NONE,
@@ -2329,43 +2325,49 @@ uint32_t red_channel_sum_pipes_size(RedChannel *channel)
     return sum;
 }
 
-void red_wait_outgoing_item(RedChannelClient *rcc)
+int red_channel_client_wait_outgoing_item(RedChannelClient *rcc,
+                                          int64_t timeout)
 {
     uint64_t end_time;
     int blocked;
 
     if (!red_channel_client_blocked(rcc)) {
-        return;
+        return TRUE;
+    }
+    if (timeout != -1) {
+        end_time = red_now() + timeout;
     }
-    end_time = red_now() + DETACH_TIMEOUT;
     spice_info("blocked");
 
     do {
-        usleep(DETACH_SLEEP_DURATION);
+        usleep(CHANNEL_BLOCKED_SLEEP_DURATION);
         red_channel_client_receive(rcc);
         red_channel_client_send(rcc);
-    } while ((blocked = red_channel_client_blocked(rcc)) && red_now() < end_time);
+    } while ((blocked = red_channel_client_blocked(rcc)) &&
+             (timeout == -1 || red_now() < end_time));
 
     if (blocked) {
         spice_warning("timeout");
-        // TODO - shutting down the socket but we still need to trigger
-        // disconnection. Right now we wait for main channel to error for that.
-        red_channel_client_shutdown(rcc);
+        return FALSE;
     } else {
         spice_assert(red_channel_client_no_item_being_sent(rcc));
+        return TRUE;
     }
 }
 
 /* TODO: more evil sync stuff. anything with the word wait in it's name. */
-void red_channel_client_wait_pipe_item_sent(RedChannelClient *rcc,
-                                            PipeItem *item)
+int red_channel_client_wait_pipe_item_sent(RedChannelClient *rcc,
+                                           PipeItem *item,
+                                           int64_t timeout)
 {
     uint64_t end_time;
     int item_in_pipe;
 
     spice_info(NULL);
 
-    end_time = red_now() + CHANNEL_PUSH_TIMEOUT;
+    if (timeout != -1) {
+        end_time = red_now() + timeout;
+    }
 
     rcc->channel->channel_cbs.hold_item(rcc, item);
 
@@ -2375,55 +2377,52 @@ void red_channel_client_wait_pipe_item_sent(RedChannelClient *rcc,
     }
     red_channel_client_push(rcc);
 
-    while((item_in_pipe = ring_item_is_linked(&item->link)) && (red_now() < end_time)) {
-        usleep(CHANNEL_PUSH_SLEEP_DURATION);
+    while((item_in_pipe = ring_item_is_linked(&item->link)) &&
+          (timeout == -1 || red_now() < end_time)) {
+        usleep(CHANNEL_BLOCKED_SLEEP_DURATION);
         red_channel_client_receive(rcc);
         red_channel_client_send(rcc);
         red_channel_client_push(rcc);
     }
 
+    red_channel_client_release_item(rcc, item, TRUE);
     if (item_in_pipe) {
         spice_warning("timeout");
-        red_channel_client_disconnect(rcc);
-    } else {
-        red_wait_outgoing_item(rcc);
-    }
-    red_channel_client_release_item(rcc, item, TRUE);
-}
-
-static void rcc_shutdown_if_pending_send(RedChannelClient *rcc)
-{
-    if (red_channel_client_blocked(rcc) || rcc->pipe_size > 0) {
-        red_channel_client_shutdown(rcc);
+        return FALSE;
     } else {
-        spice_assert(red_channel_client_no_item_being_sent(rcc));
+        return red_channel_client_wait_outgoing_item(rcc,
+                                                     timeout == -1 ? -1 : end_time - red_now());
     }
 }
 
-void red_channel_wait_all_sent(RedChannel *channel)
+int red_channel_wait_all_sent(RedChannel *channel,
+                              int64_t timeout)
 {
     uint64_t end_time;
     uint32_t max_pipe_size;
     int blocked = FALSE;
 
-    end_time = red_now() + DETACH_TIMEOUT;
+    if (timeout != -1) {
+        end_time = red_now() + timeout;
+    }
 
     red_channel_push(channel);
     while (((max_pipe_size = red_channel_max_pipe_size(channel)) ||
            (blocked = red_channel_any_blocked(channel))) &&
-           red_now() < end_time) {
+           (timeout == -1 || red_now() < end_time)) {
         spice_debug("pipe-size %u blocked %d", max_pipe_size, blocked);
-        usleep(DETACH_SLEEP_DURATION);
+        usleep(CHANNEL_BLOCKED_SLEEP_DURATION);
         red_channel_receive(channel);
         red_channel_send(channel);
         red_channel_push(channel);
     }
 
     if (max_pipe_size || blocked) {
-        spice_printerr("timeout: pending out messages exist (pipe-size %u, blocked %d)",
-                       max_pipe_size, blocked);
-        red_channel_apply_clients(channel, rcc_shutdown_if_pending_send);
+        spice_warning("timeout: pending out messages exist (pipe-size %u, blocked %d)",
+                      max_pipe_size, blocked);
+        return FALSE;
     } else {
         spice_assert(red_channel_no_item_being_sent(channel));
+        return TRUE;
     }
 }
diff --git a/server/red_channel.h b/server/red_channel.h
index 676e1ef5..9e54dcec 100644
--- a/server/red_channel.h
+++ b/server/red_channel.h
@@ -612,14 +612,20 @@ int red_client_during_migrate_at_target(RedClient *client);
 
 void red_client_migrate(RedClient *client);
 
-/* blocking function */
-void red_channel_client_wait_pipe_item_sent(RedChannelClient *rcc,
-                                            PipeItem *item);
-
-/* blocking function */
-void red_wait_outgoing_item(RedChannelClient *rcc);
+/*
+ * blocking functions.
+ *
+ * timeout is in nano sec. -1 for no timeout.
+ *
+ * Return: TRUE if waiting succeeded. FALSE if timeout expired.
+ */
 
-/* blocking function */
-void red_channel_wait_all_sent(RedChannel *channel);
+int red_channel_client_wait_pipe_item_sent(RedChannelClient *rcc,
+                                           PipeItem *item,
+                                           int64_t timeout);
+int red_channel_client_wait_outgoing_item(RedChannelClient *rcc,
+                                          int64_t timeout);
+int red_channel_wait_all_sent(RedChannel *channel,
+                              int64_t timeout);
 
 #endif
diff --git a/server/red_worker.c b/server/red_worker.c
index 9cfacfa8..dea73253 100644
--- a/server/red_worker.c
+++ b/server/red_worker.c
@@ -102,6 +102,7 @@
 #define CMD_RING_POLL_TIMEOUT 10 //milli
 #define CMD_RING_POLL_RETRIES 200
 
+#define DISPLAY_CLIENT_SHORT_TIMEOUT 15000000000ULL //nano
 #define DISPLAY_CLIENT_TIMEOUT 30000000000ULL //nano
 #define DISPLAY_CLIENT_MIGRATE_DATA_TIMEOUT 10000000000ULL //nano, 10 sec
 #define DISPLAY_CLIENT_RETRY_INTERVAL 10000 //micro
@@ -2003,8 +2004,12 @@ static void red_current_clear(RedWorker *worker, int surface_id)
     }
 }
 
-static void red_clear_surface_drawables_from_pipe(DisplayChannelClient *dcc, int surface_id,
-                                                  int wait_if_used)
+/*
+ * Return: TRUE if wait_if_used == FALSE, or otherwise, if all of the pipe items that
+ * are related to the surface have been cleared (or sent) from the pipe.
+ */
+static int red_clear_surface_drawables_from_pipe(DisplayChannelClient *dcc, int surface_id,
+                                                 int wait_if_used)
 {
     Ring *ring;
     PipeItem *item;
@@ -2012,7 +2017,7 @@ static void red_clear_surface_drawables_from_pipe(DisplayChannelClient *dcc, int
     RedChannelClient *rcc;
 
     if (!dcc) {
-        return;
+        return TRUE;
     }
 
     /* removing the newest drawables that their destination is surface_id and
@@ -2057,24 +2062,27 @@ static void red_clear_surface_drawables_from_pipe(DisplayChannelClient *dcc, int
             if (wait_if_used) {
                 break;
             } else {
-                return;
+                return TRUE;
             }
         }
     }
 
     if (!wait_if_used) {
-        return;
+        return TRUE;
     }
 
     if (item) {
-        red_channel_client_wait_pipe_item_sent(&dcc->common.base, item);
+        return red_channel_client_wait_pipe_item_sent(&dcc->common.base, item,
+                                                      DISPLAY_CLIENT_TIMEOUT);
     } else {
         /*
          * in case that the pipe didn't contain any item that is dependent on the surface, but
-         * there is one during sending.
+         * there is one during sending. Use a shorter timeout, since it is just one item
          */
-        red_wait_outgoing_item(&dcc->common.base);
+        return red_channel_client_wait_outgoing_item(&dcc->common.base,
+                                                     DISPLAY_CLIENT_SHORT_TIMEOUT);
     }
+    return TRUE;
 }
 
 static void red_clear_surface_drawables_from_pipes(RedWorker *worker,
@@ -2085,7 +2093,9 @@ static void red_clear_surface_drawables_from_pipes(RedWorker *worker,
     DisplayChannelClient *dcc;
 
     WORKER_FOREACH_DCC_SAFE(worker, item, next, dcc) {
-        red_clear_surface_drawables_from_pipe(dcc, surface_id, wait_if_used);
+        if (!red_clear_surface_drawables_from_pipe(dcc, surface_id, wait_if_used)) {
+            red_channel_client_disconnect(&dcc->common.base);
+        }
     }
 }
 
@@ -10949,6 +10959,15 @@ void handle_dev_destroy_surface_wait(void *opaque, void *payload)
     dev_destroy_surface_wait(worker, msg->surface_id);
 }
 
+static void rcc_shutdown_if_pending_send(RedChannelClient *rcc)
+{
+    if (red_channel_client_blocked(rcc) || rcc->pipe_size > 0) {
+        red_channel_client_shutdown(rcc);
+    } else {
+        spice_assert(red_channel_client_no_item_being_sent(rcc));
+    }
+}
+
 static inline void red_cursor_reset(RedWorker *worker)
 {
     if (worker->cursor) {
@@ -10966,7 +10985,11 @@ static inline void red_cursor_reset(RedWorker *worker)
         if (!worker->cursor_channel->common.during_target_migrate) {
             red_pipes_add_verb(&worker->cursor_channel->common.base, SPICE_MSG_CURSOR_RESET);
         }
-        red_channel_wait_all_sent(&worker->cursor_channel->common.base);
+        if (!red_channel_wait_all_sent(&worker->cursor_channel->common.base,
+                                       DISPLAY_CLIENT_TIMEOUT)) {
+            red_channel_apply_clients(&worker->cursor_channel->common.base,
+                                      rcc_shutdown_if_pending_send);
+        }
     }
 }
 
@@ -11249,8 +11272,16 @@ void handle_dev_stop(void *opaque, void *payload)
      * purge the pipe, send destroy_all_surfaces
      * to the client (there is no such message right now), and start
      * from scratch on the destination side */
-    red_channel_wait_all_sent(&worker->display_channel->common.base);
-    red_channel_wait_all_sent(&worker->cursor_channel->common.base);
+    if (!red_channel_wait_all_sent(&worker->display_channel->common.base,
+                                   DISPLAY_CLIENT_TIMEOUT)) {
+        red_channel_apply_clients(&worker->display_channel->common.base,
+                                  rcc_shutdown_if_pending_send);
+    }
+    if (!red_channel_wait_all_sent(&worker->cursor_channel->common.base,
+                                   DISPLAY_CLIENT_TIMEOUT)) {
+        red_channel_apply_clients(&worker->cursor_channel->common.base,
+                                  rcc_shutdown_if_pending_send);
+    }
 }
 
 static int display_channel_wait_for_migrate_data(DisplayChannel *display)