[WIP] Linux Connection ID based RSS

src/core/binding.c

@@ -1054,7 +1054,8 @@ QuicBindingProcessStatelessOperation(
         RecvPacket->Route,
         SendData,
         SendDatagram->Length,
-        1);
+        1,
+        FALSE); // ?
     SendData = NULL;
 
 Exit:
@@ -1787,9 +1788,11 @@ QuicBindingSend(
     _In_ const CXPLAT_ROUTE* Route,
     _In_ CXPLAT_SEND_DATA* SendData,
     _In_ uint32_t BytesToSend,
-    _In_ uint32_t DatagramsToSend
+    _In_ uint32_t DatagramsToSend,
+    _In_ BOOLEAN Connected
     )
 {
+    CxPlatSetHandshakeDone(SendData, Connected);
 #if QUIC_TEST_DATAPATH_HOOKS_ENABLED
     QUIC_TEST_DATAPATH_HOOKS* Hooks = MsQuicLib.TestDatapathHooks;
     if (Hooks != NULL) {
@@ -1817,6 +1820,7 @@ QuicBindingSend(
 #if QUIC_TEST_DATAPATH_HOOKS_ENABLED
     }
 #endif
+    CxPlatSetHandshakeDone(SendData, FALSE);
 
     QuicPerfCounterAdd(QUIC_PERF_COUNTER_UDP_SEND, DatagramsToSend);
     QuicPerfCounterAdd(QUIC_PERF_COUNTER_UDP_SEND_BYTES, BytesToSend);

src/core/binding.h

@@ -462,7 +462,8 @@ QuicBindingSend(
     _In_ const CXPLAT_ROUTE* Route,
     _In_ CXPLAT_SEND_DATA* SendData,
     _In_ uint32_t BytesToSend,
-    _In_ uint32_t DatagramsToSend
+    _In_ uint32_t DatagramsToSend,
+    _In_ BOOLEAN Connected // hack to identify whether the connection handshake is done
     );
 
 //

src/core/connection.c

@@ -4553,6 +4553,19 @@
                     &Frame);
             if (QUIC_SUCCEEDED(Status)) {
                 AckEliciting = TRUE;
+                if (QuicConnIsServer(Connection) && Connection->State.Connected) {
+                    QUIC_CID_HASH_ENTRY* SourceCid =
+                        CXPLAT_CONTAINING_RECORD(
+                            Connection->SourceCids.Next,
+                            QUIC_CID_HASH_ENTRY,
+                            Link);
+                    CxPlatUpdateRSSRule(
+                        Connection->Paths[0].Binding->Socket,
+                        SourceCid->CID.Length,
+                        SourceCid->CID.Data
+                    );
+                    // CXPLAT_FREE(SourceCid, QUIC_POOL_CIDHASH);
+                }
             } else if (Status == QUIC_STATUS_OUT_OF_MEMORY) {
                 QuicPacketLogDrop(Connection, Packet, "Crypto frame process OOM");
                 return FALSE;

src/core/packet_builder.c

@@ -1062,12 +1062,14 @@ QuicPacketBuilderSendBatch(
         "Sending batch. %hu datagrams",
         (uint16_t)Builder->TotalCountDatagrams);
 
+    QUIC_CONNECTION *Connection = Builder->Connection;
     QuicBindingSend(
         Builder->Path->Binding,
         &Builder->Path->Route,
         Builder->SendData,
         Builder->TotalDatagramsLength,
-        Builder->TotalCountDatagrams);
+        Builder->TotalCountDatagrams,
+        Connection->State.Connected);
 
     Builder->PacketBatchSent = TRUE;
     Builder->SendData = NULL;

src/inc/quic_datapath.h

@@ -559,6 +559,14 @@ CxPlatSocketCreateUdp(
     _Out_ CXPLAT_SOCKET** Socket
     );
 
+_IRQL_requires_max_(PASSIVE_LEVEL)
+QUIC_STATUS
+CxPlatUpdateRSSRule(
+    _In_ CXPLAT_SOCKET* Socket,
+    _In_ uint8_t CIDLength,
+    _In_ uint8_t* CIDData
+    );
+
 //
 // Creates a TCP socket for the given (optional) local address and (required)
 // remote address. This function immediately registers for upcalls from the
@@ -722,6 +730,13 @@ CxPlatSendDataIsFull(
     _In_ CXPLAT_SEND_DATA* SendData
     );
 
+_IRQL_requires_max_(DISPATCH_LEVEL)
+void
+CxPlatSetHandshakeDone(
+    _In_ CXPLAT_SEND_DATA* SendData,
+    _In_ BOOLEAN HandshakeDone
+    );
+
 //
 // Sends the data over the socket.
 //

src/platform/datapath_epoll.c

@@ -2731,6 +2731,7 @@ CxPlatDataPathSocketProcessIoCompletion(
             if (SocketContext->Binding->Type == CXPLAT_SOCKET_TCP_LISTENER) {
                 CxPlatSocketContextAcceptCompletion(SocketContext, Cqe);
             } else {
+                fprintf(stderr, "[EPOLL] Partition: %d, Thread: %d\n", SocketContext->DatapathPartition->PartitionIndex, gettid());
                 CxPlatSocketReceive(SocketContext);
             }
         }

src/platform/datapath_raw.c

@@ -341,6 +341,16 @@ RawSendDataIsFull(
 
 #define TH_ACK 0x10
 
+_IRQL_requires_max_(DISPATCH_LEVEL)
+void
+RawSetHandshakeDone(
+    _In_ CXPLAT_SEND_DATA* SendData,
+    _In_ BOOLEAN HandshakeDone
+    )
+{
+    SendData->HandshakeDone = HandshakeDone;
+}
+
 _IRQL_requires_max_(DISPATCH_LEVEL)
 QUIC_STATUS
 RawSocketSend(
@@ -376,7 +386,8 @@ RawSocketSend(
         Interface->OffloadStatus.Transmit.TransportLayerXsum,
         Route->TcpState.SequenceNumber,
         Route->TcpState.AckNumber,
-        TH_ACK);
+        TH_ACK,
+        SendData->HandshakeDone);
     CxPlatDpRawTxEnqueue(SendData);
     return QUIC_STATUS_SUCCESS;
 }

src/platform/datapath_raw.h

@@ -381,7 +381,8 @@ CxPlatFramingWriteHeaders(
     _In_ BOOLEAN SkipTransportLayerXsum,
     _In_ uint32_t TcpSeqNum,
     _In_ uint32_t TcpAckNum,
-    _In_ uint8_t TcpFlags
+    _In_ uint8_t TcpFlags,
+    _In_ BOOLEAN FakeNatRebinding
     );
 
 

src/platform/datapath_raw_linux.c

@@ -118,6 +118,7 @@ RawSocketCreateUdp(
     NewSocket->CibirIdOffsetSrc = Config->CibirIdOffsetSrc;
     NewSocket->CibirIdOffsetDst = Config->CibirIdOffsetDst;
     NewSocket->AuxSocket = INVALID_SOCKET;
+    NewSocket->ServerOwned = !!(Config->Flags & CXPLAT_SOCKET_SERVER_OWNED);
     NewSocket->UseTcp = Raw->UseTcp;
     if (Config->CibirIdLength) {
         memcpy(NewSocket->CibirId, Config->CibirId, Config->CibirIdLength);

src/platform/datapath_raw_socket.c

@@ -31,6 +31,7 @@
 #pragma warning(disable:4116) // unnamed type definition in parentheses
 #pragma warning(disable:4100) // unreferenced formal parameter
 
+static int ClientPort;
 
 uint32_t
 CxPlatGetRawSocketSize(void) {
@@ -179,7 +180,15 @@ CxPlatDpRawParseUdp(
     Packet->Reserved = L4_TYPE_UDP;
 
     Packet->Route->RemoteAddress.Ipv4.sin_port = Udp->SourcePort;
-    Packet->Route->LocalAddress.Ipv4.sin_port = Udp->DestinationPort;
+    if (htons(55555) == Udp->SourcePort) {
+        fprintf(stderr, "Source port 55555 arrived\n");
+    }
+    if (htons(55555) == Udp->DestinationPort) {
+        fprintf(stderr, "Destination port 55555 arrived, rewrite to %d\n", ntohs(ClientPort));
+        Packet->Route->LocalAddress.Ipv4.sin_port = ClientPort;
+    } else {
+        Packet->Route->LocalAddress.Ipv4.sin_port = Udp->DestinationPort;
+    }
 
     Packet->Buffer = (uint8_t*)Udp->Data;
     Packet->BufferLength = QuicNetByteSwapShort(Udp->Length) - sizeof(UDP_HEADER);
@@ -561,7 +570,7 @@ CxPlatDpRawSocketAckFin(
         Interface->OffloadStatus.Transmit.TransportLayerXsum,
         ReceivedTcpHeader->AckNumber,
         CxPlatByteSwapUint32(CxPlatByteSwapUint32(ReceivedTcpHeader->SequenceNumber) + 1),
-        TH_FIN | TH_ACK);
+        TH_FIN | TH_ACK, FALSE);
     CxPlatDpRawTxEnqueue(SendData);
 }
 
@@ -602,7 +611,7 @@ CxPlatDpRawSocketAckSyn(
         Interface->OffloadStatus.Transmit.TransportLayerXsum,
         ReceivedTcpHeader->AckNumber,
         CxPlatByteSwapUint32(CxPlatByteSwapUint32(ReceivedTcpHeader->SequenceNumber) + 1),
-        TcpFlags);
+        TcpFlags, FALSE);
     CxPlatDpRawTxEnqueue(SendData);
 
     SendData = InterlockedFetchAndClearPointer((void*)&Socket->PausedTcpSend);
@@ -622,7 +631,7 @@ CxPlatDpRawSocketAckSyn(
             Interface->OffloadStatus.Transmit.TransportLayerXsum,
             CxPlatByteSwapUint32(CxPlatByteSwapUint32(ReceivedTcpHeader->AckNumber) + 1),
             CxPlatByteSwapUint32(CxPlatByteSwapUint32(ReceivedTcpHeader->SequenceNumber) + 1),
-            TH_ACK);
+            TH_ACK, FALSE);
         CxPlatDpRawTxEnqueue(SendData);
 
         SendData = CxPlatSendDataAlloc(CxPlatRawToSocket(Socket), &SendConfig);
@@ -645,7 +654,7 @@ CxPlatDpRawSocketAckSyn(
             Interface->OffloadStatus.Transmit.TransportLayerXsum,
             ReceivedTcpHeader->AckNumber,
             CxPlatByteSwapUint32(CxPlatByteSwapUint32(ReceivedTcpHeader->SequenceNumber) + 1),
-            TH_RST | TH_ACK);
+            TH_RST | TH_ACK, FALSE);
         Socket->CachedRstSend = SendData;
     }
 }
@@ -679,7 +688,7 @@ CxPlatDpRawSocketSyn(
         Socket, Route, &SendData->Buffer, SendData->ECN,
         Interface->OffloadStatus.Transmit.NetworkLayerXsum,
         Interface->OffloadStatus.Transmit.TransportLayerXsum,
-        Route->TcpState.SequenceNumber, 0, TH_SYN);
+        Route->TcpState.SequenceNumber, 0, TH_SYN, FALSE);
     CxPlatDpRawTxEnqueue(SendData);
 }
 
@@ -694,7 +703,8 @@ CxPlatFramingWriteHeaders(
     _In_ BOOLEAN SkipTransportLayerXsum,
     _In_ uint32_t TcpSeqNum,
     _In_ uint32_t TcpAckNum,
-    _In_ uint8_t TcpFlags
+    _In_ uint8_t TcpFlags,
+    _In_ BOOLEAN FakeNatRebinding
     )
 {
     uint8_t* Transport;
@@ -735,7 +745,34 @@ CxPlatFramingWriteHeaders(
         //
         UDP = (UDP_HEADER*)(Buffer->Buffer - sizeof(UDP_HEADER));
         UDP->DestinationPort = Route->RemoteAddress.Ipv4.sin_port;
-        UDP->SourcePort = Route->LocalAddress.Ipv4.sin_port;
+
+        char* EnvPath = getenv("MSQUIC_FAKE_NAT_REBINDING");
+        if (EnvPath && EnvPath[0] == '1')
+        {
+            static BOOLEAN toggleRebinding = FALSE;
+            static uint64_t waitUntil = 1000;
+            static uint64_t count = 0;
+            if (FakeNatRebinding && !Socket->ServerOwned) {
+                if (toggleRebinding) {
+                    fprintf(stderr, "FakeNatRebinding. set to 55555 (%d)\n", htons(55555));
+                    ClientPort = Route->LocalAddress.Ipv4.sin_port;
+                    UDP->SourcePort = htons(55555);
+                } else {
+                    fprintf(stderr, "No FakeNatRebinding. set to %d (%d)\n", ntohs(Route->LocalAddress.Ipv4.sin_port), Route->LocalAddress.Ipv4.sin_port);
+                    UDP->SourcePort = Route->LocalAddress.Ipv4.sin_port;
+                }
+                count++;
+                if (count > waitUntil && count % 2 == 0) {
+                    count = waitUntil;
+                    toggleRebinding = !toggleRebinding;
+                }
+            } else {
+                UDP->SourcePort = Route->LocalAddress.Ipv4.sin_port;
+            }
+        } else {
+            UDP->SourcePort = Route->LocalAddress.Ipv4.sin_port;
+        }
+
         UDP->Length = QuicNetByteSwapShort((uint16_t)Buffer->Length + sizeof(UDP_HEADER));
         UDP->Checksum = 0;
         Transport = (uint8_t*)UDP;