External Execution Interface

docs/Execution.md

@@ -79,7 +79,7 @@ Each of these threads handle both the datapath (i.e., UDP) and QUIC layers by de
 Using a single worker thread for both layers helps MsQuic can achieve lower latency and using separate threads for the two layers can help achieve higher throughput.
 MsQuic aligns its processing logic with the rest of the networking stack (including hardware RSS) to ensure that all processing stays on the same NUMA node, and ideally, the same processor.
 
-The complexity of aligning processing across various threads and processors is the primary reason for MsQuic to manage its own threading. 
+The complexity of aligning processing across various threads and processors is the primary reason for MsQuic to manage its own threading.
 This provides developers with a performant abstraction of both functionality and threading model, which simplifies application development using MsQuic, ensuring that things "just work" efficiently for QUIC by default.
 
 Each thread manages the execution of one or more connections.
@@ -116,6 +116,78 @@ graph TD
     end
 ```
 
+## Custom Execution
+
+MsQuic also supports scenarios where the application layer creates the threads that MsQuic uses to execute on.
+In this mode, the application creates one or more execution contexts that MsQuic will use to run all its internal logic.
+The application is responsible for calling down into MsQuic to allow these execution contexts to run.
+
+To create an execution context, the app much first create an event queue object, which is a platform specific type:
+
+- Windows: IOCP
+- Linux: epoll
+- macOS: kqueue
+
+On Windows, the following types are defined:
+
+```c++
+typedef HANDLE QUIC_EVENTQ;
+
+typedef OVERLAPPED_ENTRY CXPLAT_CQE;
+
+typedef
+_IRQL_requires_max_(PASSIVE_LEVEL)
+void
+(CXPLAT_EVENT_COMPLETION)(
+    _In_ CXPLAT_CQE* Cqe
+    );
+typedef CXPLAT_EVENT_COMPLETION *CXPLAT_EVENT_COMPLETION_HANDLER;
+
+typedef struct CXPLAT_SQE {
+    OVERLAPPED Overlapped;
+    CXPLAT_EVENT_COMPLETION_HANDLER Completion;
+} CXPLAT_SQE;
+```
+
+You will also notice the definiton for `QUIC_SQE` (SQE stands for submission queue entry), which defines the format that all completion events must take so they may be generically processed from the event queue (more on this below).
+
+Once the app has the event queue, it may create the execution context with the `ExecutionCreate` function:
+
+```c++
+HANDLE IOCP = CreateIoCompletionPort(INVALID_HANDLE_VALUE, nullptr, 0, 1);
+QUIC_EXECUTION_CONFIG ExecConfig = { 0, &IOCP };
+
+QUIC_EXECUTION* ExecContext = nullptr;
+QUIC_STATUS Status = MsQuic->ExecutionCreate(QUIC_GLOBAL_EXECUTION_CONFIG_FLAG_NONE, 0, 1, &ExecConfig, &ExecContext);
+```
+
+The above code createa a new IOCP (for Windows), sets up an execution config, indicating an ideal processor of 0 and the pointer to the IOCP, and then calls MsQuic to create 1 execution context.
+An application may expand this code to create multiple execution contexts, depending on their needs.
+
+To drive this execution context, the app will need to to periodically call `ExecutionPoll` and use the platform specific function to drain completion events from the event queue.
+
+```c
+bool AllDone = false;
+while (!AllDone) {
+    uint32_t WaitTime = MsQuic->ExecutionPoll(ExecContext);
+
+    ULONG OverlappedCount = 0;
+    OVERLAPPED_ENTRY Overlapped[8];
+    if (GetQueuedCompletionStatusEx(IOCP, Overlapped, ARRAYSIZE(Overlapped), &OverlappedCount, WaitTime, FALSE)) {
+        for (ULONG i = 0; i < OverlappedCount; ++i) {
+            QUIC_SQE* Sqe = CONTAINING_RECORD(Overlapped[i].lpOverlapped, QUIC_SQE, Overlapped);
+            Sqe->Completion(&Overlapped[i]);
+        }
+    }
+}
+```
+
+Above, you can see a simple loop that properly drives a single execution context on Windows.
+`OVERLAPPED_ENTRY` objects received from `GetQueuedCompletionStatusEx` are used to get the submission queue entry and then call its completion handler.
+
+In a real application, these completion events may come both from MsQuic and the application itself, therefore, this means **the application must use the same base format for its own submission entries**.
+This is necessary to be able to share the same event queue object.
+
 # See Also
 
 [QUIC_STREAM_CALLBACK](api/QUIC_STREAM_CALLBACK.md)<br>

src/core/api.c

@@ -157,7 +157,7 @@ MsQuicConnectionClose(
 
     CXPLAT_TEL_ASSERT(!Connection->State.HandleClosed);
 
-    if (IsWorkerThread) {
+    if (MsQuicLib.CustomExecutions || IsWorkerThread) {
         //
         // Execute this blocking API call inline if called on the worker thread.
         //
@@ -749,7 +749,7 @@ MsQuicStreamClose(
 
     CXPLAT_TEL_ASSERT(!Stream->Flags.HandleClosed);
 
-    if (IsWorkerThread) {
+    if (MsQuicLib.CustomExecutions || IsWorkerThread) {
         //
         // Execute this blocking API call inline if called on the worker thread.
         //
@@ -959,7 +959,8 @@ MsQuicStreamShutdown(
     QUIC_CONN_VERIFY(Connection, !Connection->State.Freed);
 
     if (Flags & QUIC_STREAM_SHUTDOWN_FLAG_INLINE &&
-        Connection->WorkerThreadID == CxPlatCurThreadID()) {
+        (MsQuicLib.CustomExecutions ||
+         Connection->WorkerThreadID == CxPlatCurThreadID())) {
 
         CXPLAT_PASSIVE_CODE();
 
@@ -1613,7 +1614,8 @@ MsQuicSetParam(
 
     QUIC_CONN_VERIFY(Connection, !Connection->State.Freed);
 
-    if (Connection->WorkerThreadID == CxPlatCurThreadID()) {
+    if (MsQuicLib.CustomExecutions ||
+        Connection->WorkerThreadID == CxPlatCurThreadID()) {
         //
         // Execute this blocking API call inline if called on the worker thread.
         //
@@ -1738,7 +1740,8 @@ MsQuicGetParam(
 
     QUIC_CONN_VERIFY(Connection, !Connection->State.Freed);
 
-    if (Connection->WorkerThreadID == CxPlatCurThreadID()) {
+    if (MsQuicLib.CustomExecutions ||
+        Connection->WorkerThreadID == CxPlatCurThreadID()) {
         //
         // Execute this blocking API call inline if called on the worker thread.
         //

src/core/api.h

@@ -5,6 +5,32 @@
 
 --*/
 
+_IRQL_requires_max_(PASSIVE_LEVEL)
+QUIC_STATUS
+QUIC_API
+MsQuicExecutionCreate(
+    _In_ QUIC_GLOBAL_EXECUTION_CONFIG_FLAGS Flags, // Used for datapath type
+    _In_ uint32_t PollingIdleTimeoutUs,
+    _In_ uint32_t Count,
+    _In_reads_(Count) QUIC_EXECUTION_CONFIG* Configs,
+    _Out_writes_(Count) QUIC_EXECUTION** Executions
+    );
+
+_IRQL_requires_max_(PASSIVE_LEVEL)
+void
+QUIC_API
+MsQuicExecutionDelete(
+    _In_ uint32_t Count,
+    _In_reads_(Count) QUIC_EXECUTION** Executions
+    );
+
+_IRQL_requires_max_(PASSIVE_LEVEL)
+uint32_t
+QUIC_API
+MsQuicExecutionPoll(
+    _In_ QUIC_EXECUTION* Execution
+    );
+
 _IRQL_requires_max_(PASSIVE_LEVEL)
 QUIC_STATUS
 QUIC_API

src/core/library.c

@@ -138,7 +138,15 @@
     CXPLAT_FRE_ASSERT(MsQuicLib.PartitionCount > 0);
 
     uint16_t* ProcessorList = NULL;
-    if (MsQuicLib.ExecutionConfig &&
+#ifndef _KERNEL_MODE
+    if (MsQuicLib.WorkerPool != NULL) {
+        MsQuicLib.CustomPartitions = TRUE;
+        MsQuicLib.PartitionCount = (uint16_t)CxPlatWorkerPoolGetCount(MsQuicLib.WorkerPool);
+    } else if (
+#else
+    if (
+#endif
+        MsQuicLib.ExecutionConfig &&
         MsQuicLib.ExecutionConfig->ProcessorCount &&
         MsQuicLib.ExecutionConfig->ProcessorCount != MsQuicLib.PartitionCount) {
         //
@@ -196,7 +204,14 @@
             QuicPartitionInitialize(
                 &MsQuicLib.Partitions[i],
                 i,
+#ifndef _KERNEL_MODE
+                ProcessorList ? ProcessorList[i] :
+                    (MsQuicLib.CustomPartitions ?
+                        (uint16_t)CxPlatWorkerPoolGetIdealProcessor(MsQuicLib.WorkerPool, i) :
+                        i),
+#else
                 ProcessorList ? ProcessorList[i] : i,
+#endif
                 CXPLAT_HASH_SHA256,
                 ResetHashKey,
                 sizeof(ResetHashKey));
@@ -681,6 +696,7 @@
     };
 
     QUIC_STATUS Status = QUIC_STATUS_SUCCESS;
+    BOOLEAN CreatedWorkerPool = FALSE;
 
     if (AcquireLock) {
         CxPlatLockAcquire(&MsQuicLib.Lock);
@@ -699,11 +715,14 @@
     }
 
 #ifndef _KERNEL_MODE
-    MsQuicLib.WorkerPool = CxPlatWorkerPoolCreate(MsQuicLib.ExecutionConfig);
-    if (!MsQuicLib.WorkerPool) {
-        Status = QUIC_STATUS_OUT_OF_MEMORY;
-        MsQuicLibraryFreePartitions();
-        goto Exit;
+    if (MsQuicLib.WorkerPool == NULL) {
+        MsQuicLib.WorkerPool = CxPlatWorkerPoolCreate(MsQuicLib.ExecutionConfig);
+        if (!MsQuicLib.WorkerPool) {
+            Status = QUIC_STATUS_OUT_OF_MEMORY;
+            MsQuicLibraryFreePartitions();
+            goto Exit;
+        }
+        CreatedWorkerPool = TRUE;
     }
 #endif
 
@@ -728,8 +747,10 @@
     } else {
         MsQuicLibraryFreePartitions();
 #ifndef _KERNEL_MODE
-        CxPlatWorkerPoolDelete(MsQuicLib.WorkerPool);
-        MsQuicLib.WorkerPool = NULL;
+        if (CreatedWorkerPool) {
+            CxPlatWorkerPoolDelete(MsQuicLib.WorkerPool);
+            MsQuicLib.WorkerPool = NULL;
+        }
 #endif
         goto Exit;
     }
@@ -1840,6 +1861,12 @@
 
     Api->DatagramSend = MsQuicDatagramSend;
 
+#ifndef _KERNEL_MODE
+    Api->ExecutionCreate = MsQuicExecutionCreate;
+    Api->ExecutionDelete = MsQuicExecutionDelete;
+    Api->ExecutionPoll = MsQuicExecutionPoll;
+#endif
+
     Api->ConnectionPoolCreate = MsQuicConnectionPoolCreate;
 
     *QuicApi = Api;
@@ -2428,3 +2455,104 @@
     }
     return Status;
 }
+
+#ifndef _KERNEL_MODE
+
+_IRQL_requires_max_(PASSIVE_LEVEL)
+QUIC_STATUS
+QUIC_API
+MsQuicExecutionCreate(
+    _In_ QUIC_GLOBAL_EXECUTION_CONFIG_FLAGS Flags, // Used for datapath type
+    _In_ uint32_t PollingIdleTimeoutUs,
+    _In_ uint32_t Count,
+    _In_reads_(Count) QUIC_EXECUTION_CONFIG* Configs,
+    _Out_writes_(Count) QUIC_EXECUTION** Executions
+    )
+{
+    QuicTraceEvent(
+        ApiEnter,
+        "[ api] Enter %u (%p).",
+        QUIC_TRACE_API_EXECUTION_CREATE,
+        NULL);
+
+    QUIC_STATUS Status = QUIC_STATUS_SUCCESS;
+
+    UNREFERENCED_PARAMETER(Flags);
+    UNREFERENCED_PARAMETER(PollingIdleTimeoutUs);
+
+    if (MsQuicLib.LazyInitComplete) {
+        //
+        // Not allowed to change the execution config after we've already
+        // started running the library.
+        //
+        Status = QUIC_STATUS_INVALID_STATE;
+
+    } else {
+        //
+        // Clean up any previous worker pool and create a new one.
+        //
+        CxPlatWorkerPoolDelete(MsQuicLib.WorkerPool);
+        MsQuicLib.WorkerPool =
+            CxPlatWorkerPoolCreateExternal(Count, Configs, Executions);
+        if (MsQuicLib.WorkerPool == NULL) {
+            Status = QUIC_STATUS_OUT_OF_MEMORY;
+        }
+
+        MsQuicLib.CustomExecutions = TRUE;
+    }
+
+    QuicTraceEvent(
+        ApiExitStatus,
+        "[ api] Exit %u",
+        Status);
+
+    return Status;
+}
+
+_IRQL_requires_max_(PASSIVE_LEVEL)
+void
+QUIC_API
+MsQuicExecutionDelete(
+    _In_ uint32_t Count,
+    _In_reads_(Count) QUIC_EXECUTION** Executions
+    )
+{
+    QuicTraceEvent(
+        ApiEnter,
+        "[ api] Enter %u (%p).",
+        QUIC_TRACE_API_EXECUTION_DELETE,
+        NULL);
+
+    UNREFERENCED_PARAMETER(Count);
+    UNREFERENCED_PARAMETER(Executions);
+    CxPlatWorkerPoolDelete(MsQuicLib.WorkerPool);
+    MsQuicLib.WorkerPool = NULL;
+
+    QuicTraceEvent(
+        ApiExit,
+        "[ api] Exit");
+}
+
+_IRQL_requires_max_(PASSIVE_LEVEL)
+uint32_t
+QUIC_API
+MsQuicExecutionPoll(
+    _In_ QUIC_EXECUTION* Execution
+    )
+{
+    QuicTraceEvent(
+        ApiEnter,
+        "[ api] Enter %u (%p).",
+        QUIC_TRACE_API_EXECUTION_POLL,
+        NULL);
+
+    uint32_t Result = CxPlatWorkerPoolWorkerPoll(Execution);
+
+    QuicTraceEvent(
+        ApiExit,
+        "[ api] Exit");
+
+    return Result;
+}
+
+#endif

src/core/library.h

@@ -55,6 +55,11 @@ typedef struct QUIC_LIBRARY {
     //
     BOOLEAN LazyInitComplete : 1;
 
+    //
+    // Indicates the app has configured their own execution contexts.
+    //
+    BOOLEAN CustomExecutions : 1;
+
     //
     // Indicates the app has configured non-default (per-processor) partitioning.
     //

src/core/operation.h

@@ -56,14 +56,14 @@ typedef enum QUIC_API_TYPE {
     QUIC_API_TYPE_STRM_SEND,
     QUIC_API_TYPE_STRM_RECV_COMPLETE,
     QUIC_API_TYPE_STRM_RECV_SET_ENABLED,
-    QUIC_API_TYPE_STRM_PROVIDE_RECV_BUFFERS,
 
     QUIC_API_TYPE_SET_PARAM,
     QUIC_API_TYPE_GET_PARAM,
 
     QUIC_API_TYPE_DATAGRAM_SEND,
     QUIC_API_TYPE_CONN_COMPLETE_RESUMPTION_TICKET_VALIDATION,
     QUIC_API_TYPE_CONN_COMPLETE_CERTIFICATE_VALIDATION,
+    QUIC_API_TYPE_STRM_PROVIDE_RECV_BUFFERS,
 
 } QUIC_API_TYPE;