Update Posix Platforms to Use Route Queue (#3290)

2022-12-16 12:13:35 -05:00 · 2022-12-16 12:13:35 -05:00 · 9b1df5b839
--- a/src/core/binding.c
+++ b/src/core/binding.c
@ -1624,6 +1624,8 @@ QuicBindingReceive(
    uint32_t TotalChainLength = 0;
    uint32_t TotalDatagramBytes = 0;

+    CXPLAT_DBG_ASSERT(Socket == Binding->Socket);
+
    //
    // Breaks the chain of datagrams into subchains by destination CID and
    // delivers the subchains.
--- a/src/core/connection.c
+++ b/src/core/connection.c
@ -6162,6 +6162,7 @@ QuicConnParamSet(
                Connection->Paths[0].Binding = OldBinding;
                break;
            }
+            Connection->Paths[0].Route.Queue = NULL;

            //
            // TODO - Need to free any queued recv packets from old binding.
--- a/src/generated/linux/datapath_epoll.c.clog.h
+++ b/src/generated/linux/datapath_epoll.c.clog.h
@ -237,13 +237,13 @@ tracepoint(CLOG_DATAPATH_EPOLL_C, DatapathDestroyed , arg2);\
 // QuicTraceEvent(
            DatapathSend,
            "[data][%p] Send %u bytes in %hhu buffers (segment=%hu) Dst=%!ADDR!, Src=%!ADDR!",
-            Socket,
+            SocketContext->Binding,
            SendData->TotalSize,
            SendData->BufferCount,
            SendData->SegmentSize,
            CASTED_CLOG_BYTEARRAY(sizeof(*RemoteAddress), RemoteAddress),
            CASTED_CLOG_BYTEARRAY(sizeof(*LocalAddress), LocalAddress));
-// arg2 = arg2 = Socket = arg2
+// arg2 = arg2 = SocketContext->Binding = arg2
 // arg3 = arg3 = SendData->TotalSize = arg3
 // arg4 = arg4 = SendData->BufferCount = arg4
 // arg5 = arg5 = SendData->SegmentSize = arg5
--- a/src/generated/linux/datapath_epoll.c.clog.h.lttng.h
+++ b/src/generated/linux/datapath_epoll.c.clog.h.lttng.h
@ -249,13 +249,13 @@ TRACEPOINT_EVENT(CLOG_DATAPATH_EPOLL_C, DatapathDestroyed,
 // QuicTraceEvent(
            DatapathSend,
            "[data][%p] Send %u bytes in %hhu buffers (segment=%hu) Dst=%!ADDR!, Src=%!ADDR!",
-            Socket,
+            SocketContext->Binding,
            SendData->TotalSize,
            SendData->BufferCount,
            SendData->SegmentSize,
            CASTED_CLOG_BYTEARRAY(sizeof(*RemoteAddress), RemoteAddress),
            CASTED_CLOG_BYTEARRAY(sizeof(*LocalAddress), LocalAddress));
-// arg2 = arg2 = Socket = arg2
+// arg2 = arg2 = SocketContext->Binding = arg2
 // arg3 = arg3 = SendData->TotalSize = arg3
 // arg4 = arg4 = SendData->BufferCount = arg4
 // arg5 = arg5 = SendData->SegmentSize = arg5
--- a/src/generated/linux/datapath_kqueue.c.clog.h
+++ b/src/generated/linux/datapath_kqueue.c.clog.h
@ -179,13 +179,13 @@ tracepoint(CLOG_DATAPATH_KQUEUE_C, DatapathDestroyed , arg2);\
 // QuicTraceEvent(
            DatapathSend,
            "[data][%p] Send %u bytes in %hhu buffers (segment=%hu) Dst=%!ADDR!, Src=%!ADDR!",
-            Socket,
+            SocketContext->Binding,
            SendData->TotalSize,
            SendData->BufferCount,
            SendData->SegmentSize,
            CASTED_CLOG_BYTEARRAY(sizeof(*RemoteAddress), RemoteAddress),
            CASTED_CLOG_BYTEARRAY(sizeof(*LocalAddress), LocalAddress));
-// arg2 = arg2 = Socket = arg2
+// arg2 = arg2 = SocketContext->Binding = arg2
 // arg3 = arg3 = SendData->TotalSize = arg3
 // arg4 = arg4 = SendData->BufferCount = arg4
 // arg5 = arg5 = SendData->SegmentSize = arg5
--- a/src/generated/linux/datapath_kqueue.c.clog.h.lttng.h
+++ b/src/generated/linux/datapath_kqueue.c.clog.h.lttng.h
@ -192,13 +192,13 @@ TRACEPOINT_EVENT(CLOG_DATAPATH_KQUEUE_C, DatapathDestroyed,
 // QuicTraceEvent(
            DatapathSend,
            "[data][%p] Send %u bytes in %hhu buffers (segment=%hu) Dst=%!ADDR!, Src=%!ADDR!",
-            Socket,
+            SocketContext->Binding,
            SendData->TotalSize,
            SendData->BufferCount,
            SendData->SegmentSize,
            CASTED_CLOG_BYTEARRAY(sizeof(*RemoteAddress), RemoteAddress),
            CASTED_CLOG_BYTEARRAY(sizeof(*LocalAddress), LocalAddress));
-// arg2 = arg2 = Socket = arg2
+// arg2 = arg2 = SocketContext->Binding = arg2
 // arg3 = arg3 = SendData->TotalSize = arg3
 // arg4 = arg4 = SendData->BufferCount = arg4
 // arg5 = arg5 = SendData->SegmentSize = arg5
--- a/src/inc/quic_datapath.h
+++ b/src/inc/quic_datapath.h
@ -153,12 +153,10 @@ typedef enum CXPLAT_ROUTE_STATE {
 //
 typedef struct CXPLAT_ROUTE {

-#ifdef _WIN32 // TODO - Use on Posix
    //
    // The (RSS) queue that this route is primarily associated with.
    //
    void* Queue;
-#endif // _WIN32

    QUIC_ADDR RemoteAddress;
    QUIC_ADDR LocalAddress;
--- a/src/platform/datapath_epoll.c
+++ b/src/platform/datapath_epoll.c
@ -441,7 +441,7 @@ CxPlatDataPathGetProc(

 QUIC_STATUS
 CxPlatSocketSendInternal(
-    _In_ CXPLAT_SOCKET* Socket,
+    _In_ CXPLAT_SOCKET_CONTEXT* SocketContext,
    _In_ const QUIC_ADDR* LocalAddress,
    _In_ const QUIC_ADDR* RemoteAddress,
    _In_ CXPLAT_SEND_DATA* SendData,
@ -918,6 +918,8 @@ CxPlatSocketContextInitialize(

    CXPLAT_SOCKET* Binding = SocketContext->Binding;

+    CXPLAT_DBG_ASSERT(SocketContext->Binding->Datapath == SocketContext->DatapathProc->Datapath);
+
    if (!CxPlatSqeInitialize(
            SocketContext->DatapathProc->EventQ,
            &SocketContext->ShutdownSqe.Sqe,
@ -1476,6 +1478,8 @@ CxPlatSocketContextRecvComplete(
    CXPLAT_RECV_DATA* DatagramHead = NULL;
    CXPLAT_RECV_DATA* DatagramTail = NULL;

+    CXPLAT_DBG_ASSERT(SocketContext->Binding->Datapath == SocketContext->DatapathProc->Datapath);
+
    for (int CurrentMessage = 0; CurrentMessage < MessagesReceived; CurrentMessage++) {
        CXPLAT_DATAPATH_RECV_BLOCK* CurrentBlock = SocketContext->CurrentRecvBlocks[CurrentMessage];
        SocketContext->CurrentRecvBlocks[CurrentMessage] = NULL;
@ -1501,6 +1505,7 @@ CxPlatSocketContextRecvComplete(
        }
        CxPlatConvertFromMappedV6(RemoteAddr, RemoteAddr);

+        RecvPacket->Route->Queue = SocketContext;
        RecvPacket->BufferLength = SocketContext->RecvMsgHdr[CurrentMessage].msg_len;
        BytesTransferred += RecvPacket->BufferLength;

@ -1672,7 +1677,7 @@ CxPlatSocketContextSendComplete(
    do {
        Status =
            CxPlatSocketSendInternal(
-                SocketContext->Binding,
+                SocketContext,
                SendData->Bind ? &SendData->LocalAddress : NULL,
                &SendData->RemoteAddress,
                SendData,
@ -2044,30 +2049,23 @@ CxPlatSendDataAlloc(
 {
    CXPLAT_DBG_ASSERT(Socket != NULL);

-    CXPLAT_DATAPATH_PROC* DatapathProc =
-        CxPlatDataPathGetProc(Socket->Datapath, CxPlatProcCurrentNumber());
-
-    CXPLAT_SEND_DATA* SendData =
-        CxPlatPoolAlloc(&DatapathProc->SendDataPool);
-
-    if (SendData == NULL) {
-        QuicTraceEvent(
-            AllocFailure,
-            "Allocation of '%s' failed. (%llu bytes)",
-            "CXPLAT_SEND_DATA",
-            0);
-        goto Exit;
+    if (Config->Route->Queue == NULL) {
+        Config->Route->Queue = &Socket->SocketContexts[0];
    }

-    CxPlatZeroMemory(SendData, sizeof(*SendData));
+    CXPLAT_SOCKET_CONTEXT* SocketContext = Config->Route->Queue;
+    CXPLAT_DBG_ASSERT(SocketContext->Binding == Socket);
+    CXPLAT_DBG_ASSERT(SocketContext->Binding->Datapath == SocketContext->DatapathProc->Datapath);
+    CXPLAT_SEND_DATA* SendData = CxPlatPoolAlloc(&SocketContext->DatapathProc->SendDataPool);
+    if (SendData != NULL) {
+        CxPlatZeroMemory(SendData, sizeof(*SendData));
+        SendData->Owner = SocketContext->DatapathProc;
+        SendData->ECN = Config->ECN;
+        SendData->SegmentSize =
+            (Socket->Datapath->Features & CXPLAT_DATAPATH_FEATURE_SEND_SEGMENTATION)
+                ? Config->MaxPacketSize : 0;
+    }

-    SendData->Owner = DatapathProc;
-    SendData->ECN = Config->ECN;
-    SendData->SegmentSize =
-        (Socket->Datapath->Features & CXPLAT_DATAPATH_FEATURE_SEND_SEGMENTATION)
-            ? Config->MaxPacketSize : 0;
-
-Exit:
    return SendData;
 }

@ -2338,7 +2336,7 @@ CxPlatSendDataComplete(

 QUIC_STATUS
 CxPlatSocketSendInternal(
-    _In_ CXPLAT_SOCKET* Socket,
+    _In_ CXPLAT_SOCKET_CONTEXT* SocketContext,
    _In_ const QUIC_ADDR* LocalAddress,
    _In_ const QUIC_ADDR* RemoteAddress,
    _In_ CXPLAT_SEND_DATA* SendData,
@ -2346,7 +2344,6 @@ CxPlatSocketSendInternal(
    )
 {
    QUIC_STATUS Status = QUIC_STATUS_SUCCESS;
-    CXPLAT_SOCKET_CONTEXT* SocketContext = NULL;
    QUIC_ADDR MappedRemoteAddress = {0};
    struct cmsghdr *CMsg = NULL;
    struct in_pktinfo *PktInfo = NULL;
@ -2354,7 +2351,7 @@ CxPlatSocketSendInternal(
    BOOLEAN SendPending = FALSE;
    size_t TotalMessagesCount;

-    CXPLAT_DBG_ASSERT(Socket != NULL && RemoteAddress != NULL && SendData != NULL);
+    CXPLAT_DBG_ASSERT(SocketContext != NULL && RemoteAddress != NULL && SendData != NULL);
    CXPLAT_DBG_ASSERT(SendData->SentMessagesCount < CXPLAT_MAX_BATCH_SEND);
    CXPLAT_DBG_ASSERT(IsPendedSend || SendData->SentMessagesCount == 0);

@ -2369,13 +2366,6 @@ CxPlatSocketSendInternal(
    #endif
        ] = {0};

-    if (Socket->HasFixedRemoteAddress) {
-        SocketContext = &Socket->SocketContexts[0];
-    } else {
-        uint32_t ProcNumber = CxPlatProcCurrentNumber() % Socket->Datapath->ProcCount;
-        SocketContext = &Socket->SocketContexts[ProcNumber];
-    }
-
    if (!IsPendedSend) {
        CxPlatSendDataFinalizeSendBuffer(SendData);
        for (size_t i = SendData->SentMessagesCount; i < SendData->BufferCount; ++i) {
@ -2385,7 +2375,7 @@ CxPlatSocketSendInternal(
        QuicTraceEvent(
            DatapathSend,
            "[data][%p] Send %u bytes in %hhu buffers (segment=%hu) Dst=%!ADDR!, Src=%!ADDR!",
-            Socket,
+            SocketContext->Binding,
            SendData->TotalSize,
            SendData->BufferCount,
            SendData->SegmentSize,
@ -2443,7 +2433,7 @@ CxPlatSocketSendInternal(
        CMsg->cmsg_len = CMSG_LEN(sizeof(int));
        *(int *)CMSG_DATA(CMsg) = SendData->ECN;

-        if (!Socket->Connected) {
+        if (!SocketContext->Binding->Connected) {
            Mhdr->msg_controllen += RemoteAddress->Ip.sa_family == QUIC_ADDRESS_FAMILY_INET
                ? CMSG_SPACE(sizeof(struct in_pktinfo))
                : CMSG_SPACE(sizeof(struct in6_pktinfo));
@ -2575,9 +2565,13 @@ CxPlatSocketSend(
    _In_ CXPLAT_SEND_DATA* SendData
    )
 {
+    UNREFERENCED_PARAMETER(Socket);
+    CXPLAT_DBG_ASSERT(Route->Queue);
+    CXPLAT_SOCKET_CONTEXT* SocketContext = Route->Queue;
+    CXPLAT_DBG_ASSERT(SocketContext->Binding->Datapath == SocketContext->DatapathProc->Datapath);
    QUIC_STATUS Status =
        CxPlatSocketSendInternal(
-            Socket,
+            SocketContext,
            &Route->LocalAddress,
            &Route->RemoteAddress,
            SendData,
--- a/src/platform/datapath_kqueue.c
+++ b/src/platform/datapath_kqueue.c
@ -407,7 +407,7 @@ CxPlatDataPathGetProc(

 QUIC_STATUS
 CxPlatSocketSendInternal(
-    _In_ CXPLAT_SOCKET* Socket,
+    _In_ CXPLAT_SOCKET_CONTEXT* SocketContext,
    _In_ const QUIC_ADDR* LocalAddress,
    _In_ const QUIC_ADDR* RemoteAddress,
    _In_ CXPLAT_SEND_DATA* SendData,
@ -1266,6 +1266,7 @@ CxPlatSocketContextRecvComplete(
        CxPlatConvertFromMappedV6(RemoteAddr, RemoteAddr);
    }

+    RecvPacket->Route->Queue = SocketContext;
    RecvPacket->TypeOfService = 0;

    struct cmsghdr *CMsg;
@ -1430,7 +1431,7 @@ CxPlatSocketContextSendComplete(
    do {
        Status =
            CxPlatSocketSendInternal(
-                SocketContext->Binding,
+                SocketContext,
                SendData->Bind ? &SendData->LocalAddress : NULL,
                &SendData->RemoteAddress,
                SendData,
@ -1763,30 +1764,21 @@ CxPlatSendDataAlloc(
 {
    CXPLAT_DBG_ASSERT(Socket != NULL);

-    CXPLAT_DATAPATH_PROC* DatapathProc =
-        CxPlatDataPathGetProc(Socket->Datapath, CxPlatProcCurrentNumber());
-
-    CXPLAT_SEND_DATA* SendData =
-        CxPlatPoolAlloc(&DatapathProc->SendDataPool);
-
-    if (SendData == NULL) {
-        QuicTraceEvent(
-            AllocFailure,
-            "Allocation of '%s' failed. (%llu bytes)",
-            "CXPLAT_SEND_DATA",
-            0);
-        goto Exit;
+    if (Config->Route->Queue == NULL) {
+        Config->Route->Queue = &Socket->SocketContexts[0];
    }

-    CxPlatZeroMemory(SendData, sizeof(*SendData));
+    CXPLAT_SOCKET_CONTEXT* SocketContext = Config->Route->Queue;
+    CXPLAT_SEND_DATA* SendData = CxPlatPoolAlloc(&SocketContext->DatapathProc->SendDataPool);
+    if (SendData != NULL) {
+        CxPlatZeroMemory(SendData, sizeof(*SendData));
+        SendData->Owner = SocketContext->DatapathProc;
+        SendData->ECN = Config->ECN;
+        SendData->SegmentSize =
+            (Socket->Datapath->Features & CXPLAT_DATAPATH_FEATURE_SEND_SEGMENTATION)
+                ? Config->MaxPacketSize : 0;
+    }

-    SendData->Owner = DatapathProc;
-    SendData->ECN = Config->ECN;
-    SendData->SegmentSize =
-        (Socket->Datapath->Features & CXPLAT_DATAPATH_FEATURE_SEND_SEGMENTATION)
-            ? Config->MaxPacketSize : 0;
-
-Exit:
    return SendData;
 }

@ -2057,7 +2049,7 @@ CxPlatSendDataComplete(

 QUIC_STATUS
 CxPlatSocketSendInternal(
-    _In_ CXPLAT_SOCKET* Socket,
+    _In_ CXPLAT_SOCKET_CONTEXT* SocketContext,
    _In_ const QUIC_ADDR* LocalAddress,
    _In_ const QUIC_ADDR* RemoteAddress,
    _In_ CXPLAT_SEND_DATA* SendData,
@ -2065,7 +2057,6 @@ CxPlatSocketSendInternal(
    )
 {
    QUIC_STATUS Status = QUIC_STATUS_SUCCESS;
-    CXPLAT_SOCKET_CONTEXT* SocketContext = NULL;
    ssize_t SentByteCount = 0;
    QUIC_ADDR MappedRemoteAddress = {0};
    struct cmsghdr *CMsg = NULL;
@ -2079,14 +2070,7 @@ CxPlatSocketSendInternal(

    char ControlBuffer[CMSG_SPACE(sizeof(struct in6_pktinfo)) + CMSG_SPACE(sizeof(int))] = {0};

-    CXPLAT_DBG_ASSERT(Socket != NULL && RemoteAddress != NULL && SendData != NULL);
-
-    if (Socket->HasFixedRemoteAddress) {
-        SocketContext = &Socket->SocketContexts[0];
-    } else {
-        uint32_t ProcNumber = CxPlatProcCurrentNumber() % Socket->Datapath->ProcCount;
-        SocketContext = &Socket->SocketContexts[ProcNumber];
-    }
+    CXPLAT_DBG_ASSERT(SocketContext != NULL && RemoteAddress != NULL && SendData != NULL);

    if (!IsPendedSend) {
        CxPlatSendDataFinalizeSendBuffer(SendData);
@ -2097,7 +2081,7 @@ CxPlatSocketSendInternal(
        QuicTraceEvent(
            DatapathSend,
            "[data][%p] Send %u bytes in %hhu buffers (segment=%hu) Dst=%!ADDR!, Src=%!ADDR!",
-            Socket,
+            SocketContext->Binding,
            SendData->TotalSize,
            SendData->BufferCount,
            SendData->SegmentSize,
@ -2147,7 +2131,7 @@ CxPlatSocketSendInternal(
    CMsg->cmsg_len = CMSG_LEN(sizeof(int));
    *(int *)CMSG_DATA(CMsg) = SendData->ECN;

-    if (!Socket->Connected) {
+    if (!SocketContext->Binding->Connected) {
        Mhdr.msg_name = &MappedRemoteAddress;
        Mhdr.msg_namelen = sizeof(MappedRemoteAddress);
        Mhdr.msg_controllen += CMSG_SPACE(sizeof(struct in6_pktinfo));
@ -2262,9 +2246,12 @@ CxPlatSocketSend(
    _In_ CXPLAT_SEND_DATA* SendData
    )
 {
+    UNREFERENCED_PARAMETER(Socket);
+    CXPLAT_DBG_ASSERT(Route->Queue);
+    CXPLAT_SOCKET_CONTEXT* SocketContext = Route->Queue;
    QUIC_STATUS Status =
        CxPlatSocketSendInternal(
-            Socket,
+            SocketContext,
            &Route->LocalAddress,
            &Route->RemoteAddress,
            SendData,