Merge pull request #1706 from matt335672/neutrinordp-flow-control2

Fix neutrinordp buffering. Addresses #1634 and #1608

common/trans.c

@@ -302,7 +302,7 @@ trans_check_wait_objs(struct trans *self)
     int to_read = 0;
     int read_so_far = 0;
     int rv = 0;
-    int cur_source;
+    enum xrdp_source cur_source;
 
     if (self == 0)
     {
@@ -371,7 +371,7 @@ trans_check_wait_objs(struct trans *self)
         }
         else if (self->trans_can_recv(self, self->sck, 0))
         {
-            cur_source = 0;
+            cur_source = XRDP_SOURCE_NONE;
             if (self->si != 0)
             {
                 cur_source = self->si->cur_source;
@@ -633,7 +633,7 @@ trans_write_copy_s(struct trans *self, struct stream *out_s)
     init_stream(wait_s, size);
     if (self->si != 0)
     {
-        if ((self->si->cur_source != 0) &&
+        if ((self->si->cur_source != XRDP_SOURCE_NONE) &&
             (self->si->cur_source != self->my_source))
         {
             self->si->source[self->si->cur_source] += size;

common/trans.h

@@ -50,16 +50,43 @@ typedef int (*trans_can_recv_proc) (struct trans *self, int sck, int millis);
 
 /* optional source info */
 
-#define XRDP_SOURCE_NONE    0
-#define XRDP_SOURCE_CLIENT  1
-#define XRDP_SOURCE_SESMAN  2
-#define XRDP_SOURCE_CHANSRV 3
-#define XRDP_SOURCE_MOD     4
+enum xrdp_source
+{
+    XRDP_SOURCE_NONE = 0,
+    XRDP_SOURCE_CLIENT,
+    XRDP_SOURCE_SESMAN,
+    XRDP_SOURCE_CHANSRV,
+    XRDP_SOURCE_MOD,
 
+    XRDP_SOURCE_MAX_COUNT
+};
+
+/*
+ * @brief Provide flow control mechanism for (primarily) xrdp
+ *
+ * There is one of these data structures per-program.
+ *
+ * While input is being read from a 'struct trans' and processed, the
+ * cur_source member is set to the my_source member from the transport.
+ * During this processing, trans_write_copy() may be called to send output
+ * on another struct trans. If this happens, and the ouput needs to be
+ * buffered, trans_write_copy() can add the number of bytes generated by
+ * the input trans to the source field for the cur_source. This allows us to
+ * see how much output has been buffered for each input source.
+ *
+ * When the program assembles 'struct trans' objects to scan for input
+ * (normally in trans_get_wait_objs()), it is able to see how much buffered
+ * output is registered for each input. Inputs which have too much buffered
+ * output owing are skipped, and not considered for input.
+ *
+ * This provides a simple means of providing back-pressure on an input
+ * where the data it is providing is being processed and then sent out on
+ * a much slower link.
+ */
 struct source_info
 {
-    int cur_source;
-    int source[7];
+    enum xrdp_source cur_source;
+    int source[XRDP_SOURCE_MAX_COUNT];
 };
 
 struct trans
@@ -88,7 +115,7 @@ struct trans
     trans_send_proc trans_send;
     trans_can_recv_proc trans_can_recv;
     struct source_info *si;
-    int my_source;
+    enum xrdp_source my_source;
 };
 
 struct trans*

mc/mc.h

@@ -26,6 +26,8 @@
 
 #define CURRENT_MOD_VER 3
 
+struct source_info;
+
 struct mod
 {
   int size; /* size of this struct */
@@ -91,7 +93,7 @@ struct mod
   tintptr handle; /* pointer to self as long */
   tintptr wm;
   tintptr painter;
-  tintptr si;
+  struct source_info *si;
   /* mod data */
   int sck;
   int width;

neutrinordp/xrdp-neutrinordp.c

@@ -24,6 +24,7 @@
 #include "xrdp-neutrinordp.h"
 #include "xrdp-color.h"
 #include "xrdp_rail.h"
+#include "trans.h"
 #include "log.h"
 #include <freerdp/settings.h>
 
@@ -39,6 +40,9 @@
 #define LOG_LEVEL 1
 #endif
 
+/* Max amount of buffered output data before we stop generating more */
+#define MAX_QUEUED_MODULE_OUTPUT_DATA 50000
+
 #define LLOG(_level, _args) \
     do { if (_level < LOG_LEVEL) { g_write _args ; } } while (0)
 #define LLOGLN(_level, _args) \
@@ -68,6 +72,13 @@ verifyColorMap(struct mod *mod)
     LLOGLN(0, ("The colormap is all NULL"));
 }
 
+/*****************************************************************************/
+static int
+get_queued_module_output_data(struct mod *mod)
+{
+    return (mod->si != NULL) ? mod->si->source[XRDP_SOURCE_MOD] : 0;
+}
+
 /*****************************************************************************/
 /* return error */
 static int
@@ -516,14 +527,26 @@ lxrdp_get_wait_objs(struct mod *mod, tbus *read_objs, int *rcount,
     boolean ok;
 
     LLOGLN(12, ("lxrdp_get_wait_objs:"));
-    rfds = (void **)read_objs;
-    wfds = (void **)write_objs;
-    ok = freerdp_get_fds(mod->inst, rfds, rcount, wfds, wcount);
-
-    if (!ok)
+    /*
+     * Don't check this module for activity if our queued output data
+     * has already reached the limit
+     */
+    if (get_queued_module_output_data(mod) > MAX_QUEUED_MODULE_OUTPUT_DATA)
     {
-        LLOGLN(0, ("lxrdp_get_wait_objs: freerdp_get_fds failed"));
-        return 1;
+        *rcount = 0;
+        *wcount = 0;
+    }
+    else
+    {
+        rfds = (void **)read_objs;
+        wfds = (void **)write_objs;
+        ok = freerdp_get_fds(mod->inst, rfds, rcount, wfds, wcount);
+
+        if (!ok)
+        {
+            LLOGLN(0, ("lxrdp_get_wait_objs: freerdp_get_fds failed"));
+            return 1;
+        }
     }
 
     return 0;
@@ -536,12 +559,32 @@ lxrdp_check_wait_objs(struct mod *mod)
     boolean ok;
 
     LLOGLN(12, ("lxrdp_check_wait_objs:"));
-    ok = freerdp_check_fds(mod->inst);
-
-    if (!ok)
+    /*
+     * Only process the freerdp file descriptors if our queued output data
+     * has not reached the limit
+     */
+    if (get_queued_module_output_data(mod) <= MAX_QUEUED_MODULE_OUTPUT_DATA)
     {
-        LLOGLN(0, ("lxrdp_check_wait_objs: freerdp_check_fds failed"));
-        return 1;
+        /*
+         * Before checking the file descriptors, set the source info
+         * current source, so any data queued on output trans objects
+         * gets attributed to this module
+         */
+        if (mod->si)
+        {
+            mod->si->cur_source = XRDP_SOURCE_MOD;
+        }
+        ok = freerdp_check_fds(mod->inst);
+        if (mod->si)
+        {
+            mod->si->cur_source = XRDP_SOURCE_NONE;
+        }
+
+        if (!ok)
+        {
+            LLOGLN(0, ("lxrdp_check_wait_objs: freerdp_check_fds failed"));
+            return 1;
+        }
     }
 
     return 0;

neutrinordp/xrdp-neutrinordp.h

@@ -61,6 +61,8 @@ struct pointer_item
 
 #define CURRENT_MOD_VER 4
 
+struct source_info;
+
 struct mod
 {
     int size; /* size of this struct */
@@ -183,7 +185,7 @@ struct mod
     tintptr handle; /* pointer to self as long */
     tintptr wm;
     tintptr painter;
-    tintptr si;
+    struct source_info *si;
 
     /* mod data */
     int sck;

vnc/vnc.c

@@ -1905,7 +1905,6 @@ lib_mod_connect(struct vnc *v)
     int error;
     int i;
     int check_sec_result;
-    struct source_info *si;
 
     v->server_msg(v, "VNC started connecting", 0);
     check_sec_result = 1;
@@ -1955,8 +1954,7 @@ lib_mod_connect(struct vnc *v)
     g_sprintf(text, "VNC connecting to %s %s", v->ip, con_port);
     v->server_msg(v, text, 0);
 
-    si = (struct source_info *) (v->si);
-    v->trans->si = si;
+    v->trans->si = v->si;
     v->trans->my_source = XRDP_SOURCE_MOD;
 
     error = trans_connect(v->trans, v->ip, con_port, 3000);

vnc/vnc.h

@@ -56,6 +56,8 @@ enum vnc_resize_status
     VRS_DONE
 };
 
+struct source_info;
+
 struct vnc
 {
     int size; /* size of this struct */
@@ -124,7 +126,7 @@ struct vnc
     tintptr handle; /* pointer to self as long */
     tintptr wm;
     tintptr painter;
-    tintptr si;
+    struct source_info *si;
     /* mod data */
     int server_width;
     int server_height;

xrdp/xrdp_mm.c

@@ -471,7 +471,7 @@ xrdp_mm_setup_mod1(struct xrdp_mm *self)
             self->mod->server_composite = server_composite;
             self->mod->server_paint_rects = server_paint_rects;
             self->mod->server_session_info = server_session_info;
-            self->mod->si = (tintptr) &(self->wm->session->si);
+            self->mod->si = &(self->wm->session->si);
         }
     }
 

xrdp/xrdp_types.h

@@ -30,6 +30,9 @@
 
 #define MAX_NR_CHANNELS 16
 #define MAX_CHANNEL_NAME 16
+
+struct source_info;
+
 /* lib */
 struct xrdp_mod
 {
@@ -156,7 +159,7 @@ struct xrdp_mod
   tintptr handle; /* pointer to self as int */
   tintptr wm; /* struct xrdp_wm* */
   tintptr painter;
-  tintptr si;
+  struct source_info *si;
 };
 
 /* header for bmp file */

xup/xup.c

@@ -149,7 +149,6 @@ lib_mod_connect(struct mod *mod)
     int use_uds;
     struct stream *s;
     char con_port[256];
-    struct source_info *si;
 
     LIB_DEBUG(mod, "in lib_mod_connect");
 
@@ -203,8 +202,7 @@ lib_mod_connect(struct mod *mod)
         }
     }
 
-    si = (struct source_info *) (mod->si);
-    mod->trans->si = si;
+    mod->trans->si = mod->si;
     mod->trans->my_source = XRDP_SOURCE_MOD;
 
     while (1)

xup/xup.h

@@ -29,6 +29,8 @@
 
 #define CURRENT_MOD_VER 4
 
+struct source_info;
+
 struct mod
 {
   int size; /* size of this struct */
@@ -149,7 +151,7 @@ struct mod
   tintptr handle; /* pointer to self as long */
   tintptr wm;
   tintptr painter;
-  tintptr si;
+  struct source_info *si;
   /* mod data */
   int width;
   int height;