Changeset d9ce049 in mainline

Timestamp:

2011-10-04T20:40:05Z (13 years ago)

Author:

Jiri Svoboda <jiri@…>

Branches:

lfn, master, serial, ticket/834-toolchain-update, topic/msim-upgrade, topic/simplify-dev-export

Children:

8c7a054

Parents:

32105348

Message:

Implement RECEIVE user call.

Location:

uspace/srv/net/tl/tcp

Files:

• conn.c (modified) (6 diffs)
• iqueue.c (modified) (1 diff)
• state.c (modified) (2 diffs)
• tcp_type.h (modified) (2 diffs)
• test.c (modified) (2 diffs)
• tqueue.c (modified) (4 diffs)
• tqueue.h (modified) (1 diff)

uspace/srv/net/tl/tcp/conn.c

@@ -70 +70 @@
 
         /* Allocate receive buffer */
+        fibril_mutex_initialize(&conn->rcv_buf_lock);
+        fibril_condvar_initialize(&conn->rcv_buf_cv);
         conn->rcv_buf_size = RCV_BUF_SIZE;
         conn->rcv_buf_used = 0;
+
         conn->rcv_buf = calloc(1, conn->rcv_buf_size);
         if (conn->rcv_buf == NULL) {
@@ -306 +309 @@
         if ((seg->ctrl & CTL_ACK) != 0) {
                 conn->snd_una = seg->ack;
-                tcp_tqueue_remove_acked(conn);
+
+                /*
+                 * Prune acked segments from retransmission queue and
+                 * possibly transmit more data.
+                 */
+                tcp_tqueue_ack_received(conn);
         }
 
@@ -459 +467 @@
                 /* Update SND.UNA */
                 conn->snd_una = seg->ack;
-
-                /* Prune acked segments from retransmission queue */
-                tcp_tqueue_remove_acked(conn);
         }
 
@@ -468 +473 @@
                 conn->snd_wl1 = seg->seq;
                 conn->snd_wl2 = seg->ack;
-        }
+
+                log_msg(LVL_DEBUG, "Updating send window, SND.WND=%" PRIu32
+                    ", SND.WL1=%" PRIu32 ", SND.WL2=%" PRIu32,
+                    conn->snd_wnd, conn->snd_wl1, conn->snd_wl2);
+        }
+
+        /*
+         * Prune acked segments from retransmission queue and
+         * possibly transmit more data.
+         */
+        tcp_tqueue_ack_received(conn);
 
         return cp_continue;
@@ -659 +674 @@
         tcp_conn_trim_seg_to_wnd(conn, seg);
 
+        fibril_mutex_lock(&conn->rcv_buf_lock);
+
         /* Determine how many bytes to copy */
         text_size = tcp_segment_text_size(seg);
@@ -664 +681 @@
 
         /* Copy data to receive buffer */
-        tcp_segment_text_copy(seg, conn->rcv_buf, xfer_size);
+        tcp_segment_text_copy(seg, conn->rcv_buf + conn->rcv_buf_used,
+            xfer_size);
+        conn->rcv_buf_used += xfer_size;
+
+        /* Signal to the receive function that new data has arrived */
+        fibril_condvar_broadcast(&conn->rcv_buf_cv);
+        fibril_mutex_unlock(&conn->rcv_buf_lock);
 
         log_msg(LVL_DEBUG, "Received %zu bytes of data.", xfer_size);

uspace/srv/net/tl/tcp/iqueue.c

@@ -104 +104 @@
 
         while (!seq_no_segment_acceptable(iqueue->conn, iqe->seg)) {
-                log_msg(LVL_DEBUG, "Skipping unacceptable segment");
+                log_msg(LVL_DEBUG, "Skipping unacceptable segment (RCV.NXT=%"
+                    PRIu32 ", RCV.NXT+RCV.WND=%" PRIu32 ", SEG.SEQ=%" PRIu32
+                    ", SEG.LEN=%" PRIu32 ")", iqueue->conn->rcv_nxt,
+                    iqueue->conn->rcv_nxt + iqueue->conn->rcv_wnd,
+                    iqe->seg->seq, iqe->seg->len);
 
                 list_remove(&iqe->link);

uspace/srv/net/tl/tcp/state.c

@@ -35 +35 @@
  */
 
+#include <fibril_synch.h>
 #include <io/log.h>
 #include <macros.h>
@@ -107 +108 @@
     xflags_t *xflags)
 {
+        size_t xfer_size;
+
         log_msg(LVL_DEBUG, "tcp_uc_receive()");
+
+        fibril_mutex_lock(&conn->rcv_buf_lock);
+
+        /* Wait for data to become available */
+        while (conn->rcv_buf_used == 0) {
+                log_msg(LVL_DEBUG, "tcp_uc_receive() - wait for data");
+                fibril_condvar_wait(&conn->rcv_buf_cv, &conn->rcv_buf_lock);
+        }
+
+        /* Copy data from receive buffer to user buffer */
+        xfer_size = min(size, conn->rcv_buf_used);
+        memcpy(buf, conn->rcv_buf, xfer_size);
+        *rcvd = xfer_size;
+
+        /* Remove data from receive buffer */
+        memmove(conn->rcv_buf, conn->rcv_buf + xfer_size, conn->rcv_buf_used -
+            xfer_size);
+        conn->rcv_buf_used -= xfer_size;
+        conn->rcv_wnd += xfer_size;
+
+        fibril_mutex_unlock(&conn->rcv_buf_lock);
+
+        /* TODO */
+        *xflags = 0;
+
+        /* Send new size of receive window */
+        tcp_tqueue_ctrl_seg(conn, CTL_ACK);
+
+        log_msg(LVL_DEBUG, "tcp_uc_receive() - returning %zu bytes",
+            xfer_size);
 }
 

uspace/srv/net/tl/tcp/tcp_type.h

@@ -37 +37 @@
 
 #include <adt/list.h>
+#include <fibril_synch.h>
 #include <sys/types.h>
 
@@ -114 +115 @@
         size_t rcv_buf_size;
         size_t rcv_buf_used;
+        fibril_mutex_t rcv_buf_lock;
+        fibril_condvar_t rcv_buf_cv;
 
         /** Send buffer */

uspace/srv/net/tl/tcp/test.c

@@ -45 +45 @@
 #include "test.h"
 
+#define RCV_BUF_SIZE 64
+
 static void test_srv(void *arg)
 {
         tcp_conn_t *conn;
         tcp_sock_t sock;
+        char rcv_buf[RCV_BUF_SIZE + 1];
+        size_t rcvd;
+        xflags_t xflags;
 
         printf("test_srv()\n");
@@ -54 +59 @@
         sock.addr.ipv4 = 0x7f000001;
         tcp_uc_open(80, &sock, ap_passive, &conn);
+
+        while (true) {
+                printf("User receive...\n");
+                tcp_uc_receive(conn, rcv_buf, RCV_BUF_SIZE, &rcvd, &xflags);
+                rcv_buf[rcvd] = '\0';
+                printf("User received %zu bytes '%s'.\n", rcvd, rcv_buf);
+
+                async_usleep(1000*1000*2);
+        }
 }
 

uspace/srv/net/tl/tcp/tqueue.c

@@ -72 +72 @@
         seg->wnd = conn->rcv_wnd;
 
+        log_msg(LVL_DEBUG, "SEG.SEQ=%" PRIu32 ", SEG.WND=%" PRIu32,
+            seg->seq, seg->wnd);
+
         if ((seg->ctrl & CTL_ACK) != 0)
                 seg->ack = conn->rcv_nxt;
@@ -88 +91 @@
 void tcp_tqueue_new_data(tcp_conn_t *conn)
 {
+        size_t avail_wnd;
         size_t data_size;
         tcp_segment_t *seg;
@@ -93 +97 @@
         log_msg(LVL_DEBUG, "tcp_tqueue_new_data()");
 
-        data_size = min(conn->snd_buf_used, conn->snd_wnd);
+        /* Number of free sequence numbers in send window */
+        avail_wnd = (conn->snd_una + conn->snd_wnd) - conn->snd_nxt;
+
+        data_size = min(conn->snd_buf_used, avail_wnd);
         log_msg(LVL_DEBUG, "conn->snd_buf_used = %zu, SND.WND = %zu, "
             "data_size = %zu", conn->snd_buf_used, conn->snd_wnd, data_size);
@@ -116 +123 @@
 }
 
-/** Remove ACKed segments from retransmission queue.
+/** Remove ACKed segments from retransmission queue and possibly transmit
+ * more data.
  *
  * This should be called when SND.UNA is updated due to incoming ACK.
  */
-void tcp_tqueue_remove_acked(tcp_conn_t *conn)
+void tcp_tqueue_ack_received(tcp_conn_t *conn)
 {
         (void) conn;
+
+        tcp_tqueue_new_data(conn);
 }
 

uspace/srv/net/tl/tcp/tqueue.h

@@ -42 +42 @@
 extern void tcp_tqueue_seg(tcp_conn_t *, tcp_segment_t *);
 extern void tcp_tqueue_new_data(tcp_conn_t *);
-extern void tcp_tqueue_remove_acked(tcp_conn_t *);
+extern void tcp_tqueue_ack_received(tcp_conn_t *);
 extern void tcp_transmit_segment(tcp_sockpair_t *, tcp_segment_t *);
 extern void tcp_header_setup(tcp_conn_t *, tcp_segment_t *, tcp_header_t *);