started work on fastpath, restructure PDU in

libxrdp/Makefile.am

@@ -55,7 +55,6 @@ libxrdp_la_SOURCES = \
   xrdp_orders.c \
   xrdp_rdp.c \
   xrdp_sec.c \
-  xrdp_tcp.c \
   xrdp_bitmap_compress.c \
   xrdp_bitmap32_compress.c \
   xrdp_jpeg_compress.c \

libxrdp/libxrdp.c

@@ -32,8 +32,6 @@ libxrdp_init(tbus id, struct trans *trans)
     session->rdp = xrdp_rdp_create(session, trans);
     session->orders = xrdp_orders_create(session, (struct xrdp_rdp *)session->rdp);
     session->client_info = &(((struct xrdp_rdp *)session->rdp)->client_info);
-    make_stream(session->s);
-    init_stream(session->s, 8192 * 2);
     return session;
 }
 
@@ -48,7 +46,6 @@ libxrdp_exit(struct xrdp_session *session)
 
     xrdp_orders_delete((struct xrdp_orders *)session->orders);
     xrdp_rdp_delete((struct xrdp_rdp *)session->rdp);
-    free_stream(session->s);
     g_free(session);
     return 0;
 }
@@ -69,19 +66,38 @@ libxrdp_process_incomming(struct xrdp_session *session)
 
 /******************************************************************************/
 int EXPORT_CC
-libxrdp_process_data(struct xrdp_session *session)
+libxrdp_process_data(struct xrdp_session *session, struct stream *s)
 {
     int cont;
     int rv;
     int code;
     int term;
     int dead_lock_counter;
+    struct xrdp_rdp *rdp;
+    struct stream *ls;
+
+    if (session->in_process_data != 0)
+    {
+        g_writeln("libxrdp_process_data: error reentry");
+        return 1;
+    }
+    session->in_process_data++;
+
+    ls = 0;
+    if (s == 0)
+    {
+        make_stream(ls);
+        init_stream(ls, 8192 * 4);
+        s = ls;
+    }
 
     term = 0;
     cont = 1;
     rv = 0;
     dead_lock_counter = 0;
 
+    rdp = (struct xrdp_rdp *) (session->rdp);
+
     while ((cont || !session->up_and_running) && !term)
     {
         if (session->is_term != 0)
@@ -94,8 +110,7 @@ libxrdp_process_data(struct xrdp_session *session)
 
         code = 0;
 
-        if (xrdp_rdp_recv((struct xrdp_rdp *)(session->rdp),
+        if (xrdp_rdp_recv(rdp, s, &code) != 0)
-                          session->s, &code) != 0)
         {
             rv = 1;
             break;
@@ -106,13 +121,14 @@ libxrdp_process_data(struct xrdp_session *session)
         switch (code)
         {
             case -1:
-                xrdp_rdp_send_demand_active((struct xrdp_rdp *)session->rdp);
+                xrdp_rdp_send_demand_active(rdp);
 
-                // send Monitor Layout PDU for multimon
+                /* send Monitor Layout PDU for multimon */
-                if (session->client_info->monitorCount > 0 && session->client_info->multimon == 1)
+                if (session->client_info->monitorCount > 0 &&
+                    session->client_info->multimon == 1)
                 {
                     DEBUG(("sending monitor layout pdu"));
-                    if (xrdp_rdp_send_monitorlayout((struct xrdp_rdp *)session->rdp) != 0)
+                    if (xrdp_rdp_send_monitorlayout(rdp) != 0)
                     {
                       g_writeln("xrdp_rdp_send_monitorlayout: error");
                     }
@@ -124,13 +140,11 @@ libxrdp_process_data(struct xrdp_session *session)
                 dead_lock_counter++;
                 break;
             case RDP_PDU_CONFIRM_ACTIVE: /* 3 */
-                xrdp_rdp_process_confirm_active((struct xrdp_rdp *)session->rdp,
+                xrdp_rdp_process_confirm_active(rdp, s);
-                                                session->s);
                 break;
             case RDP_PDU_DATA: /* 7 */
 
-                if (xrdp_rdp_process_data((struct xrdp_rdp *)session->rdp,
+                if (xrdp_rdp_process_data(rdp, s) != 0)
-                                          session->s) != 0)
                 {
                     DEBUG(("libxrdp_process_data returned non zero"));
                     cont = 0;
@@ -149,17 +163,24 @@ libxrdp_process_data(struct xrdp_session *session)
             /*This situation can happen and this is a workaround*/
             cont = 0;
             g_writeln("Serious programming error we were locked in a deadly loop") ;
-            g_writeln("remaining :%d", session->s->end - session->s->next_packet);
+            g_writeln("remaining :%d", s->end - s->next_packet);
-            session->s->next_packet = 0;
+            s->next_packet = 0;
         }
 
         if (cont)
         {
-            cont = (session->s->next_packet != 0) &&
+            cont = (s->next_packet != 0) &&
-                   (session->s->next_packet < session->s->end);
+                   (s->next_packet < s->end);
         }
     }
 
+    if (s == ls)
+    {
+        free_stream(s);
+    }
+
+    session->in_process_data--;
+
     return rv;
 }
 
@@ -769,7 +790,7 @@ libxrdp_reset(struct xrdp_session *session,
     /* process till up and running */
     session->up_and_running = 0;
 
-    if (libxrdp_process_data(session) != 0)
+    if (libxrdp_process_data(session, 0) != 0)
     {
         g_writeln("non handled error from libxrdp_process_data");
     }

libxrdp/libxrdp.h

@@ -37,20 +37,13 @@
 #include "file_loc.h"
 #include "xrdp_client_info.h"
 
-/* tcp */
-struct xrdp_tcp
-{
-  struct trans* trans;
-  struct xrdp_iso* iso_layer; /* owner */
-};
-
 /* iso */
 struct xrdp_iso
 {
   struct xrdp_mcs* mcs_layer; /* owner */
-  struct xrdp_tcp* tcp_layer;
   int requestedProtocol;
   int selectedProtocol;
+  struct trans* trans;
 };
 
 /* used in mcs */

libxrdp/libxrdpinc.h

@@ -70,8 +70,8 @@ struct xrdp_session
   void* orders;
   struct xrdp_client_info* client_info;
   int up_and_running;
-  struct stream* s;
   int (*is_term)(void);
+  int in_process_data; /* inc / dec libxrdp_process_data calls */
 };
 
 struct xrdp_session* DEFAULT_CC
@@ -83,7 +83,7 @@ libxrdp_disconnect(struct xrdp_session* session);
 int DEFAULT_CC
 libxrdp_process_incomming(struct xrdp_session* session);
 int DEFAULT_CC
-libxrdp_process_data(struct xrdp_session* session);
+libxrdp_process_data(struct xrdp_session* session, struct stream *s);
 int DEFAULT_CC
 libxrdp_send_palette(struct xrdp_session* session, int* palette);
 int DEFAULT_CC

libxrdp/xrdp_iso.c

@@ -30,7 +30,7 @@ xrdp_iso_create(struct xrdp_mcs *owner, struct trans *trans)
     DEBUG(("   in xrdp_iso_create"));
     self = (struct xrdp_iso *)g_malloc(sizeof(struct xrdp_iso), 1);
     self->mcs_layer = owner;
-    self->tcp_layer = xrdp_tcp_create(self, trans);
+    self->trans = trans;
     DEBUG(("   out xrdp_iso_create"));
     return self;
 }
@@ -44,7 +44,6 @@ xrdp_iso_delete(struct xrdp_iso *self)
         return;
     }
 
-    xrdp_tcp_delete(self->tcp_layer);
     g_free(self);
 }
 
@@ -91,13 +90,21 @@ xrdp_iso_recv_msg(struct xrdp_iso *self, struct stream *s, int *code, int *len)
 {
     int ver;  // TPKT Version
     int plen; // TPKT PacketLength
+    int do_read;
 
     *code = 0; // X.224 Packet Type
     *len = 0;  // X.224 Length Indicator
 
-    if (xrdp_tcp_recv(self->tcp_layer, s, 4) != 0)
+    /* early in connection sequence, iso needs to do a force read */
+    do_read = s != self->trans->in_s;
+
+    if (do_read)
     {
-        return 1;
+        init_stream(s, 4);
+        if (trans_force_read_s(self->trans, s, 4) != 0)
+        {
+            return 1;
+        }
     }
 
     in_uint8(s, ver);
@@ -115,9 +122,13 @@ xrdp_iso_recv_msg(struct xrdp_iso *self, struct stream *s, int *code, int *len)
         return 1;
     }
 
-    if (xrdp_tcp_recv(self->tcp_layer, s, plen - 4) != 0)
+    if (do_read)
     {
-        return 1;
+        init_stream(s, plen - 4);
+        if (trans_force_read_s(self->trans, s, plen - 4) != 0)
+        {
+            return 1;
+        }
     }
 
     if (!s_check_rem(s, 2))
@@ -147,6 +158,7 @@ xrdp_iso_recv_msg(struct xrdp_iso *self, struct stream *s, int *code, int *len)
 
     return 0;
 }
+
 /*****************************************************************************/
 /* returns error */
 int APP_CC
@@ -177,10 +189,7 @@ xrdp_iso_recv(struct xrdp_iso *self, struct stream *s)
 static int APP_CC
 xrdp_iso_send_rdpnegrsp(struct xrdp_iso *self, struct stream *s, int code)
 {
-    if (xrdp_tcp_init(self->tcp_layer, s) != 0)
+    init_stream(s, 8192 * 4); /* 32 KB */
-    {
-        return 1;
-    }
 
     /* TPKT HEADER - 4 bytes */
     out_uint8(s, 3);    /* version */
@@ -217,7 +226,7 @@ xrdp_iso_send_rdpnegrsp(struct xrdp_iso *self, struct stream *s, int code)
 
     s_mark_end(s);
 
-    if (xrdp_tcp_send(self->tcp_layer, s) != 0)
+    if (trans_force_write_s(self->trans, s) != 0)
     {
         return 1;
     }
@@ -228,10 +237,7 @@ xrdp_iso_send_rdpnegrsp(struct xrdp_iso *self, struct stream *s, int code)
 static int APP_CC
 xrdp_iso_send_rdpnegfailure(struct xrdp_iso *self, struct stream *s, int code, int failureCode)
 {
-    if (xrdp_tcp_init(self->tcp_layer, s) != 0)
+    init_stream(s, 8192 * 4); /* 32 KB */
-    {
-        return 1;
-    }
 
     /* TPKT HEADER - 4 bytes */
     out_uint8(s, 3);    /* version */
@@ -250,7 +256,7 @@ xrdp_iso_send_rdpnegfailure(struct xrdp_iso *self, struct stream *s, int code, i
     out_uint32_le(s, failureCode); /* failure code */
     s_mark_end(s);
 
-    if (xrdp_tcp_send(self->tcp_layer, s) != 0)
+    if (trans_force_write_s(self->trans, s) != 0)
     {
         return 1;
     }
@@ -379,7 +385,7 @@ xrdp_iso_incoming(struct xrdp_iso *self)
 int APP_CC
 xrdp_iso_init(struct xrdp_iso *self, struct stream *s)
 {
-    xrdp_tcp_init(self->tcp_layer, s);
+    init_stream(s, 8192 * 4); /* 32 KB */
     s_push_layer(s, iso_hdr, 7);
     return 0;
 }
@@ -401,7 +407,7 @@ xrdp_iso_send(struct xrdp_iso *self, struct stream *s)
     out_uint8(s, ISO_PDU_DT);
     out_uint8(s, 0x80);
 
-    if (xrdp_tcp_send(self->tcp_layer, s) != 0)
+    if (trans_force_write_s(self->trans, s) != 0)
     {
         return 1;
     }

libxrdp/xrdp_mcs.c

@@ -75,8 +75,8 @@ xrdp_mcs_delete(struct xrdp_mcs *self)
 }
 
 /*****************************************************************************/
-/* This function sends channel join confirm*/
+/* This function sends channel join confirm */
-/* returns error = 1 ok = 0*/
+/* returns error = 1 ok = 0 */
 static int APP_CC
 xrdp_mcs_send_cjcf(struct xrdp_mcs *self, int userid, int chanid)
 {
@@ -151,6 +151,12 @@ xrdp_mcs_recv(struct xrdp_mcs *self, struct stream *s, int *chan)
         /* this is channels getting added from the client */
         if (appid == MCS_CJRQ)
         {
+            if (s == self->iso_layer->trans->in_s)
+            {
+                /* this should not happen */
+                g_writeln("xrdp_mcs_recv: error, MCS_CJRQ at wrong time");
+                return 1;
+            }
             if (!s_check_rem(s, 4))
             {
                 return 1;
@@ -165,13 +171,12 @@ xrdp_mcs_recv(struct xrdp_mcs *self, struct stream *s, int *chan)
             {
                 log_message(LOG_LEVEL_ERROR,"Non handled error from xrdp_mcs_send_cjcf") ;
             }
-
             continue;
         }
 
         if (appid == MCS_SDRQ || appid == MCS_SDIN)
         {
-            break ;
+            break;
         }
         else
         {
@@ -949,12 +954,12 @@ xrdp_mcs_send(struct xrdp_mcs *self, struct stream *s, int chan)
 void APP_CC
 close_rdp_socket(struct xrdp_mcs *self)
 {
-    if (self->iso_layer->tcp_layer != 0)
+    if (self->iso_layer != 0)
     {
-        if (self->iso_layer->tcp_layer->trans != 0)
+        if (self->iso_layer->trans != 0)
         {
-            g_tcp_close(self->iso_layer->tcp_layer->trans->sck);
+            g_tcp_close(self->iso_layer->trans->sck);
-            self->iso_layer->tcp_layer->trans->sck = 0 ;
+            self->iso_layer->trans->sck = 0 ;
             g_writeln("xrdp_mcs_disconnect - socket closed");
             return;
         }

libxrdp/xrdp_rdp.c

@@ -198,8 +198,10 @@ xrdp_rdp_create(struct xrdp_session *session, struct trans *trans)
     /* read ini settings */
     xrdp_rdp_read_config(&self->client_info);
     /* create sec layer */
-    self->sec_layer = xrdp_sec_create(self, trans, self->client_info.crypt_level,
+    self->sec_layer = xrdp_sec_create(self, trans,
-                                      self->client_info.channel_code, self->client_info.multimon);
+                                      self->client_info.crypt_level,
+                                      self->client_info.channel_code,
+                                      self->client_info.multimon);
     /* default 8 bit v1 color bitmap cache entries and size */
     self->client_info.cache1_entries = 600;
     self->client_info.cache1_size = 256;
@@ -503,10 +505,10 @@ xrdp_rdp_incoming(struct xrdp_rdp *self)
                         MCS_USERCHANNEL_BASE;
     DEBUG(("out xrdp_rdp_incoming mcs channel %d", self->mcs_channel));
     g_strncpy(self->client_info.client_addr,
-              self->sec_layer->mcs_layer->iso_layer->tcp_layer->trans->addr,
+              self->sec_layer->mcs_layer->iso_layer->trans->addr,
               sizeof(self->client_info.client_addr) - 1);
     g_strncpy(self->client_info.client_port,
-              self->sec_layer->mcs_layer->iso_layer->tcp_layer->trans->port,
+              self->sec_layer->mcs_layer->iso_layer->trans->port,
               sizeof(self->client_info.client_port) - 1);
     return 0;
 }
@@ -1568,6 +1570,7 @@ xrdp_rdp_process_data_font(struct xrdp_rdp *self, struct stream *s)
         xrdp_rdp_send_fontmap(self);
 
         self->session->up_and_running = 1;
+        g_writeln("yeah, up_and_running");
         DEBUG(("up_and_running set"));
         xrdp_rdp_send_data_update_sync(self);
     }

xrdp/xrdp_process.c

@@ -61,7 +61,7 @@ xrdp_process_delete(struct xrdp_process *self)
 
 /*****************************************************************************/
 static int APP_CC
-xrdp_process_loop(struct xrdp_process *self)
+xrdp_process_loop(struct xrdp_process *self, struct stream *s)
 {
     int rv;
 
@@ -69,7 +69,7 @@ xrdp_process_loop(struct xrdp_process *self)
 
     if (self->session != 0)
     {
-        rv = libxrdp_process_data(self->session);
+        rv = libxrdp_process_data(self->session, s);
     }
 
     if ((self->wm == 0) && (self->session->up_and_running) && (rv == 0))
@@ -114,18 +114,117 @@ xrdp_process_mod_end(struct xrdp_process *self)
     return 0;
 }
 
+/*****************************************************************************/
+static int APP_CC
+xrdp_process_get_pdu_bytes(const char *aheader)
+{
+    int rv;
+    const tui8 *header;
+
+    rv = -1;
+    header = (const tui8 *) aheader;
+    if (header[0] == 0x03)
+    {
+        /* TPKT */
+        rv = (header[2] << 8) | header[3];
+    }
+    else if (header[0] == 0x30)
+    {
+        /* TSRequest (NLA) */
+        if (header[1] & 0x80)
+        {
+            if ((header[1] & ~(0x80)) == 1)
+            {
+                rv = header[2];
+                rv += 3;
+            }
+            else if ((header[1] & ~(0x80)) == 2)
+            {
+                rv = (header[2] << 8) | header[3];
+                rv += 4;
+            }
+            else
+            {
+                g_writeln("xrdp_process_get_packet_bytes: error TSRequest!");
+                return -1;
+            }
+        }
+        else
+        {
+            rv = header[1];
+            rv += 2;
+        }
+    }
+    else
+    {
+        /* Fast-Path */
+        if (header[1] & 0x80)
+        {
+            rv = ((header[1] & 0x7F) << 8) | header[2];
+        }
+        else
+        {
+            rv = header[1];
+        }
+    }
+    return rv;
+}
+
 /*****************************************************************************/
 static int DEFAULT_CC
 xrdp_process_data_in(struct trans *self)
 {
     struct xrdp_process *pro;
+    struct stream *s;
+    int len;
 
     DEBUG(("xrdp_process_data_in"));
     pro = (struct xrdp_process *)(self->callback_data);
 
-    if (xrdp_process_loop(pro) != 0)
+    s = pro->server_trans->in_s;
+    switch (pro->server_trans->extra_flags)
     {
-        return 1;
+        case 0:
+            /* early in connection sequence, we're in this mode */
+            if (xrdp_process_loop(pro, 0) != 0)
+            {
+                g_writeln("xrdp_process_data_in: "
+                          "xrdp_process_loop failed");
+                return 1;
+            }
+            if (pro->session->up_and_running)
+            {
+                pro->server_trans->extra_flags = 1;
+                pro->server_trans->header_size = 4;
+            }
+            break;
+
+        case 1:
+            /* we have enough now to get the PDU bytes */
+            len = xrdp_process_get_pdu_bytes(s->p);
+            if (len == -1)
+            {
+                g_writeln("xrdp_process_data_in: "
+                          "xrdp_process_get_packet_bytes failed");
+                return 1;
+            }
+            pro->server_trans->header_size = len;
+            pro->server_trans->extra_flags = 2;
+            break;
+
+        case 2:
+            /* the whole PDU is read in now process */
+            s->p = s->data;
+            if (xrdp_process_loop(pro, s) != 0)
+            {
+                g_writeln("xrdp_process_data_in: "
+                          "xrdp_process_loop failed");
+                return 1;
+            }
+            init_stream(s, 0);
+            pro->server_trans->header_size = 4;
+            pro->server_trans->extra_flags = 1;
+            break;
     }
 
     return 0;
@@ -145,8 +244,12 @@ xrdp_process_main_loop(struct xrdp_process *self)
 
     DEBUG(("xrdp_process_main_loop"));
     self->status = 1;
+    self->server_trans->extra_flags = 0;
+    self->server_trans->header_size = 0;
+    self->server_trans->no_stream_init_on_data_in = 1;
     self->server_trans->trans_data_in = xrdp_process_data_in;
     self->server_trans->callback_data = self;
+    init_stream(self->server_trans->in_s, 8192 * 4);
     self->session = libxrdp_init((tbus)self, self->server_trans);
     /* this callback function is in xrdp_wm.c */
     self->session->callback = callback;