Added failure path in case of Event Enqueue fails

src/platform/datapath_epoll.c

@@ -1021,10 +1021,20 @@ CxPlatSocketContextUninitialize(
         //
         epoll_ctl(*SocketContext->DatapathPartition->EventQ, EPOLL_CTL_DEL, SocketContext->SocketFd, NULL);
 
-        CXPLAT_FRE_ASSERT(
-            CxPlatEventQEnqueue(
+        if (!CxPlatEventQEnqueue(
                 SocketContext->DatapathPartition->EventQ,
-                &SocketContext->ShutdownSqe));
+                &SocketContext->ShutdownSqe)) {
+            int Errno = errno;
+            QuicTraceEvent(
+                DatapathErrorStatus,
+                "[data][%p] ERROR, %u, %s.",
+                SocketContext->Binding,
+                Errno,
+                "CxPlatEventQEnqueue failed (Shutdown)");
+
+            // Queue can’t run the completion, so do it inline to finish teardown.
+            CxPlatSocketContextUninitializeEventComplete(&SocketContext->ShutdownSqe.Cqe);
+        }
     }
 }
 
@@ -2183,10 +2193,20 @@ SocketSend(
     CxPlatLockRelease(&SocketContext->TxQueueLock);
     if (SendPending) {
         if (FlushTxQueue) {
-            CXPLAT_FRE_ASSERT(
-                CxPlatEventQEnqueue(
+            if (!CxPlatEventQEnqueue(
                     SocketContext->DatapathPartition->EventQ,
-                    &SocketContext->FlushTxSqe));
+                    &SocketContext->FlushTxSqe)) {
+                int Errno = errno;
+                QuicTraceEvent(
+                    DatapathErrorStatus,
+                    "[data][%p] ERROR, %u, %s.",
+                    SocketContext->Binding,
+                    Errno,
+                    "CxPlatEventQEnqueue failed (FlushTx)");
+
+                // Run the completion inline to keep draining sends.
+                CxPlatSocketContextFlushTxEventComplete(&SocketContext->FlushTxSqe.Cqe);
+            }
         }
         return;
     }

src/platform/datapath_kqueue.c

@@ -1012,9 +1012,20 @@ CxPlatSocketContextUninitialize(
             &SocketContext->IoSqe,
             EVFILT_READ,
             EV_DELETE);
-        CxPlatEventQEnqueue(
-            SocketContext->DatapathPartition->EventQ,
-            &SocketContext->ShutdownSqe);
+        if (!CxPlatEventQEnqueue(
+                SocketContext->DatapathPartition->EventQ,
+                &SocketContext->ShutdownSqe)) {
+            int Errno = errno;
+            QuicTraceEvent(
+                DatapathErrorStatus,
+                "[data][%p] ERROR, %u, %s.",
+                SocketContext->Binding,
+                Errno,
+                "CxPlatEventQEnqueue failed (Shutdown)");
+
+            // Queue can’t run the completion, so run it inline.
+            CxPlatSocketContextUninitializeEventComplete(&SocketContext->ShutdownSqe.Cqe);
+        }
     }
 }
 

src/platform/datapath_raw_xdp_linux.c

@@ -1231,7 +1231,18 @@ CxPlatXdpExecute(
             XdpPartitionShutdown,
             "[ xdp][%p] XDP partition shutdown",
             Partition);
-        CxPlatEventQEnqueue(Partition->EventQ, &Partition->ShutdownSqe);
+        if (!CxPlatEventQEnqueue(Partition->EventQ, &Partition->ShutdownSqe)) {
+            QuicTraceEvent(
+                LibraryErrorStatus,
+                "[xdp] ERROR, %u, %s.",
+                errno,
+                "CxPlatEventQEnqueue failed (Shutdown)");
+            //
+            // The event queue can’t deliver the shutdown SQE, so run the completion
+            // inline to drop the reference.
+            //
+            CxPlatPartitionShutdownEventComplete(&Partition->ShutdownSqe.Cqe);
+        }
         return FALSE;
     }
 

src/platform/datapath_raw_xdp_win.c

@@ -1952,7 +1952,18 @@ CxPlatXdpExecute(
             Queue->TxXsk = NULL;
             Queue = Queue->Next;
         }
-        CxPlatEventQEnqueue(Partition->EventQ, &Partition->ShutdownSqe);
+        if (!CxPlatEventQEnqueue(Partition->EventQ, &Partition->ShutdownSqe)) {
+            QuicTraceEvent(
+                LibraryErrorStatus,
+                "[ xdp] ERROR, %u, %s.",
+                GetLastError(),
+                "CxPlatEventQEnqueue failed (Shutdown)");
+            //
+            // Manually drop the partition’s ref since the completion callback
+            // won’t fire without the SQE being queued.
+            //
+            CxPlatDpRawRelease((XDP_DATAPATH*)Partition->Xdp);
+        }
         return FALSE;
     }
 

src/platform/platform_worker.c

@@ -92,6 +92,7 @@ typedef struct QUIC_CACHEALIGN CXPLAT_WORKER {
     BOOLEAN StoppingThread : 1;
     BOOLEAN StoppedThread : 1;
     BOOLEAN DestroyedThread : 1;
+    BOOLEAN EventQClosed : 1;
 #if DEBUG // Debug flags - Must not be in the bitfield.
     BOOLEAN ThreadStarted;
     BOOLEAN ThreadFinished;
@@ -218,14 +219,47 @@ CxPlatWorkerPoolInitWorker(
     return TRUE;
 }
 
+void
+CxPlatWorkerCleanupEventQueue(
+    _In_ CXPLAT_WORKER* Worker
+    )
+{
+    if (!Worker->InitializedEventQ) {
+        return;
+    }
+
+    if (Worker->InitializedUpdatePollSqe) {
+        CxPlatSqeCleanup(&Worker->EventQ, &Worker->UpdatePollSqe);
+        Worker->InitializedUpdatePollSqe = FALSE;
+    }
+    if (Worker->InitializedWakeSqe) {
+        CxPlatSqeCleanup(&Worker->EventQ, &Worker->WakeSqe);
+        Worker->InitializedWakeSqe = FALSE;
+    }
+    if (Worker->InitializedShutdownSqe) {
+        CxPlatSqeCleanup(&Worker->EventQ, &Worker->ShutdownSqe);
+        Worker->InitializedShutdownSqe = FALSE;
+    }
+
+    Worker->EventQClosed = TRUE;
+    CxPlatEventQCleanup(&Worker->EventQ);
+    Worker->InitializedEventQ = FALSE;
+}
+
 void
 CxPlatWorkerPoolDestroyWorker(
     _In_ CXPLAT_WORKER* Worker
     )
 {
     if (Worker->InitializedThread) {
         Worker->StoppingThread = TRUE;
-        CxPlatEventQEnqueue(&Worker->EventQ, &Worker->ShutdownSqe);
+        if (!CxPlatEventQEnqueue(&Worker->EventQ, &Worker->ShutdownSqe)) {
+            QuicTraceEvent(
+                LibraryError,
+                "[ lib] ERROR, %s.",
+                "CxPlatEventQEnqueue(shutdown) Manually closing the event queue.");
+            CxPlatWorkerCleanupEventQueue(Worker);
+        }
         CxPlatThreadWait(&Worker->Thread);
         CxPlatThreadDelete(&Worker->Thread);
 #if DEBUG
@@ -236,18 +270,9 @@ CxPlatWorkerPoolDestroyWorker(
     } else {
         // TODO - Handle synchronized cleanup for external event queues?
     }
-    if (Worker->InitializedUpdatePollSqe) {
-        CxPlatSqeCleanup(&Worker->EventQ, &Worker->UpdatePollSqe);
-    }
-    if (Worker->InitializedWakeSqe) {
-        CxPlatSqeCleanup(&Worker->EventQ, &Worker->WakeSqe);
-    }
-    if (Worker->InitializedShutdownSqe) {
-        CxPlatSqeCleanup(&Worker->EventQ, &Worker->ShutdownSqe);
-    }
-    if (Worker->InitializedEventQ) {
-        CxPlatEventQCleanup(&Worker->EventQ);
-    }
+
+    CxPlatWorkerCleanupEventQueue(Worker);
+
     if (Worker->InitializedECLock) {
         CxPlatLockUninitialize(&Worker->ECLock);
     }
@@ -498,7 +523,19 @@ CxPlatWorkerPoolAddExecutionContext(
     CxPlatLockRelease(&Worker->ECLock);
 
     if (QueueEvent) {
-        CxPlatEventQEnqueue(&Worker->EventQ, &Worker->UpdatePollSqe);
+        if (!CxPlatEventQEnqueue(&Worker->EventQ, &Worker->UpdatePollSqe)) {
+            QuicTraceEvent(
+                LibraryErrorStatus,
+                "[ lib] ERROR, %u, %s.",
+                GetLastError(),
+                "CxPlatEventQEnqueue failed (updatepoll)");
+
+            //
+            // The event queue isn’t able to deliver the SQE, so execute the
+            // completion inline to move the pending contexts into the active list.
+            //
+            CxPlatUpdateExecutionContexts(Worker);
+        }
     }
 }
 
@@ -509,7 +546,19 @@ CxPlatWakeExecutionContext(
 {
     CXPLAT_WORKER* Worker = (CXPLAT_WORKER*)Context->CxPlatContext;
     if (!InterlockedFetchAndSetBoolean(&Worker->Running)) {
-        CxPlatEventQEnqueue(&Worker->EventQ, &Worker->WakeSqe);
+        if (!CxPlatEventQEnqueue(&Worker->EventQ, &Worker->WakeSqe)) {
+            QuicTraceEvent(
+                LibraryErrorStatus,
+                "[ lib] ERROR, %u, %s.",
+                GetLastError(),
+                "CxPlatEventQEnqueue failed (wake)");
+
+            //
+            // Failed to wake the worker, so clear the flag so a future attempt
+            // can retry once the caller observes the failure.
+            //
+            InterlockedFetchAndClearBoolean(&Worker->Running);
+        }
     }
 }
 
@@ -691,6 +740,27 @@ CxPlatProcessEvents(
             Cqes,
             ARRAYSIZE(Cqes),
             Worker->State.WaitTime);
+
+    if (CqeCount == 0) {
+        if (Worker->EventQClosed) {
+            Worker->StoppedThread = TRUE;
+            return;
+        }
+    }
+#if _WIN32
+        DWORD Err = GetLastError();
+        if (Err == ERROR_ABANDONED_WAIT_0 || Err == ERROR_INVALID_HANDLE) {
+            Worker->EventQClosed = TRUE;
+            Worker->StoppedThread = TRUE;
+            return;
+        }
+#elif defined(CX_PLATFORM_LINUX) || defined(CX_PLATFORM_DARWIN)
+        if (errno == EBADF || errno == EINVAL) {
+            Worker->EventQClosed = TRUE;
+            Worker->StoppedThread = TRUE;
+            return;
+        }
+#endif
     uint32_t CurrentCqeCount = CqeCount;
     CXPLAT_CQE* CurrentCqe = Cqes;
 
@@ -764,6 +834,10 @@ CXPLAT_THREAD_CALLBACK(CxPlatWorkerThread, Context)
 
         CxPlatProcessEvents(Worker);
 
+        if (Worker->EventQClosed && Worker->StoppedThread) {
+            break;
+        }
+
         if (Worker->State.NoWorkCount == 0) {
             Worker->State.LastWorkTime = Worker->State.TimeNow;
         } else if (Worker->State.NoWorkCount > CXPLAT_WORKER_IDLE_WORK_THRESHOLD_COUNT) {