Another fix to window scaling: support queueing more than 64 KByte in ooseq data

src/core/pbuf.c

@@ -952,6 +952,56 @@ pbuf_copy_partial(struct pbuf *buf, void *dataptr, u16_t len, u16_t offset)
   return copied_total;
 }
 
+#if LWIP_TCP && TCP_QUEUE_OOSEQ && LWIP_WND_SCALE
+/**
+ * This method modifies a 'pbuf chain', so that its total length is
+ * smaller than 64K. The remainder of the original pbuf chain is stored
+ * in *rest.
+ * This function never creates new pbufs, but splits an existing chain
+ * in two parts. The tot_len of the modified packet queue will likely be
+ * smaller than 64K.
+ * 'packet queues' are not supported by this function.
+ *
+ * @param p the pbuf queue to be splitted
+ * @param rest pointer to store the remainder (after the first 64K)
+ */
+void pbuf_split_64k(struct pbuf *p, struct pbuf **rest)
+{
+  *rest = NULL;
+  if ((p != NULL) && (p->next != NULL)) {
+    u16_t tot_len_front = p->len;
+    struct pbuf *i = p;
+    struct pbuf *r = p->next;
+
+    /* continue until the total length (summed up as u16_t) overflows */
+    while ((r != NULL) && ((u16_t)(tot_len_front + r->len) > tot_len_front)) {
+      tot_len_front += r->len;
+      i = r;
+      r = r->next;
+    }
+    /* i now points to last packet of the first segment. Set next
+       pointer to NULL */
+    i->next = NULL;
+
+    if (r != NULL) {
+      /* Update the tot_len field in the first part */
+      for (i = p; i != NULL; i = i->next) {
+        i->tot_len -= r->tot_len;
+        LWIP_ASSERT("tot_len/len mismatch in last pbuf",
+                    (i->next != NULL) || (i->tot_len == i->len));
+      }
+      if (p->flags & PBUF_FLAG_TCP_FIN) {
+        r->flags |= PBUF_FLAG_TCP_FIN;
+      }
+
+      /* tot_len field in rest does not need modifications */
+      /* reference counters do not need modifications */
+      *rest = r;
+    }
+  }
+}
+#endif /* LWIP_TCP && TCP_QUEUE_OOSEQ && LWIP_WND_SCALE */
+
 /**
  * Copy application supplied data into a pbuf.
  * This function can only be used to copy the equivalent of buf->tot_len data.

src/core/tcp.c

@@ -1124,37 +1124,58 @@ tcp_fasttmr_start:
 err_t
 tcp_process_refused_data(struct tcp_pcb *pcb)
 {
-  err_t err;
-  u8_t refused_flags = pcb->refused_data->flags;
-  /* set pcb->refused_data to NULL in case the callback frees it and then
-     closes the pcb */
-  struct pbuf *refused_data = pcb->refused_data;
-  pcb->refused_data = NULL;
-  /* Notify again application with data previously received. */
-  LWIP_DEBUGF(TCP_INPUT_DEBUG, ("tcp_input: notify kept packet\n"));
-  TCP_EVENT_RECV(pcb, refused_data, ERR_OK, err);
-  if (err == ERR_OK) {
-    /* did refused_data include a FIN? */
-    if (refused_flags & PBUF_FLAG_TCP_FIN) {
-      /* correct rcv_wnd as the application won't call tcp_recved()
-         for the FIN's seqno */
-      if (pcb->rcv_wnd != TCP_WND) {
-        pcb->rcv_wnd++;
+#if TCP_QUEUE_OOSEQ && LWIP_WND_SCALE
+  struct pbuf *rest;
+  while (pcb->refused_data != NULL)
+#endif /* TCP_QUEUE_OOSEQ && LWIP_WND_SCALE */
+  {
+    err_t err;
+    u8_t refused_flags = pcb->refused_data->flags;
+    /* set pcb->refused_data to NULL in case the callback frees it and then
+       closes the pcb */
+    struct pbuf *refused_data = pcb->refused_data;
+#if TCP_QUEUE_OOSEQ && LWIP_WND_SCALE
+    pbuf_split_64k(refused_data, &rest);
+    pcb->refused_data = rest;
+#else /* TCP_QUEUE_OOSEQ && LWIP_WND_SCALE */
+    pcb->refused_data = NULL;
+#endif /* TCP_QUEUE_OOSEQ && LWIP_WND_SCALE */
+    /* Notify again application with data previously received. */
+    LWIP_DEBUGF(TCP_INPUT_DEBUG, ("tcp_input: notify kept packet\n"));
+    TCP_EVENT_RECV(pcb, refused_data, ERR_OK, err);
+    if (err == ERR_OK) {
+      /* did refused_data include a FIN? */
+      if (refused_flags & PBUF_FLAG_TCP_FIN
+#if TCP_QUEUE_OOSEQ && LWIP_WND_SCALE
+          && (rest == NULL)
+#endif /* TCP_QUEUE_OOSEQ && LWIP_WND_SCALE */
+         ) {
+        /* correct rcv_wnd as the application won't call tcp_recved()
+           for the FIN's seqno */
+        if (pcb->rcv_wnd != TCP_WND) {
+          pcb->rcv_wnd++;
+        }
+        TCP_EVENT_CLOSED(pcb, err);
+        if (err == ERR_ABRT) {
+          return ERR_ABRT;
+        }
       }
-      TCP_EVENT_CLOSED(pcb, err);
-      if (err == ERR_ABRT) {
-        return ERR_ABRT;
+    } else if (err == ERR_ABRT) {
+      /* if err == ERR_ABRT, 'pcb' is already deallocated */
+      /* Drop incoming packets because pcb is "full" (only if the incoming
+         segment contains data). */
+      LWIP_DEBUGF(TCP_INPUT_DEBUG, ("tcp_input: drop incoming packets, because pcb is \"full\"\n"));
+      return ERR_ABRT;
+    } else {
+      /* data is still refused, pbuf is still valid (go on for ACK-only packets) */
+#if TCP_QUEUE_OOSEQ && LWIP_WND_SCALE
+      if (rest != NULL) {
+        pbuf_cat(refused_data, rest);
       }
+#endif /* TCP_QUEUE_OOSEQ && LWIP_WND_SCALE */
+      pcb->refused_data = refused_data;
+      return ERR_INPROGRESS;
     }
-  } else if (err == ERR_ABRT) {
-    /* if err == ERR_ABRT, 'pcb' is already deallocated */
-    /* Drop incoming packets because pcb is "full" (only if the incoming
-       segment contains data). */
-    LWIP_DEBUGF(TCP_INPUT_DEBUG, ("tcp_input: drop incoming packets, because pcb is \"full\"\n"));
-    return ERR_ABRT;
-  } else {
-    /* data is still refused, pbuf is still valid (go on for ACK-only packets) */
-    pcb->refused_data = refused_data;
   }
   return ERR_OK;
 }

src/core/tcp_in.c

@@ -364,12 +364,24 @@ tcp_input(struct pbuf *p, struct netif *inp)
           }
         }
 
+#if TCP_QUEUE_OOSEQ && LWIP_WND_SCALE
+        while (recv_data != NULL) {
+          struct pbuf *rest = NULL;
+          pbuf_split_64k(recv_data, &rest);
+#else /* TCP_QUEUE_OOSEQ && LWIP_WND_SCALE */
         if (recv_data != NULL) {
+#endif /* TCP_QUEUE_OOSEQ && LWIP_WND_SCALE */
+
           LWIP_ASSERT("pcb->refused_data == NULL", pcb->refused_data == NULL);
           if (pcb->flags & TF_RXCLOSED) {
             /* received data although already closed -> abort (send RST) to
                notify the remote host that not all data has been processed */
             pbuf_free(recv_data);
+#if TCP_QUEUE_OOSEQ && LWIP_WND_SCALE
+            if (rest != NULL) {
+              pbuf_free(rest);
+            }
+#endif /* TCP_QUEUE_OOSEQ && LWIP_WND_SCALE */
             tcp_abort(pcb);
             goto aborted;
           }
@@ -377,13 +389,29 @@ tcp_input(struct pbuf *p, struct netif *inp)
           /* Notify application that data has been received. */
           TCP_EVENT_RECV(pcb, recv_data, ERR_OK, err);
           if (err == ERR_ABRT) {
+#if TCP_QUEUE_OOSEQ && LWIP_WND_SCALE
+            if (rest != NULL) {
+              pbuf_free(rest);
+            }
+#endif /* TCP_QUEUE_OOSEQ && LWIP_WND_SCALE */
             goto aborted;
           }
 
           /* If the upper layer can't receive this data, store it */
           if (err != ERR_OK) {
+#if TCP_QUEUE_OOSEQ && LWIP_WND_SCALE
+            if (rest != NULL) {
+              pbuf_cat(recv_data, rest);
+            }
+#endif /* TCP_QUEUE_OOSEQ && LWIP_WND_SCALE */
             pcb->refused_data = recv_data;
             LWIP_DEBUGF(TCP_INPUT_DEBUG, ("tcp_input: keep incoming packet, because pcb is \"full\"\n"));
+#if TCP_QUEUE_OOSEQ && LWIP_WND_SCALE
+            break;
+          } else {
+            /* Upper layer received the data, go on with the rest if > 64K */
+            recv_data = rest;
+#endif /* TCP_QUEUE_OOSEQ && LWIP_WND_SCALE */
           }
         }
 
@@ -1382,6 +1410,9 @@ tcp_receive(struct tcp_pcb *pcb)
           if (cseg->p->tot_len > 0) {
             /* Chain this pbuf onto the pbuf that we will pass to
                the application. */
+            /* With window scaling, this can overflow recv_data->tot_len, but
+               that's not a problem since we explicitly fix that before passing
+               recv_data to the application. */
             if (recv_data) {
               pbuf_cat(recv_data, cseg->p);
             } else {

src/include/lwip/pbuf.h

@@ -172,6 +172,9 @@ struct pbuf *pbuf_coalesce(struct pbuf *p, pbuf_layer layer);
 err_t pbuf_fill_chksum(struct pbuf *p, u16_t start_offset, const void *dataptr,
                        u16_t len, u16_t *chksum);
 #endif /* LWIP_CHECKSUM_ON_COPY */
+#if LWIP_TCP && TCP_QUEUE_OOSEQ && LWIP_WND_SCALE
+void pbuf_split_64k(struct pbuf *p, struct pbuf **rest);
+#endif /* LWIP_TCP && TCP_QUEUE_OOSEQ && LWIP_WND_SCALE */
 
 u8_t pbuf_get_at(struct pbuf* p, u16_t offset);
 u16_t pbuf_memcmp(struct pbuf* p, u16_t offset, const void* s2, u16_t n);

test/unit/core/test_pbuf.c

@@ -9,6 +9,9 @@
 #if LWIP_DNS
 #error "This test needs DNS turned off (as it mallocs on init)"
 #endif
+#if !LWIP_TCP || !TCP_QUEUE_OOSEQ || !LWIP_WND_SCALE
+#error "This test needs TCP OOSEQ queueing and window scaling enabled"
+#endif
 
 /* Setups/teardown functions */
 
@@ -23,6 +26,16 @@ pbuf_teardown(void)
 }
 
 
+#define TESTBUFSIZE_1 65535
+#define TESTBUFSIZE_2 65530
+#define TESTBUFSIZE_3 50050
+static u8_t testbuf_1[TESTBUFSIZE_1];
+static u8_t testbuf_1a[TESTBUFSIZE_1];
+static u8_t testbuf_2[TESTBUFSIZE_2];
+static u8_t testbuf_2a[TESTBUFSIZE_2];
+static u8_t testbuf_3[TESTBUFSIZE_3];
+static u8_t testbuf_3a[TESTBUFSIZE_3];
+
 /* Test functions */
 
 /** Call pbuf_copy on a pbuf with zero length */
@@ -61,13 +74,79 @@ START_TEST(test_pbuf_copy_zero_pbuf)
 }
 END_TEST
 
+START_TEST(test_pbuf_split_64k_on_small_pbufs)
+{
+  struct pbuf *p, *rest=NULL;
+  LWIP_UNUSED_ARG(_i);
+
+  p = pbuf_alloc(PBUF_RAW, 1, PBUF_POOL);
+  pbuf_split_64k(p, &rest);
+  fail_unless(p->tot_len == 1);
+  pbuf_free(p);
+}
+END_TEST
+
+START_TEST(test_pbuf_queueing_bigger_than_64k)
+{
+  int i;
+  err_t err;
+  struct pbuf *p1, *p2, *p3, *rest2=NULL, *rest3=NULL;
+  LWIP_UNUSED_ARG(_i);
+
+  for(i = 0; i < TESTBUFSIZE_1; i++)
+    testbuf_1[i] = rand();
+  for(i = 0; i < TESTBUFSIZE_2; i++)
+    testbuf_2[i] = rand();
+  for(i = 0; i < TESTBUFSIZE_3; i++)
+    testbuf_3[i] = rand();
+
+  p1 = pbuf_alloc(PBUF_RAW, TESTBUFSIZE_1, PBUF_POOL);
+  fail_unless(p1 != NULL);
+  p2 = pbuf_alloc(PBUF_RAW, TESTBUFSIZE_2, PBUF_POOL);
+  fail_unless(p2 != NULL);
+  p3 = pbuf_alloc(PBUF_RAW, TESTBUFSIZE_3, PBUF_POOL);
+  fail_unless(p3 != NULL);
+  err = pbuf_take(p1, testbuf_1, TESTBUFSIZE_1);
+  fail_unless(err == ERR_OK);
+  err = pbuf_take(p2, testbuf_2, TESTBUFSIZE_2);
+  fail_unless(err == ERR_OK);
+  err = pbuf_take(p3, testbuf_3, TESTBUFSIZE_3);
+  fail_unless(err == ERR_OK);
+
+  pbuf_cat(p1, p2);
+  pbuf_cat(p1, p3);
+
+  pbuf_split_64k(p1, &rest2);
+  fail_unless(p1->tot_len == TESTBUFSIZE_1);
+  fail_unless(rest2->tot_len == (u16_t)((TESTBUFSIZE_2+TESTBUFSIZE_3) & 0xFFFF));
+  pbuf_split_64k(rest2, &rest3);
+  fail_unless(rest2->tot_len == TESTBUFSIZE_2);
+  fail_unless(rest3->tot_len == TESTBUFSIZE_3);
+
+  pbuf_copy_partial(p1, testbuf_1a, TESTBUFSIZE_1, 0);
+  pbuf_copy_partial(rest2, testbuf_2a, TESTBUFSIZE_2, 0);
+  pbuf_copy_partial(rest3, testbuf_3a, TESTBUFSIZE_3, 0);
+  for(i = 0; i < TESTBUFSIZE_1; i++)
+    fail_unless(testbuf_1[i] == testbuf_1a[i]);
+  for(i = 0; i < TESTBUFSIZE_2; i++)
+    fail_unless(testbuf_2[i] == testbuf_2a[i]);
+  for(i = 0; i < TESTBUFSIZE_3; i++)
+    fail_unless(testbuf_3[i] == testbuf_3a[i]);
+
+  pbuf_free(p1);
+  pbuf_free(rest2);
+  pbuf_free(rest3);
+}
+END_TEST
 
 /** Create the suite including all tests for this module */
 Suite *
 pbuf_suite(void)
 {
   TFun tests[] = {
-    test_pbuf_copy_zero_pbuf
+    test_pbuf_copy_zero_pbuf,
+    test_pbuf_split_64k_on_small_pbufs,
+    test_pbuf_queueing_bigger_than_64k
   };
   return create_suite("PBUF", tests, sizeof(tests)/sizeof(TFun), pbuf_setup, pbuf_teardown);
 }

test/unit/lwipopts.h

@@ -46,6 +46,9 @@
 #define MEMP_NUM_TCP_SEG                TCP_SND_QUEUELEN
 #define TCP_SND_BUF                     (12 * TCP_MSS)
 #define TCP_WND                         (10 * TCP_MSS)
+#define LWIP_WND_SCALE                  1
+#define TCP_RCV_SCALE                   0
+#define PBUF_POOL_SIZE                  400 // pbuf tests need ~200KByte
 
 /* Minimal changes to opt.h required for etharp unit tests: */
 #define ETHARP_SUPPORT_STATIC_ENTRIES   1