Changeset 2101

Timestamp:

06/16/10 15:11:16 (20 months ago)

Author:

mwhitworth

Message:

Add to TCP.

Location:

Whitix/branches/netchannel/user/sdk/network/tcp

Files:

• tcp.c (modified) (16 diffs)
• tcp.h (modified) (6 diffs)
• tcp_congestion.c (modified) (5 diffs)
• tcp_congestion.h (modified) (2 diffs)
• tcp_events.c (modified) (2 diffs)
• tcp_input.c (modified) (2 diffs)
• tcp_input.h (modified) (1 diff)
• tcp_output.c (modified) (5 diffs)
• tcp_states.c (modified) (3 diffs)
• tcp_timer.c (modified) (2 diffs)
• tcp_timer.h (modified) (1 diff)

Whitix/branches/netchannel/user/sdk/network/tcp/tcp.c

@@ -7 +7 @@
 #include <net/ipv4.h>
 #include <net/tcp.h>
+#include <limits.h>
 #include <errno.h>
 #include <net/aio.h>
@@ -67 +68 @@
         pthread_mutex_init(&tcpSock->sendUpdate, NULL);
 
+        /* Stats */
+        tcpSock->stats.minWin = ULONG_MAX;
+
         /* Add to list, used when writing out profiling and statistics data */
         if (!tcpHead)
@@ -100 +104 @@
 static void TcpRecvAio(struct TcpSocket* tcpSock, char* data, int length)
 {
+        Socket* socket = tcpSock->socket;
+        
+        while (socket->recvAio->flags & ASYNC_ACTIVE)
+                SysYield();
+
+        _SockEventFireList(socket, socket->recvAio, data, length);
 }
 
@@ -186 +196 @@
         ulong seq = NetToHostLong(header->seqNum);
         ulong ack = NetToHostLong(header->ackNum);
-        ulong dataLen = length - TcpGetHeaderSize(header) - sizeof(struct IpHeader);
+        struct IpHeader* ipHeader = (struct IpHeader*)ChanRecvBufferData(tcpSock->channel, buffer);
+        ulong dataLen = length - TcpGetHeaderSize(header) - IpGetHeaderSize(ipHeader);
         ulong seqEnd = seq + dataLen;
         int err = TCP_ERROR;
@@ -277 +288 @@
         if (dataLen > 0)
         {
-                ChanRecvBuffAddRead(tcpSock->channel, buffer, sizeof(struct IpHeader) +
-                        TcpGetHeaderSize(header));
+                ChanRecvBuffAddRead(tcpSock->channel, buffer, IpGetHeaderSize(ipHeader) + TcpGetHeaderSize(header));
 
                 tcpSock->stats.totalBytesRecv += dataLen;
@@ -305 +315 @@
                 tcpSock->ackMissed++;
 
-                if (tcpSock->ackMissedBytes >= 3*tcpSock->mss || tcpSock->ackMissed > 0)
+//              if (TcpInSlowStart(tcpSock) || tcpSock->ackMissedBytes >= 3*tcpSock->mss || tcpSock->ackMissed > 3)
                 {
                         tcpSock->ackMissed = 0;
@@ -318 +328 @@
                 
                 if (tcpSock->socket && tcpSock->socket->recvAio)
-                        TcpRecvAio(tcpSock, ChanRecvBufferData(tcpSock->channel, buffer), dataLen);
+                        TcpRecvAio(tcpSock, ChanRecvBufferData(tcpSock->channel, buffer) + IpGetHeaderSize(ipHeader) + TcpGetHeaderSize(header), dataLen);
         }
 
@@ -351 +361 @@
 
 dropPacket:
-        tcpSock->stats.droppedPackets++;
-        tcpSock->stats.droppedBytes += dataLen;
+        TcpDropPacket(tcpSock, dataLen);
         return err;
 }
@@ -695 +704 @@
 {
         struct TcpSocket* tcpSock = (struct TcpSocket*)(socket->priv);
+        struct HostEntry* entry;
+
+        if ((entry = HostFindEntry(TcpDestAddress(tcpSock), HOST_CLIENT)))
+        {
+                tcpSock->rto = entry->rto;
+                tcpSock->rtt = entry->rtt;
+                tcpSock->mdev = entry->mdev;
+        }else{
+                tcpSock->rto = TCP_RTO_DEFAULT;
+        tcpSock->rtt = TCP_RTT_DEFAULT;
+                tcpSock->localThresh = 0x10000;
+        }
 
         /* Window setup */
@@ -704 +725 @@
         tcpSock->localCwndBytes = tcpSock->mss;
         tcpSock->prevUpRatio = 3;
-        tcpSock->localThresh = 0x10000;
-
-        /* Retx */
-        tcpSock->rto = TCP_RTO_DEFAULT;
-
-        TCP_DEBUG(("mss (local) = %u", tcpSock->mss));
 
         if (TcpSendSyn(socket) < 0)
@@ -715 +730 @@
 
         TcpSetTimer(tcpSock, TCP_SYN_TIMEOUT, TcpResendSyn);
-
         TcpSocketAdd(socket);
 
@@ -728 +742 @@
 
         /* Set up statistics */
+        tcpSock->stats.destAddr = NetToHostLong(TcpDestAddress(tcpSock));
         tcpSock->stats.destPort = TcpDestPort(tcpSock);
 
@@ -738 +753 @@
         ChannelOptions options;
         struct TcpSocket* tcpSock = TcpInfo(socket);
+        struct Ipv4EndPoint* dest = (struct Ipv4EndPoint*)address;
 
         chanSrc.address = 0;
         chanSrc.port = 0; /* IP channel code will sort out a port. */
 
-        options.flags = CHAN_KEEP_SEND_BUFFERS;
+        if (!dest->port)
+                return -1;
+
+        options.flags = CHAN_KEEP_SEND_BUFFERS | CHAN_RECORD_SEND_TIME;
         options.sendBuffers = 0;
         options.recvBuffers = 0;
@@ -769 +788 @@
         chanDest.port = 0;
 
-        options.flags = CHAN_KEEP_SEND_BUFFERS;
+        options.flags = CHAN_KEEP_SEND_BUFFERS | CHAN_RECORD_SEND_TIME;
         options.sendBuffers = 0;
         options.recvBuffers = 0;
@@ -841 +860 @@
         srcIp.port = TcpSourcePort(tcpSock);
 
-        options.flags = CHAN_KEEP_SEND_BUFFERS;
+        options.flags = CHAN_KEEP_SEND_BUFFERS | CHAN_RECORD_SEND_TIME;
         options.sendBuffers = 0;
         options.recvBuffers = 0;
@@ -888 +907 @@
         address->port = destIp.port;
         address->address = NetToHostLong(destIp.address);
+
+        childInfo->stats.destAddr = address->address;
+        childInfo->stats.destPort = address->port;
 
         return 0;

Whitix/branches/netchannel/user/sdk/network/tcp/tcp.h

@@ -8 +8 @@
 #include <net/ipv4.h>
 #include <net/tcp.h>
+
+#include "../stats/stats.h"
 
 #define CHAN_TYPE_TCP           3
@@ -20 +22 @@
         ulong droppedPackets, droppedBytes;
         ulong retxBytesSent, retxPacketsSent;
-        ulong wrongSumPackets, wrongSumBytes;
+        ulong minWin, maxWin;
 
         /* (Mainly) protocol */
@@ -65 +67 @@
         struct UserTimer retxTimer;
         int retxTimerHnd;
-        int rto;
+        int rto, rtt, mdev;
 
         /* Sequence numbers. */
@@ -111 +113 @@
 
         /* Events */
-        struct AsyncCtx* remWinChanged; 
+        struct AsyncCtx *remWinChanged, *retxTimerOut, *localWinChanged, *droppedPkts; 
 };
+
+extern int _netDebug;
+
+#ifndef DEBUG_TCP
+#define DEBUG_TCP 0x1
+#endif
 
 //#define TCP_DEBUG_Y
 
 #ifdef TCP_DEBUG_Y
-#define TCP_DEBUG(args) do { printf("[TCP]: %d: %s: ", SysGetCurrentThreadId(), __func__); printf args; printf(" (%s:%d)\n", __FILE__, __LINE__); } while (0)
+#define TCP_DEBUG(args) do { if (_netDebug & DEBUG_TCP) { printf("[TCP]: %d: %s: ", SysGetCurrentThreadId(), __func__); printf args; printf(" (%s:%d)\n", __FILE__, __LINE__); } } while (0)
 #else
 #define TCP_DEBUG(args)
@@ -132 +140 @@
 {
         return ((struct Ipv4EndPoint*)(&socket->channel->src))->port;
+}
+
+static inline ulong TcpDestAddress(struct TcpSocket* socket)
+{
+        return ((struct Ipv4EndPoint*)(&socket->channel->dest))->address;
 }
 
@@ -156 +169 @@
 #define TCP_AGAIN       -2
 
+/* Buffer flags */
+#define TCP_FLAG_RETX   0x01
+
 #endif

Whitix/branches/netchannel/user/sdk/network/tcp/tcp_congestion.c

@@ +1 @@
+#include <net/socket.h>
+#include <net/inet_raw.h>
+
 #include "tcp.h"
 #include "seq.h"
@@ -4 +7 @@
 #include "../ipv4.h"
 #include "tcp_congestion.h"
+#include "tcp_timer.h"
 
 int TcpInSlowStart(struct TcpSocket* tcpSock)
@@ -13 +17 @@
 {
         ChanSendBuffer* curr;
+        struct IpHeader* ipHeader;
+        int ret;
 
         pthread_mutex_lock(&tcpSock->retxLock);
@@ -35 +41 @@
                 return 0;
 
-        TCP_DEBUG(("curr, fast retransmit!"));
-        SysExit(0);
+        ipHeader = (struct IpHeader*)(ChanSendBufferData(tcpSock->channel, curr));
+        TCP_DEBUG(("Resending length=%u", NetToHostShort(ipHeader->totalLength)));
+
+        ChanSendBuffSetFlags(curr, TCP_FLAG_RETX);
+
+        if ((ret = ChanBufferSend(tcpSock->channel, curr, NetToHostShort(ipHeader->totalLength))) < 0)
+        {
+                TCP_DEBUG(("Could not retransmit in retransmission timeout handler, ret=%d", ret));
+                tcpSock->retx = curr;
+        }
 
 out:
@@ -112 +126 @@
 void TcpRetxTimeout(void* data)
 {
-        printf("TcpRetxTimeout\n");
-}
+        struct TcpSocket* socket = (struct TcpSocket*)data;
+        ChanSendBuffer* retx;
+        struct IpHeader* ipHeader;
+        int ret;
+    ulong oldRto = socket->rto;
+
+        pthread_mutex_lock(&socket->retxLock);
+        if ((retx = socket->retx))
+                socket->retx = ChanSendBuffGetPriv(socket->retx);
+        pthread_mutex_unlock(&socket->retxLock);
+
+        if (retx)
+        {
+                ipHeader = (struct IpHeader*)(ChanSendBufferData(socket->channel, retx));
+                TCP_DEBUG(("Resending length=%u", NetToHostShort(ipHeader->totalLength)));
+
+                ChanSendBuffSetFlags(retx, TCP_FLAG_RETX);
+
+                if ((ret = ChanBufferSend(socket->channel, retx, NetToHostShort(ipHeader->totalLength))) < 0)
+                {
+                        TCP_DEBUG(("Could not retransmit in retransmission timeout handler, ret=%d", ret));
+                        socket->retx = retx;
+                }
+        }
+
+        socket->rto = MIN(socket->rto << 1, TCP_MSL);
+        TCP_DEBUG(("rto = %u", socket->rto));
+
+        TcpRetxTimerStart(socket);
+
+        if (socket->socket && socket->retxTimerOut)
+    {
+        /* Let application know retransmit timed out. Supply the timing information. */
+        struct TcpRetxInfo txInf;
+
+        txInf.rtoOld = oldRto;
+        txInf.rtoNew = socket->rto;
+        txInf.rtt = socket->rtt;
+        txInf.mdev = socket->mdev;
+
+                _SockEventFireList(socket->socket, socket->retxTimerOut, &txInf, sizeof(struct TcpRetxInfo));
+   }
+}
+
+void TcpUpdateRtt(struct TcpSocket* tcpSock, ChanSendBuffer* buffer)
+{
+        int diff = 0;
+        struct Time time, sendTime, result;
+
+        if (ChanSendBuffGetFlags(buffer) & TCP_FLAG_RETX)
+                return;
+
+        ChanSendBuffGetTime(buffer, &sendTime);
+        SysGetTime(&time);
+        TimerSub(&time, &sendTime, &result);
+        diff = result.seconds*1000;
+
+        if (diff <= 1000)
+                diff = 1000;
+
+        /* The new RTT estimate is 3/4*old + 1/4*new */
+        diff -= (tcpSock->rtt >> 2);
+        tcpSock->rtt += diff;
+
+        if (diff < 0)
+                diff = -diff;
+
+        diff -= (tcpSock->mdev >> 2);
+        tcpSock->mdev += diff;
+
+        /* Retransmission timeout */
+        tcpSock->rto = MIN(((tcpSock->rtt >> 2) + tcpSock->mdev) >> 1, TCP_MSL);
+
+        if (tcpSock->rto < 1000)
+                tcpSock->rto = 1000;
+
+        TCP_DEBUG(("rto = %u\n", tcpSock->rto));
+}

Whitix/branches/netchannel/user/sdk/network/tcp/tcp_congestion.h

@@ -2 +2 @@
 #define TCP_CONGESTION_H
 
+#include <net/channels.h>
 #include <types.h>
 
@@ -10 +11 @@
 void TcpCongestion(struct TcpSocket* tcpSock);
 void TcpRetxTimeout(void* data);
+void TcpUpdateRtt(struct TcpSocket* tcpSock, ChanRecvBuffer* buffer);
 
 #endif

Whitix/branches/netchannel/user/sdk/network/tcp/tcp_events.c

@@ -29 +29 @@
 int TcpRetxTimerOutRegister(Socket* socket, struct TcpEventCtx* ctx)
 {
-        /* 1. Verify socket is TCP */
-        /* 2. Use internal buffer when posting events. Have general TCP state structure */
+        struct TcpSocket* tcpSock = TcpInfo(socket);
+        ctx->ctx.socket = socket;
 
-        return 0;
+        return SocketAsyncAdd(&tcpSock->retxTimerOut, &ctx->ctx);
 }
 
-int TcpDroppedPktsRegister(Socket* socket, struct TcpEventCtx* ctx, int flags)
+int TcpDroppedPktsRegister(Socket* socket, struct TcpEventCtx* ctx)
 {
-        /* 1. Verify socket is TCP */
-        /* 2. flags could be CHECKSUM, DUP, etc. */
-        return 0;
+        struct TcpSocket* tcpSock = TcpInfo(socket);
+        ctx->ctx.socket = socket;
+
+        return SocketAsyncAdd(&tcpSock->droppedPkts, &ctx->ctx);
 }
 
 int TcpLocalWinRegister(Socket* socket, struct TcpEventCtx* ctx)
 {
-        return 0;
+        struct TcpSocket* tcpSock = TcpInfo(socket);
+        ctx->ctx.socket = socket;
+
+        return SocketAsyncAdd(&tcpSock->localWinChanged, &ctx->ctx);
 }
 
@@ -54 +58 @@
         return SocketAsyncAdd(&tcpSock->remWinChanged, &ctx->ctx);
 }
-
-int TcpStateUpdateRegister(Socket* socket, struct TcpEventCtx* ctx, int seconds)
-{
-        return 0;
-}
-
-int TcpSetTransfer(Socket* socket, int length, int blockSize)
-{
-        return 0;
-}

Whitix/branches/netchannel/user/sdk/network/tcp/tcp_input.c

@@ -24 +24 @@
 }
 
+void TcpDropPacket(struct TcpSocket* tcp, int length)
+{
+        tcp->stats.droppedPackets++;
+        tcp->stats.droppedBytes += length;
+
+        if (tcp->socket && tcp->droppedPkts)
+        {
+                _SockEventFireList(tcp->socket, tcp->droppedPkts, NULL, 0);
+        }
+}
+
 struct TcpHeader* TcpReadPacket(struct TcpSocket* tcpSock, ChanRecvBuffer** buffer, int*
                 length)
@@ -45 +56 @@
         if (tcpHeader->checkSum != TcpCalcChecksum(ipHeader, tcpHeader, *length))
         {
-                tcpSock->stats.wrongSumPackets++;
-                tcpSock->stats.wrongSumBytes += *length - sizeof(struct IpHeader);
+                TcpDropPacket(tcpSock, *length - sizeof(struct IpHeader));
                 goto error;
         }

Whitix/branches/netchannel/user/sdk/network/tcp/tcp_input.h

@@ -5 +5 @@
                 length);
 int TcpSocketRecv(Socket* socket, void* buffer, unsigned long length, int flags);
+void TcpDropPacket(struct TcpSocket* tcpSock, int length);
 
 #endif

Whitix/branches/netchannel/user/sdk/network/tcp/tcp_output.c

@@ -302 +302 @@
         {
                 TcpAddToTxQueue(tcpSock, buffer, length);
-        }else{
-//              if (TcpInSlowStart(tcpSock) || (length + TCP_HEADER_MAX_SIZE + sizeof (struct IpHeader)) >= tcpSock->mss)
-//              {
-                        ret = TcpTryToSend(tcpSock, buffer, length);
-//              }else{
-//                      printf("SMALL\n");
-//                      SysExit(0);
-//              }
-        }
+        }else
+                ret = TcpTryToSend(tcpSock, buffer, length);
 
         if (ret == TCP_AGAIN || ret >= 0)
@@ -349 +342 @@
                         tcpSock->bytesInFlight -= (endSeq-seq);
 
-//                      TCP_DEBUG(("%d, %d bytes", tcpSock->pktsInFlight, tcpSock->bytesInFlight));
-
-                        if (tcpSock->pktsInFlight == 0)
-                        {
-                                TCP_DEBUG(("queue = %#X (%d)\n", tcpSock->tx, tcpSock->stats.totalBytesSent));
-                        }
-
                         tcpSock->dupAckSeq = tcpSock->lastRetx = seq;
+
+                        /* Update round-trip time estimate */
+                        TcpUpdateRtt(tcpSock, buffer);
 
                         ChanSendBuffSetPriv(buffer, NULL);
@@ -384 +373 @@
 }
 
+
 int TcpSendPacket(struct TcpSocket* socket, struct TcpHeader* header, const void* buffer, unsigned long length, int flags, ChanSendBuffer** chanBuff)
 {
@@ -403 +393 @@
         if (buffer)
                 memcpy(data + sizeof(struct IpHeader) + tcpHeaderSize, buffer, length); 
-
         pthread_mutex_lock(&socket->sendUpdate);
 
@@ -424 +413 @@
         pthread_mutex_unlock(&socket->sendUpdate);
         return (ret < 0) ? ret : length;
+
 }
 

Whitix/branches/netchannel/user/sdk/network/tcp/tcp_states.c

@@ -34 +34 @@
         }
 
-        if (states[tcpSock->state] != NULL)
+        if (states[tcpSock->state] != NULL /*&& (TcpGetTs() & 1)*/)
                 ret = states[tcpSock->state](tcpSock, header, buffer, length);
+
+#if 0
+        else
+                printf("Dropped! (%u, %d)\n", length, tcpSock->state);
+#endif
 
         if (ret == TCP_ERROR)
@@ -61 +66 @@
                 ChanRecvBuffFree(tcpSock->channel, buffer);
 
-        /* Event handling */
+        /* Event handling and general statistical update */
         if (!ret && tcpSock->socket)
         {
@@ -70 +75 @@
         }
 
+        if (tcpSock->remoteWindow < tcpSock->stats.minWin)
+                tcpSock->stats.minWin = tcpSock->remoteWindow;
+
+        if (tcpSock->remoteWindow > tcpSock->stats.maxWin)
+                tcpSock->stats.maxWin = tcpSock->remoteWindow;
+
         return 0;
 }

Whitix/branches/netchannel/user/sdk/network/tcp/tcp_timer.c

@@ -37 +37 @@
 void TcpTimeWaitClose(struct TcpSocket* tcpSock)
 {
-        printf("TIME-WAIT: CLOSE\n");
-
         TcpChangeState(tcpSock, TCP_CLOSED);
 }
@@ -69 +67 @@
         spec.seconds = tcpSock->rto/1000;
         spec.useconds = tcpSock->rto % 1000;
-        
+
         SysTimerModify(tcpSock->retxTimerHnd, &spec, NULL);
 }

Whitix/branches/netchannel/user/sdk/network/tcp/tcp_timer.h

@@ -5 +5 @@
 #define TCP_MSL         2*60 /* minutes */
 #define TCP_RTO_DEFAULT         3*1000
+#define TCP_RTT_DEFAULT         1*1000
 
 int TcpTimerInit(struct TcpSocket* tcpSock);