Add draft of FastTransport::ClientSession

Author: ongardie

ft/run2.sh

@@ -1,15 +1,29 @@
 #!/bin/bash
 set -e
-(cd ..; make ./obj.fasttransport/FastEcho)
+
+SERVER="python server.py"
+CLIENT="python client.py"
+while getopts "sc" OPT; do
+    case $OPT in
+        s)
+            SERVER="../obj.fasttransport/FastEcho"
+            ;;
+        c)
+            CLIENT="../obj.fasttransport/FastTelnet -x"
+            ;;
+    esac
+done
+
+(cd ..; make ./obj.fasttransport/Fast{Echo,Telnet})
 TIME=5
 CAP_FILE=/tmp/x.cap
 sudo rm -f $CAP_FILE
 sudo dumpcap -a duration:$TIME -i lo -w $CAP_FILE &
 sleep .1
-../obj.fasttransport/FastEcho &
+$SERVER &
 SERVER_PID=$!
 sleep .1
-python client.py &
+$CLIENT &
 CLIENT_PID=$!
 trap "kill $SERVER_PID; kill $CLIENT_PID" exit
 sleep $TIME

ft/transport.py

@@ -51,7 +51,7 @@
 
 # The fraction of packets that will be dropped on transmission.
 # This should be 0 for production!
-PACKET_LOSS = 0.05
+PACKET_LOSS = 0.00
 
 WINDOW_SIZE = 10
 REQ_ACK_AFTER = 5
@@ -936,7 +936,6 @@ class _ClientChannel(object):
         # beginSending() transitions from IDLE to SENDING.
         # processReceivedData() transitions from SENDING to RECEIVING.
         state = None
-        session = None
         rpcId = None
         currentRpc = None
         outboundMsg = None
@@ -1062,11 +1061,9 @@ def connect(self, service):
         header = Header()
         header.direction = Header.CLIENT_TO_SERVER
         header.clientSessionHint = self._id
-        header.serverSessionHint = self._serverSessionHint
-        header.sessionToken = self._token
-        header.rpcId = 0
         header.serverSessionHint = 0
         header.sessionToken = 0
+        header.rpcId = 0
         header.channelId = 0
         header.payloadType = Header.PT_SESSION_OPEN
         self._transport._sendOne(self.getAddress(), header, Buffer([]))
@@ -1089,16 +1086,14 @@ def getAddress(self):
         return self._service.address
 
     def processInboundPacket(self, payloadCM):
-        """Return whether the session is still valid."""
-
         self._lastActivityTime = gettime()
 
         header = Header.fromString(payloadCM.payload[:Header.LENGTH])
 
         if header.channelId >= self._numChannels:
             if header.payloadType == Header.PT_SESSION_OPEN:
                 self._processSessionOpenResponse(payloadCM)
-            return True
+            return
 
         channel = self._channels[header.channelId]
         if channel.rpcId == header.rpcId:
@@ -1117,12 +1112,9 @@ def processInboundPacket(self, payloadCM):
                 self._serverSessionHint = None
                 self._token = None
                 self.connect(self._service)
-                return False
         else:
-            if (0 < channel.rpcId - header.rpcId < 1024 and
-                header.payloadType == Header.PT_DATA and header.requestAck):
+            if header.payloadType == Header.PT_DATA and header.requestAck:
                 raise NotImplementedError("faked full ACK response")
-        return True
 
     def startRpc(self, rpc):
         """Queue an RPC for transmission on this session.
@@ -1140,15 +1132,6 @@ def startRpc(self, rpc):
             channel.currentRpc = rpc
             channel.outboundMsg.beginSending(rpc.getRequestBuffer())
 
-    def getActiveChannels(self):
-        """Used for timers."""
-        if not self._isConnected():
-            return
-        for channelId in range(self._numChannels):
-            channel = self._channels[channelId]
-            if channel.state != channel.IDLE_STATE:
-                yield channelId
-
     def close(self):
         debug("Aborting session")
         for channel in self._channels:
@@ -1312,9 +1295,7 @@ def _tryProcessPacket(self):
             else:
                 if header.clientSessionHint < len(self._clientSessions):
                     session = self._clientSessions[header.clientSessionHint]
-                    stillValid = session.processInboundPacket(payloadCM)
-                    if not stillValid:
-                        self._clientSessions.put(session)
+                    session.processInboundPacket(payloadCM)
         return True
 
     def poll(self):

src/Driver.cc

@@ -41,6 +41,15 @@ UDPDriver::release(char *payload, uint32_t len)
     packetBufsUtilized--;
 }
 
+UDPDriver::UDPDriver()
+    : socketFd(-1), packetBufsUtilized(0)
+{
+    int fd = socket(AF_INET, SOCK_DGRAM, 0);
+    if (fd == -1)
+        throw UnrecoverableDriverException(errno);
+    socketFd = fd;
+}
+
 UDPDriver::UDPDriver(const sockaddr *addr, socklen_t addrlen)
     : socketFd(-1), packetBufsUtilized(0)
 {

src/Driver.h

@@ -102,6 +102,7 @@ class UDPDriver : public Driver {
                             Buffer::Iterator *payload);
     virtual bool tryRecvPacket(Received *received);
     virtual void release(char *payload, uint32_t len);
+    UDPDriver();
     UDPDriver(const sockaddr *addr, socklen_t addrlen);
     virtual ~UDPDriver();
   private: